From 026c00b8d226702ff87a70be8b32552c6a50da8f Mon Sep 17 00:00:00 2001 From: Dennis-yxchen Date: Sat, 4 Jul 2026 17:02:50 +0800 Subject: [PATCH] Add gated skill quality reporter --- .gitignore | 5 +- openspace/cloud/__init__.py | 12 +- openspace/cloud/config.py | 20 + openspace/cloud/redaction.py | 8 + openspace/cloud/skill_quality_reporter.py | 254 +++++++++++++ openspace/cloud/telemetry_payloads.py | 16 + openspace/skill_engine/analyzer.py | 90 +++++ .../test_analyzer_skill_quality_reporting.py | 334 +++++++++++++++++ tests/test_skill_quality_reporter.py | 345 ++++++++++++++++++ 9 files changed, 1082 insertions(+), 2 deletions(-) create mode 100644 openspace/cloud/skill_quality_reporter.py create mode 100644 tests/test_analyzer_skill_quality_reporting.py create mode 100644 tests/test_skill_quality_reporter.py diff --git a/.gitignore b/.gitignore index e6bd5dc..4170974 100644 --- a/.gitignore +++ b/.gitignore @@ -21,8 +21,11 @@ Desktop.ini # Development-only repository folders docs/ -tests/ scripts/ +tests/* +!tests/ +!tests/test_skill_quality_reporter.py +!tests/test_analyzer_skill_quality_reporting.py # Local agent/project memory OPENSPACE.md diff --git a/openspace/cloud/__init__.py b/openspace/cloud/__init__.py index 5808f9c..621485e 100644 --- a/openspace/cloud/__init__.py +++ b/openspace/cloud/__init__.py @@ -10,10 +10,15 @@ Provides: - ``generate_embedding`` — OpenAI embedding generation - ``CloudTelemetryOutbox`` — redacted telemetry retry queue - ``TaskTraceExporter`` — openspace_task_trace_v2 archive builder + - ``CloudSkillQualityReporter`` — analyzer skill quality telemetry reporter - telemetry payload helpers — schema-aligned telemetry builders """ -from openspace.cloud.config import CloudConfig, load_cloud_config +from openspace.cloud.config import ( + CloudConfig, + load_cloud_config, + load_cloud_skill_quality_reporting_enabled, +) def __getattr__(name: str): @@ -44,6 +49,9 @@ def __getattr__(name: str): if name == "CloudTaskTraceReporter": from openspace.cloud.task_trace_reporter import CloudTaskTraceReporter return CloudTaskTraceReporter + if name == "CloudSkillQualityReporter": + from openspace.cloud.skill_quality_reporter import CloudSkillQualityReporter + return CloudSkillQualityReporter if name in { "build_task_report_payload", "build_skill_use_report_payload", @@ -64,11 +72,13 @@ __all__ = [ "PackagePlacementResolver", "CloudConfig", "load_cloud_config", + "load_cloud_skill_quality_reporting_enabled", "SkillSearchEngine", "generate_embedding", "CloudTelemetryOutbox", "TaskTraceExporter", "CloudTaskTraceReporter", + "CloudSkillQualityReporter", "build_task_report_payload", "build_skill_use_report_payload", "build_evolve_report_payload", diff --git a/openspace/cloud/config.py b/openspace/cloud/config.py index e6c5507..5b76562 100644 --- a/openspace/cloud/config.py +++ b/openspace/cloud/config.py @@ -16,11 +16,13 @@ OPENSPACE_CLOUD_MODE_ENV = "OPENSPACE_CLOUD_MODE" OPENSPACE_CLOUD_BASE_URL_ENV = "OPENSPACE_CLOUD_BASE_URL" OPENSPACE_CLOUD_API_KEY_ENV = "OPENSPACE_CLOUD_API_KEY" OPENSPACE_CLOUD_TELEMETRY_MODE_ENV = "OPENSPACE_CLOUD_TELEMETRY_MODE" +OPENSPACE_CLOUD_SKILL_QUALITY_REPORTING_ENV = "OPENSPACE_CLOUD_SKILL_QUALITY_REPORTING" DEFAULT_CLOUD_BASE_URL = "https://open-space.cloud" _ALLOWED_CLOUD_MODES = {"off", "live"} _ALLOWED_TELEMETRY_MODES = {"off", "outbox"} +_TRUE_CONFIG_VALUES = {"1", "true", "yes", "on", "enabled"} class CloudConfigError(RuntimeError): @@ -64,6 +66,10 @@ def _normalize_telemetry_mode(value: str) -> TelemetryMode: return mode # type: ignore[return-value] +def _normalize_enabled_flag(value: str) -> bool: + return (value or "").strip().lower() in _TRUE_CONFIG_VALUES + + def normalize_cloud_base_url(value: str) -> str: base_url = (value or DEFAULT_CLOUD_BASE_URL).strip().rstrip("/") parsed = urlparse(base_url) @@ -104,6 +110,20 @@ def load_cloud_config() -> CloudConfig: ) +def load_cloud_skill_quality_reporting_enabled() -> bool: + """Return whether analyzer skill-quality telemetry may be emitted. + + This gate is separate from broader telemetry mode so live clients do not + send first-class quality fields until server support is explicitly enabled. + """ + + load_runtime_env() + host_env = read_host_mcp_env() + return _normalize_enabled_flag( + _get_cloud_env(OPENSPACE_CLOUD_SKILL_QUALITY_REPORTING_ENV, host_env) + ) + + def require_cloud_enabled(config: CloudConfig | None = None) -> CloudConfig: cfg = config or load_cloud_config() if not cfg.enabled: diff --git a/openspace/cloud/redaction.py b/openspace/cloud/redaction.py index 02c672c..618cf9d 100644 --- a/openspace/cloud/redaction.py +++ b/openspace/cloud/redaction.py @@ -77,6 +77,14 @@ _TELEMETRY_ALLOWED_KEYS = { "duration_ms", "error_code", "failure_reason", + "quality_event_kind", + "quality_schema_version", + "denominator", + "skill_applied", + "task_completed", + "skill_phase_failed", + "completed", + "fallback", "package_id", "package_path", "cloud_skill_id", diff --git a/openspace/cloud/skill_quality_reporter.py b/openspace/cloud/skill_quality_reporter.py new file mode 100644 index 0000000..12183e0 --- /dev/null +++ b/openspace/cloud/skill_quality_reporter.py @@ -0,0 +1,254 @@ +"""Best-effort analyzer skill quality telemetry reporter.""" + +from __future__ import annotations + +import asyncio +from pathlib import Path +from typing import Any + +from openspace.cloud.client import OpenSpaceClient +from openspace.cloud.config import ( + load_cloud_config, + load_cloud_skill_quality_reporting_enabled, +) +from openspace.cloud.local_mapping import CloudLocalMappingStore +from openspace.cloud.redaction import REDACTION_POLICY_VERSION +from openspace.cloud.telemetry_outbox import CloudTelemetryOutbox +from openspace.cloud.telemetry_payloads import ( + build_skill_use_report_payload, + short_cloud_request_id, +) +from openspace.config.constants import PROJECT_ROOT +from openspace.skill_engine.types import ExecutionAnalysis, SkillJudgment +from openspace.utils.logging import Logger + + +logger = Logger.get_logger(__name__) + +QUALITY_EVENT_KIND = "skill_judgment" +QUALITY_SCHEMA_VERSION = "skill_quality_v1" +QUALITY_DENOMINATOR = "analyzer_judged_skill_use" + + +class CloudSkillQualityReporter: + """Report persisted analyzer judgments as skill-use telemetry. + + The MVP fallback signal is trajectory-derived and conservative. In a + failed multi-skill trajectory, every reported skill is marked incomplete + until a structured per-skill outcome exists. + """ + + def __init__( + self, + *, + client: OpenSpaceClient | None = None, + mapping_store: CloudLocalMappingStore | None = None, + outbox: CloudTelemetryOutbox | None = None, + workspace_root: str | Path | None = None, + ) -> None: + self._client = client + self._mapping_store = mapping_store + self._outbox = outbox + self._workspace_root = Path(workspace_root).resolve() if workspace_root else PROJECT_ROOT + + async def maybe_report_analysis( + self, + analysis: ExecutionAnalysis, + *, + session_id: str | None = None, + ) -> dict[str, Any]: + try: + return await asyncio.to_thread( + self._maybe_report_analysis_sync, + analysis, + session_id=session_id, + ) + except Exception as exc: + logger.warning( + "Skill quality telemetry reporter failed for task %s: %s", + getattr(analysis, "task_id", ""), + exc, + ) + return { + "status": "skipped", + "reason": "reporter_error", + "error": type(exc).__name__, + } + + def _maybe_report_analysis_sync( + self, + analysis: ExecutionAnalysis, + *, + session_id: str | None = None, + ) -> dict[str, Any]: + cfg = load_cloud_config() + if not cfg.enabled or cfg.telemetry_mode != "outbox" or not cfg.api_key: + return {"status": "skipped", "reason": "cloud_telemetry_disabled"} + if not load_cloud_skill_quality_reporting_enabled(): + return {"status": "skipped", "reason": "skill_quality_reporting_disabled"} + + client = self._client or OpenSpaceClient(cfg, mapping_store=self._mapping_store) + mapping_store = self._mapping_store or client._local_mapping_store() + outbox = self._outbox or CloudTelemetryOutbox(mapping_store.db_path) + + outcomes: list[dict[str, Any]] = [] + for judgment in list(getattr(analysis, "skill_judgments", []) or []): + local_skill_id = str(getattr(judgment, "skill_id", "") or "").strip() + if not local_skill_id: + outcomes.append({"status": "skipped", "reason": "missing_local_skill_id"}) + continue + + try: + binding = mapping_store.get_binding_by_local(local_skill_id) + except Exception as exc: + outcomes.append( + { + "status": "failed", + "reason": "mapping_lookup_failed", + "local_skill_id": local_skill_id, + "error": type(exc).__name__, + } + ) + continue + cloud_skill_id = str(getattr(binding, "cloud_skill_id", "") or "").strip() + if not cloud_skill_id: + outcomes.append( + { + "status": "skipped", + "reason": "local_only", + "local_skill_id": local_skill_id, + } + ) + continue + + payload = build_skill_quality_judgment_payload( + analysis, + judgment, + cloud_skill_id=cloud_skill_id, + session_id=session_id, + ) + try: + ack = client.report_telemetry("skill-use-reported", payload) + outcomes.append( + { + "status": "reported", + "local_skill_id": local_skill_id, + "cloud_skill_id": cloud_skill_id, + "request_id": payload["request_id"], + "ack": ack, + } + ) + except Exception as exc: + try: + row = outbox.enqueue( + endpoint="/api/v2/telemetry/skill-use-reported", + payload=payload, + workspace_root=self._workspace_root, + ) + outbox.mark_failed( + row.request_id, + row.payload_hash, + error=str(exc), + ) + outcomes.append( + { + "status": "queued", + "reason": "report_failed", + "local_skill_id": local_skill_id, + "cloud_skill_id": cloud_skill_id, + "request_id": row.request_id, + "payload_hash": row.payload_hash, + "error": type(exc).__name__, + } + ) + except Exception as outbox_exc: + outcomes.append( + { + "status": "failed", + "reason": "outbox_enqueue_failed", + "local_skill_id": local_skill_id, + "cloud_skill_id": cloud_skill_id, + "request_id": payload["request_id"], + "error": type(outbox_exc).__name__, + } + ) + + return _summarize_outcomes(outcomes) + + +def build_skill_quality_judgment_payload( + analysis: ExecutionAnalysis, + judgment: SkillJudgment, + *, + cloud_skill_id: str, + session_id: str | None = None, +) -> dict[str, Any]: + local_skill_id = str(judgment.skill_id) + skill_phase_failed_ids = set(getattr(analysis, "skill_phase_failed_skill_ids", []) or []) + skill_applied = bool(judgment.skill_applied) + task_completed = bool(analysis.task_completed) + skill_phase_failed = local_skill_id in skill_phase_failed_ids + completed = skill_applied and task_completed and not skill_phase_failed + fallback = skill_phase_failed or not task_completed + status = "success" if completed and not fallback else "failed" + + return build_skill_use_report_payload( + request_id=short_cloud_request_id( + "skill-quality-judgment", + analysis.task_id, + local_skill_id, + cloud_skill_id, + ), + occurred_at=_analysis_timestamp_iso(analysis), + status=status, + task_id=analysis.task_id, + cloud_skill_id=cloud_skill_id, + session_id=session_id, + local_skill_id=local_skill_id, + duration_ms=None, + failure_reason=None if status == "success" else "unknown", + redaction_level="abstract_only", + redaction_performed_by="client", + redaction_policy_version=REDACTION_POLICY_VERSION, + quality_event_kind=QUALITY_EVENT_KIND, + quality_schema_version=QUALITY_SCHEMA_VERSION, + denominator=QUALITY_DENOMINATOR, + skill_applied=skill_applied, + task_completed=task_completed, + skill_phase_failed=skill_phase_failed, + completed=completed, + fallback=fallback, + ) + + +def _analysis_timestamp_iso(analysis: ExecutionAnalysis) -> str: + timestamp = getattr(analysis, "timestamp", "") + if hasattr(timestamp, "isoformat"): + return timestamp.isoformat() + return str(timestamp) + + +def _summarize_outcomes(outcomes: list[dict[str, Any]]) -> dict[str, Any]: + if any(item.get("status") == "reported" for item in outcomes): + status = "reported" + elif any(item.get("status") == "queued" for item in outcomes): + status = "queued" + elif any(item.get("status") == "failed" for item in outcomes): + status = "failed" + else: + status = "skipped" + + reason = None if outcomes else "no_skill_judgments" + if status == "skipped" and outcomes: + reason = "no_cloud_bound_skill_judgments" + result: dict[str, Any] = { + "status": status, + "reported_count": sum(1 for item in outcomes if item.get("status") == "reported"), + "queued_count": sum(1 for item in outcomes if item.get("status") == "queued"), + "skipped_count": sum(1 for item in outcomes if item.get("status") == "skipped"), + "failed_count": sum(1 for item in outcomes if item.get("status") == "failed"), + "outcomes": outcomes, + } + if reason: + result["reason"] = reason + return result diff --git a/openspace/cloud/telemetry_payloads.py b/openspace/cloud/telemetry_payloads.py index e416a4a..247c56f 100644 --- a/openspace/cloud/telemetry_payloads.py +++ b/openspace/cloud/telemetry_payloads.py @@ -90,6 +90,14 @@ def build_skill_use_report_payload( redaction_level: str = "abstract_only", redaction_performed_by: str = "client", redaction_policy_version: str | None = None, + quality_event_kind: str | None = None, + quality_schema_version: str | None = None, + denominator: str | None = None, + skill_applied: bool | None = None, + task_completed: bool | None = None, + skill_phase_failed: bool | None = None, + completed: bool | None = None, + fallback: bool | None = None, ) -> dict[str, Any]: """Build a payload accepted by POST /api/v2/telemetry/skill-use-reported.""" @@ -110,6 +118,14 @@ def build_skill_use_report_payload( _put_optional(payload, "failure_reason", failure_reason) _put_optional(payload, "error_code", error_code) _put_optional(payload, "redaction_policy_version", redaction_policy_version) + _put_optional(payload, "quality_event_kind", quality_event_kind) + _put_optional(payload, "quality_schema_version", quality_schema_version) + _put_optional(payload, "denominator", denominator) + _put_optional(payload, "skill_applied", skill_applied) + _put_optional(payload, "task_completed", task_completed) + _put_optional(payload, "skill_phase_failed", skill_phase_failed) + _put_optional(payload, "completed", completed) + _put_optional(payload, "fallback", fallback) if extras: payload["extras"] = dict(extras) return payload diff --git a/openspace/skill_engine/analyzer.py b/openspace/skill_engine/analyzer.py index 575fbb4..30a7256 100644 --- a/openspace/skill_engine/analyzer.py +++ b/openspace/skill_engine/analyzer.py @@ -45,6 +45,12 @@ if TYPE_CHECKING: logger = Logger.get_logger(__name__) +def _make_skill_quality_reporter() -> Any: + from openspace.cloud.skill_quality_reporter import CloudSkillQualityReporter + + return CloudSkillQualityReporter() + + # Maximum characters of conversation log to include in the analysis prompt. _MAX_CONVERSATION_CHARS = 80_000 @@ -360,6 +366,11 @@ class ExecutionAnalyzer: analysis, observed_tool_keys=context.get("used_tool_keys", set()), ) + await self._report_skill_quality_after_record_analysis( + analysis, + context, + execution_result=execution_result, + ) evo_types = [s.evolution_type.value for s in analysis.evolution_suggestions] logger.info( f"Execution analysis saved for task {task_id}: " @@ -406,6 +417,34 @@ class ExecutionAnalyzer: return False return True + async def _report_skill_quality_after_record_analysis( + self, + analysis: ExecutionAnalysis, + context: Dict[str, Any], + *, + execution_result: Dict[str, Any] | None = None, + ) -> None: + try: + reporter = _make_skill_quality_reporter() + outcome = await reporter.maybe_report_analysis( + analysis, + session_id=_analysis_session_id( + context, + execution_result=execution_result, + ), + ) + logger.debug( + "Skill quality telemetry outcome for task %s: %s", + analysis.task_id, + outcome.get("status") if isinstance(outcome, dict) else outcome, + ) + except Exception as exc: + logger.warning( + "Skill quality telemetry skipped after analyzer persistence for task %s: %s", + analysis.task_id, + exc, + ) + async def get_evolution_candidates( self, limit: int = 20 ) -> List[ExecutionAnalysis]: @@ -468,6 +507,7 @@ class ExecutionAnalyzer: analysis, observed_tool_keys=context.get("used_tool_keys", set()), ) + await self._report_skill_quality_after_record_analysis(analysis, context) evo_types = [s.evolution_type.value for s in analysis.evolution_suggestions] logger.info( "Execution packet analysis saved for task %s: completed=%s, " @@ -523,6 +563,7 @@ class ExecutionAnalyzer: "packet_tool_records": packet_tool_records, "execution_status": str(runtime_meta.get("status") or "unknown"), "iterations": int(runtime_meta.get("iterations") or 0), + "session_id": _packet_session_id(packet), "recording_dir": _packet_recording_dir(packet), "packet": packet, } @@ -1706,3 +1747,52 @@ def _packet_recording_dir(packet: "EvidencePacket") -> str: for ref in packet.selected_refs.get("recording_ref", []): return str(ref.uri or ref.metadata.get("recording_dir") or "") return "" + + +def _analysis_session_id( + context: Dict[str, Any], + *, + execution_result: Dict[str, Any] | None = None, +) -> str | None: + if execution_result is not None: + session_id = _clean_session_id(execution_result.get("session_id")) + if session_id: + return session_id + session_id = _clean_session_id(context.get("session_id")) + if session_id: + return session_id + packet = context.get("packet") + if packet is not None: + return _packet_session_id(packet) + return None + + +def _packet_session_id(packet: "EvidencePacket") -> str | None: + scope = getattr(packet, "scope", None) + session_id = _clean_session_id(getattr(scope, "session_id", None)) + if session_id: + return session_id + selected_refs = getattr(packet, "selected_refs", None) + if not isinstance(selected_refs, dict): + return None + refs = [ + ref + for ref_type in sorted(selected_refs) + for ref in selected_refs.get(ref_type, []) or [] + ] + for ref in refs: + session_id = _clean_session_id(getattr(ref, "session_id", None)) + if session_id: + return session_id + for ref in refs: + metadata = getattr(ref, "metadata", None) + if isinstance(metadata, dict): + session_id = _clean_session_id(metadata.get("session_id")) + if session_id: + return session_id + return None + + +def _clean_session_id(value: Any) -> str | None: + text = str(value or "").strip() + return text or None diff --git a/tests/test_analyzer_skill_quality_reporting.py b/tests/test_analyzer_skill_quality_reporting.py new file mode 100644 index 0000000..f9e073a --- /dev/null +++ b/tests/test_analyzer_skill_quality_reporting.py @@ -0,0 +1,334 @@ +import asyncio +import sys +import types +from datetime import datetime +from types import SimpleNamespace + +aiohttp_stub = types.ModuleType("aiohttp") +aiohttp_stub.ClientSession = type("ClientSession", (), {}) +aiohttp_stub.ClientResponse = type("ClientResponse", (), {}) +aiohttp_stub.ClientResponseError = type("ClientResponseError", (Exception,), {}) +aiohttp_stub.ClientTimeout = type( + "ClientTimeout", + (), + {"__init__": lambda self, *args, **kwargs: None}, +) +sys.modules.setdefault("aiohttp", aiohttp_stub) + +yarl_stub = types.ModuleType("yarl") + + +class URL(str): + def __truediv__(self, other): + return URL(self.rstrip("/") + "/" + str(other).lstrip("/")) + + +yarl_stub.URL = URL +sys.modules.setdefault("yarl", yarl_stub) + +import openspace.skill_engine.analyzer as analyzer_module +from openspace.skill_engine.analyzer import ExecutionAnalyzer +from openspace.skill_engine.evidence.types import ( + EvidencePacket, + EvidenceScope, + PacketBudget, + ResourceRef, +) +from openspace.skill_engine.types import ExecutionAnalysis, SkillJudgment + + +class FakeStore: + def __init__(self, *, fail_record=False): + self.fail_record = fail_record + self.events = [] + + def load_analyses_for_task(self, task_id): + return None + + async def record_analysis(self, analysis, observed_tool_keys=None): + self.events.append("record") + if self.fail_record: + raise RuntimeError("record failed") + + def close(self): + pass + + +class FakeReporter: + def __init__(self, events, *, fail=False): + self.events = events + self.fail = fail + + async def maybe_report_analysis(self, analysis, *, session_id=None): + self.events.append(("report", analysis.task_id, session_id)) + if self.fail: + raise RuntimeError("reporter failed") + return {"status": "reported"} + + +def _run(coro): + return asyncio.run(coro) + + +def _analysis(task_id="task-1"): + return ExecutionAnalysis( + task_id=task_id, + timestamp=datetime(2026, 1, 2, 3, 4, 5), + task_completed=True, + skill_judgments=[SkillJudgment("local-skill-1", True, "ignored")], + ) + + +async def _raw_json(*args, **kwargs): + return {"ok": True} + + +def _execution_analyzer(store, analysis): + analyzer = ExecutionAnalyzer(store=store, llm_client=object(), enabled=True) + analyzer._load_recording_context = lambda rec_path, execution_result: { + "selected_skills": ["local-skill-1"], + "skill_contents": {"local-skill-1": "content"}, + "used_tool_keys": {"shell:bash"}, + "traj_records": [{"tool": "bash"}], + "execution_status": "failed", + "iterations": 1, + "session_id": "context-session", + } + analyzer._build_analysis_prompt = lambda context: "prompt" + analyzer._run_analysis_loop = _raw_json + analyzer._parse_analysis = lambda task_id, raw_json, context: analysis + return analyzer + + +def test_analyze_execution_reports_after_record_analysis(monkeypatch, tmp_path): + store = FakeStore() + analysis = _analysis() + events = store.events + monkeypatch.setattr( + analyzer_module, + "_make_skill_quality_reporter", + lambda: FakeReporter(events), + ) + analyzer = _execution_analyzer(store, analysis) + result = _run( + analyzer.analyze_execution( + "task-1", + str(tmp_path), + {"status": "failed", "session_id": "execution-session"}, + ) + ) + assert result is analysis + assert events == ["record", ("report", "task-1", "execution-session")] + + +def test_analyze_execution_does_not_report_when_parse_returns_none( + monkeypatch, + tmp_path, +): + store = FakeStore() + events = store.events + monkeypatch.setattr( + analyzer_module, + "_make_skill_quality_reporter", + lambda: FakeReporter(events), + ) + analyzer = _execution_analyzer(store, _analysis()) + analyzer._parse_analysis = lambda task_id, raw_json, context: None + result = _run( + analyzer.analyze_execution( + "task-1", + str(tmp_path), + {"status": "failed", "session_id": "execution-session"}, + ) + ) + assert result is None + assert events == [] + + +def test_analyze_execution_does_not_report_when_record_analysis_raises( + monkeypatch, + tmp_path, +): + store = FakeStore(fail_record=True) + events = store.events + monkeypatch.setattr( + analyzer_module, + "_make_skill_quality_reporter", + lambda: FakeReporter(events), + ) + analyzer = _execution_analyzer(store, _analysis()) + result = _run( + analyzer.analyze_execution( + "task-1", + str(tmp_path), + {"status": "failed", "session_id": "execution-session"}, + ) + ) + assert result is None + assert events == ["record"] + + +def test_reporter_exception_is_non_fatal_after_persistence(monkeypatch, tmp_path): + store = FakeStore() + analysis = _analysis() + events = store.events + monkeypatch.setattr( + analyzer_module, + "_make_skill_quality_reporter", + lambda: FakeReporter(events, fail=True), + ) + analyzer = _execution_analyzer(store, analysis) + result = _run( + analyzer.analyze_execution( + "task-1", + str(tmp_path), + {"status": "failed", "session_id": "execution-session"}, + ) + ) + assert result is analysis + assert events == ["record", ("report", "task-1", "execution-session")] + + +def _packet(task_id="packet-task", *, session_id=None, selected_refs=None): + return SimpleNamespace( + packet_type="analysis", + packet_id="packet-1", + scope=SimpleNamespace(task_id=task_id, session_id=session_id), + selected_refs=selected_refs or {}, + ) + + +def test_load_packet_context_uses_packet_session_id_without_execution_result(): + packet = EvidencePacket( + packet_id="packet-ctx", + trigger_job_id="trigger-1", + packet_type="analysis", + profile_name="quality_signal", + subprofile="default", + manifest_watermark=1, + scope=EvidenceScope(task_id="packet-task", session_id="scope-session"), + selected_refs={ + "runtime_snapshot": [ + ResourceRef( + "runtime-1", + "runtime_snapshot", + metadata={ + "status": "failed", + "iterations": 2, + "active_skills": ["local-skill-1"], + "instruction_preview": "inspect packet context", + }, + ) + ], + "tool_result": [ + ResourceRef( + "tool-1", + "tool_result", + metadata={ + "tool_key": "shell:bash", + "status": "failed", + "result_preview": "boom", + }, + ) + ], + }, + expanded_snippets=[], + readable_paths=[], + instructions={}, + budget=PacketBudget(max_chars=1000, used_chars=0), + redaction_status="ok", + build_status="ok", + missing_ref_types=[], + ) + analyzer = ExecutionAnalyzer(store=FakeStore(), llm_client=object(), enabled=True) + + context = analyzer._load_packet_context(packet, task_id="packet-task") + + assert context["session_id"] == "scope-session" + assert context["task_id"] == "packet-task" + assert context["packet"] is packet + assert context["used_tool_keys"] == {"shell:bash"} + + +def _packet_analyzer(store, analysis, packet): + analyzer = ExecutionAnalyzer(store=store, llm_client=object(), enabled=True) + analyzer._load_packet_context = lambda packet, task_id: { + "selected_skills": ["local-skill-1"], + "skill_contents": {"local-skill-1": "content"}, + "used_tool_keys": {"shell:bash"}, + "packet_tool_records": [{"tool": "bash"}], + "traj_records": [{"tool": "bash"}], + "execution_status": "failed", + "iterations": 1, + "packet": packet, + } + analyzer._build_packet_analysis_prompt = lambda packet, context: "prompt" + analyzer._run_analysis_loop = _raw_json + analyzer._parse_analysis = lambda task_id, raw_json, context: analysis + return analyzer + + +def test_analyze_packet_reports_after_record_analysis(monkeypatch): + store = FakeStore() + packet = _packet(session_id="scope-session") + analysis = _analysis("packet-task") + events = store.events + monkeypatch.setattr( + analyzer_module, + "_make_skill_quality_reporter", + lambda: FakeReporter(events), + ) + analyzer = _packet_analyzer(store, analysis, packet) + result = _run(analyzer.analyze_packet(packet)) + assert result is analysis + assert events == ["record", ("report", "packet-task", "scope-session")] + + +def test_session_id_extraction_order_and_missing_values(): + assert ( + analyzer_module._analysis_session_id( + {"session_id": "context-session"}, + execution_result={"session_id": "execution-session"}, + ) + == "execution-session" + ) + assert ( + analyzer_module._analysis_session_id( + {"session_id": "context-session"}, + execution_result={}, + ) + == "context-session" + ) + + packet = SimpleNamespace( + scope=EvidenceScope(session_id="scope-session"), + selected_refs={}, + ) + assert analyzer_module._packet_session_id(packet) == "scope-session" + + packet = SimpleNamespace( + scope=EvidenceScope(), + selected_refs={ + "tool_result": [ + ResourceRef("ref-1", "tool_result", session_id="ref-session") + ] + }, + ) + assert analyzer_module._packet_session_id(packet) == "ref-session" + + packet = SimpleNamespace( + scope=EvidenceScope(), + selected_refs={ + "tool_result": [ + ResourceRef( + "ref-1", + "tool_result", + metadata={"session_id": "metadata-session"}, + ) + ] + }, + ) + assert analyzer_module._packet_session_id(packet) == "metadata-session" + + packet = SimpleNamespace(scope=EvidenceScope(), selected_refs={}) + assert analyzer_module._packet_session_id(packet) is None diff --git a/tests/test_skill_quality_reporter.py b/tests/test_skill_quality_reporter.py new file mode 100644 index 0000000..a9cc901 --- /dev/null +++ b/tests/test_skill_quality_reporter.py @@ -0,0 +1,345 @@ +import asyncio +import json +from datetime import datetime + +from openspace.cloud.config import ( + OPENSPACE_CLOUD_API_KEY_ENV, + OPENSPACE_CLOUD_MODE_ENV, + OPENSPACE_CLOUD_SKILL_QUALITY_REPORTING_ENV, + OPENSPACE_CLOUD_TELEMETRY_MODE_ENV, +) +from openspace.cloud.local_mapping import SkillCloudBinding +from openspace.cloud.redaction import redact_telemetry_payload +from openspace.cloud.skill_quality_reporter import ( + QUALITY_DENOMINATOR, + QUALITY_EVENT_KIND, + QUALITY_SCHEMA_VERSION, + CloudSkillQualityReporter, + build_skill_quality_judgment_payload, +) +from openspace.cloud.telemetry_outbox import CloudTelemetryOutbox +from openspace.cloud.telemetry_payloads import short_cloud_request_id +from openspace.skill_engine.types import ExecutionAnalysis, SkillJudgment + + +def _run(coro): + return asyncio.run(coro) + + +def _patch_host_env(monkeypatch): + import openspace.cloud.config as cloud_config + + monkeypatch.setattr(cloud_config, "load_runtime_env", lambda: None) + monkeypatch.setattr(cloud_config, "read_host_mcp_env", lambda: {}) + + +def _set_cloud_env( + monkeypatch, + *, + mode="live", + telemetry_mode="outbox", + api_key="test-key", + quality=True, +): + _patch_host_env(monkeypatch) + for key in ( + OPENSPACE_CLOUD_MODE_ENV, + OPENSPACE_CLOUD_TELEMETRY_MODE_ENV, + OPENSPACE_CLOUD_API_KEY_ENV, + OPENSPACE_CLOUD_SKILL_QUALITY_REPORTING_ENV, + ): + monkeypatch.delenv(key, raising=False) + monkeypatch.setenv(OPENSPACE_CLOUD_MODE_ENV, mode) + monkeypatch.setenv(OPENSPACE_CLOUD_TELEMETRY_MODE_ENV, telemetry_mode) + if api_key is not None: + monkeypatch.setenv(OPENSPACE_CLOUD_API_KEY_ENV, api_key) + if quality: + monkeypatch.setenv(OPENSPACE_CLOUD_SKILL_QUALITY_REPORTING_ENV, "1") + + +class FakeClient: + def __init__(self, *, fail=False): + self.fail = fail + self.calls = [] + + def report_telemetry(self, event, payload): + self.calls.append((event, payload)) + if self.fail: + raise RuntimeError("server does not support quality fields yet") + return {"ok": True} + + +class FakeMappingStore: + def __init__(self, tmp_path, bindings): + self.db_path = tmp_path / "mapping.db" + self.bindings = bindings + self.lookups = [] + + def get_binding_by_local(self, local_skill_id): + self.lookups.append(local_skill_id) + return self.bindings.get(local_skill_id) + + +def _analysis( + *, + task_id="task-1", + timestamp=None, + task_completed=True, + judgments=None, + phase_failed_ids=None, +): + return ExecutionAnalysis( + task_id=task_id, + timestamp=timestamp or datetime(2026, 1, 2, 3, 4, 5, 123456), + task_completed=task_completed, + execution_note=( + "RAW_EXECUTION_NOTE prompt transcript /tmp/private/file.diff token hash " + "redacted_preview" + ), + tool_issues=["RAW_TOOL_ISSUE traceback /home/user/project/file.py"], + skill_judgments=judgments + or [ + SkillJudgment( + skill_id="local-skill-1", + skill_applied=True, + note="RAW_SKILL_NOTE prompt diff token", + ) + ], + skill_phase_failed_skill_ids=phase_failed_ids or [], + ) + + +def test_gates_default_disabled_and_enabled(monkeypatch, tmp_path): + cases = [ + {"mode": "off", "telemetry_mode": "outbox", "api_key": "k", "quality": True}, + {"mode": "live", "telemetry_mode": "off", "api_key": "k", "quality": True}, + {"mode": "live", "telemetry_mode": "outbox", "api_key": None, "quality": True}, + {"mode": "live", "telemetry_mode": "outbox", "api_key": "k", "quality": False}, + ] + for index, case in enumerate(cases): + _set_cloud_env(monkeypatch, **case) + client = FakeClient() + reporter = CloudSkillQualityReporter( + client=client, + mapping_store=FakeMappingStore( + tmp_path, + {"local-skill-1": SkillCloudBinding("local-skill-1", "cloud-skill-1")}, + ), + outbox=CloudTelemetryOutbox(tmp_path / f"{index}-outbox.db"), + ) + result = _run(reporter.maybe_report_analysis(_analysis())) + assert result["status"] == "skipped" + assert client.calls == [] + + _set_cloud_env(monkeypatch, quality=True) + client = FakeClient() + reporter = CloudSkillQualityReporter( + client=client, + mapping_store=FakeMappingStore( + tmp_path, + {"local-skill-1": SkillCloudBinding("local-skill-1", "cloud-skill-1")}, + ), + outbox=CloudTelemetryOutbox(tmp_path / "enabled-outbox.db"), + ) + result = _run(reporter.maybe_report_analysis(_analysis())) + assert result["status"] == "reported" + assert len(client.calls) == 1 + assert client.calls[0][0] == "skill-use-reported" + + +def test_local_only_and_missing_cloud_binding_skipped(monkeypatch, tmp_path): + _set_cloud_env(monkeypatch, quality=True) + client = FakeClient() + reporter = CloudSkillQualityReporter( + client=client, + mapping_store=FakeMappingStore( + tmp_path, + { + "missing-cloud": SkillCloudBinding("missing-cloud", None), + "bound": SkillCloudBinding("bound", "cloud-bound"), + }, + ), + outbox=CloudTelemetryOutbox(tmp_path / "outbox.db"), + ) + result = _run( + reporter.maybe_report_analysis( + _analysis( + judgments=[ + SkillJudgment("unbound", True, "do not upload"), + SkillJudgment("missing-cloud", True, "do not upload"), + SkillJudgment("bound", True, "upload"), + ] + ) + ) + ) + assert result["reported_count"] == 1 + assert result["skipped_count"] == 2 + assert len(client.calls) == 1 + payload = client.calls[0][1] + assert payload["local_skill_id"] == "bound" + assert payload["cloud_skill_id"] == "cloud-bound" + + +def test_request_id_exact_and_independent_of_mutable_fields(): + judgment = SkillJudgment("local-skill-1", True, "free text ignored") + payload = build_skill_quality_judgment_payload( + _analysis(judgments=[judgment]), + judgment, + cloud_skill_id="cloud-skill-1", + session_id="session-a", + ) + assert payload["request_id"] == short_cloud_request_id( + "skill-quality-judgment", + "task-1", + "local-skill-1", + "cloud-skill-1", + ) + + failed_payload = build_skill_quality_judgment_payload( + _analysis( + timestamp=datetime(2026, 1, 3, 0, 0, 0), + task_completed=False, + judgments=[SkillJudgment("local-skill-1", False, "ignored")], + ), + SkillJudgment("local-skill-1", False, "ignored"), + cloud_skill_id="cloud-skill-1", + session_id="session-b", + ) + assert failed_payload["request_id"] == payload["request_id"] + + +def test_payload_fields_stability_status_and_privacy(): + judgment = SkillJudgment("local-skill-1", True, "RAW_SKILL_NOTE prompt diff") + analysis = _analysis(judgments=[judgment]) + payload = build_skill_quality_judgment_payload( + analysis, + judgment, + cloud_skill_id="cloud-skill-1", + session_id="session-a", + ) + assert payload == build_skill_quality_judgment_payload( + analysis, + judgment, + cloud_skill_id="cloud-skill-1", + session_id="session-a", + ) + assert payload["occurred_at"] == analysis.timestamp.isoformat() + assert "duration_ms" not in payload + assert payload["status"] == "success" + assert "failure_reason" not in payload + assert payload["quality_event_kind"] == QUALITY_EVENT_KIND + assert payload["quality_schema_version"] == QUALITY_SCHEMA_VERSION + assert payload["denominator"] == QUALITY_DENOMINATOR + assert payload["skill_applied"] is True + assert payload["task_completed"] is True + assert payload["skill_phase_failed"] is False + assert payload["completed"] is True + assert payload["fallback"] is False + assert "extras" not in payload + + failed = build_skill_quality_judgment_payload( + _analysis( + task_completed=False, + judgments=[SkillJudgment("local-skill-1", True, "partial_success ignored")], + ), + SkillJudgment("local-skill-1", True, "partial_success ignored"), + cloud_skill_id="cloud-skill-1", + ) + assert failed["status"] == "failed" + assert failed["failure_reason"] == "unknown" + assert failed["completed"] is False + assert failed["fallback"] is True + assert failed["status"] != "partial_success" + + encoded = json.dumps({**payload, **failed}, sort_keys=True) + for forbidden in ( + "RAW_SKILL_NOTE", + "RAW_EXECUTION_NOTE", + "RAW_TOOL_ISSUE", + "prompt", + "transcript", + "file.diff", + "traceback", + "token", + "redacted_preview", + ): + assert forbidden not in encoded + + +def test_phase_failed_success_candidate_reports_failed_unknown(): + judgment = SkillJudgment("local-skill-1", True, "ignored") + payload = build_skill_quality_judgment_payload( + _analysis(judgments=[judgment], phase_failed_ids=["local-skill-1"]), + judgment, + cloud_skill_id="cloud-skill-1", + ) + assert payload["status"] == "failed" + assert payload["failure_reason"] == "unknown" + assert payload["skill_phase_failed"] is True + assert payload["completed"] is False + assert payload["fallback"] is True + + +def test_outbox_redaction_preserves_quality_fields(monkeypatch, tmp_path): + _set_cloud_env(monkeypatch, quality=True) + judgment = SkillJudgment("local-skill-1", False, "RAW note") + payload = build_skill_quality_judgment_payload( + _analysis(task_completed=False, judgments=[judgment]), + judgment, + cloud_skill_id="cloud-skill-1", + ) + redacted = redact_telemetry_payload(payload) + for key in ( + "quality_event_kind", + "quality_schema_version", + "denominator", + "skill_applied", + "task_completed", + "skill_phase_failed", + "completed", + "fallback", + ): + assert key in redacted + + outbox = CloudTelemetryOutbox(tmp_path / "outbox.db") + row = outbox.enqueue( + endpoint="/api/v2/telemetry/skill-use-reported", + payload=payload, + ) + assert row.payload_redacted["quality_event_kind"] == QUALITY_EVENT_KIND + assert row.payload_redacted["denominator"] == QUALITY_DENOMINATOR + assert row.payload_redacted["fallback"] is True + + +def test_repeated_report_uses_same_outbox_row_for_same_payload(monkeypatch, tmp_path): + _set_cloud_env(monkeypatch, quality=True) + client = FakeClient(fail=True) + outbox = CloudTelemetryOutbox(tmp_path / "outbox.db") + reporter = CloudSkillQualityReporter( + client=client, + mapping_store=FakeMappingStore( + tmp_path, + {"local-skill-1": SkillCloudBinding("local-skill-1", "cloud-skill-1")}, + ), + outbox=outbox, + ) + analysis = _analysis() + first = _run(reporter.maybe_report_analysis(analysis, session_id="session-a")) + second = _run(reporter.maybe_report_analysis(analysis, session_id="session-a")) + assert first["queued_count"] == 1 + assert second["queued_count"] == 1 + failed_rows = outbox.list_by_status("failed") + assert len(failed_rows) == 1 + + third = _run(reporter.maybe_report_analysis(analysis, session_id="session-b")) + assert third["queued_count"] == 1 + failed_rows = outbox.list_by_status("failed") + assert len(failed_rows) == 2 + assert {row.request_id for row in failed_rows} == { + short_cloud_request_id( + "skill-quality-judgment", + "task-1", + "local-skill-1", + "cloud-skill-1", + ) + }