mirror of
https://github.com/HKUDS/OpenSpace.git
synced 2026-08-28 05:15:00 +00:00
Add gated skill quality reporter
This commit is contained in:
Dennis-yxchen
parent
e37f65f12e
commit
026c00b8d2
9 changed files with 1082 additions and 2 deletions
5
.gitignore
vendored
5
.gitignore
vendored
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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",
|
||||
|
|
|
|||
254
openspace/cloud/skill_quality_reporter.py
Normal file
254
openspace/cloud/skill_quality_reporter.py
Normal file
|
|
@ -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
|
||||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
334
tests/test_analyzer_skill_quality_reporting.py
Normal file
334
tests/test_analyzer_skill_quality_reporting.py
Normal file
|
|
@ -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
|
||||
345
tests/test_skill_quality_reporter.py
Normal file
345
tests/test_skill_quality_reporter.py
Normal file
|
|
@ -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",
|
||||
)
|
||||
}
|
||||
Loading…
Add table
Reference in a new issue