"""Deterministic quality signal detector.""" from __future__ import annotations import hashlib import json from dataclasses import replace from typing import Any, Iterable, Mapping, Sequence from openspace.skill_engine.evidence import EvidenceScope, EvidenceStore, ResourceRef from .linkers import ( EvidenceIndexes, SkillContextCandidate, build_evidence_indexes, has_representative_tool_evidence, latest_ref, metadata_values, ref_sort_key, ref_tool_key, related_tool_incidents, related_tool_results, resolve_skill_contexts, tool_use_key, ) from .types import ( ACTION_MANUAL_REVIEW, ACTION_OBSERVE_ONLY, ACTION_TRIGGER_REVIEW, SIGNAL_AGGREGATE_WITHOUT_INCIDENT, SIGNAL_TOOL_CALL_FAILED, SIGNAL_TOOL_FAILURE_AFFECTS_SKILL, STATUS_ACTIONABLE_PARTIAL, STATUS_AGGREGATE_ONLY, STATUS_AMBIGUOUS_SUBJECT, STATUS_CONFLICTING, STATUS_COMPLETE, STATUS_EXTERNAL_ONLY, STATUS_MISSING_SKILL_CONTEXT, QualitySignal, choose_dominant_signal, stable_unique, ) CHECKPOINT_TASK_SESSION_PERSISTED = "task_session_persisted" _BASE_REF_TYPES = [ "tool_event", "tool_result", "tool_incident", "tool_quality_record", "skill_event", "skill_record", "skill_file", "execution_analysis", "transcript_message", ] _SKILL_CONTEXT_REF_TYPES = ["skill_record", "skill_file"] _TOOL_CONTEXT_REF_TYPES = ["tool_quality_record", "tool_incident"] class QualitySignalDetector: def __init__(self, evidence_store: EvidenceStore) -> None: self.evidence_store = evidence_store self.warnings: list[str] = [] def scan_checkpoint( self, *, checkpoint_name: str, scope: EvidenceScope, manifest_watermark: int, ) -> list[QualitySignal]: self.warnings = [] if checkpoint_name != CHECKPOINT_TASK_SESSION_PERSISTED: return [] refs = self._load_refs(scope, manifest_watermark) indexes = build_evidence_indexes(refs) signals: list[QualitySignal] = [] for tool_event in indexes.failed_tool_events: tool_call_signal = self._tool_call_failed_signal( tool_event, indexes, manifest_watermark=manifest_watermark, ) if tool_call_signal is not None: signals.append(tool_call_signal) skill_signal = self._tool_failure_affects_skill_signal( tool_event, indexes, manifest_watermark=manifest_watermark, ) if skill_signal is not None: signals.append(skill_signal) signals.extend( self._aggregate_without_incident_signals( indexes, manifest_watermark=manifest_watermark, ) ) return signals def _load_refs( self, scope: EvidenceScope, manifest_watermark: int, ) -> list[ResourceRef]: refs = self.evidence_store.query_refs( scope, ref_types=list(_BASE_REF_TYPES), watermark=manifest_watermark, ) indexes = build_evidence_indexes(refs) skill_ids = sorted( { *indexes.skill_events_by_skill_id, *[item for item in scope.skill_ids if item], } ) tool_keys = sorted( { *[ ref_tool_key(ref) for bucket in indexes.tool_events_by_use_id.values() for ref in bucket if ref_tool_key(ref) ], *[item for item in scope.tool_keys if item], } ) extra_refs: list[ResourceRef] = [] if skill_ids: extra_refs.extend( self.evidence_store.query_refs( EvidenceScope(skill_ids=tuple(skill_ids)), ref_types=list(_SKILL_CONTEXT_REF_TYPES), watermark=manifest_watermark, ) ) if tool_keys: extra_refs.extend( ref for ref in self.evidence_store.query_refs( EvidenceScope( session_id=scope.session_id, task_id=scope.task_id, source_task_ids=scope.source_task_ids, agent_ids=scope.agent_ids, tool_keys=tuple(tool_keys), ), ref_types=list(_TOOL_CONTEXT_REF_TYPES), watermark=manifest_watermark, ) if _same_checkpoint_context(scope, ref) ) by_ref_id = {ref.ref_id: ref for ref in [*refs, *extra_refs] if ref.ref_id} return sorted(by_ref_id.values(), key=ref_sort_key) def _tool_call_failed_signal( self, tool_event: ResourceRef, indexes: EvidenceIndexes, *, manifest_watermark: int, ) -> QualitySignal | None: tool_key = ref_tool_key(tool_event) if not tool_key: self.warnings.append(f"malformed_tool_event_missing_tool_key:{tool_event.ref_id}") return None results = related_tool_results(tool_event, indexes) incidents = related_tool_incidents(tool_event, indexes) raw_backrefs = stable_unique( [tool_event.ref_id, *[ref.ref_id for ref in results], *[ref.ref_id for ref in incidents]] ) return self._signal( signal_type=SIGNAL_TOOL_CALL_FAILED, subject_type="tool", subject_id=tool_key, actionability=ACTION_OBSERVE_ONLY, evidence_status=STATUS_COMPLETE, failure_signature=stable_failure_signature( SIGNAL_TOOL_CALL_FAILED, tool_event=tool_event, result_refs=results, ), raw_backrefs=raw_backrefs, source_ref=tool_event, tool_key=tool_key, skill_id=None, skill_version=None, manifest_watermark=manifest_watermark, metadata=_summary_metadata( tool_event, result_refs=results, incident_refs=incidents, linkage="tool_event_failed", ) | {"policy_reason": "observe_tool_event_failed"}, ) def _tool_failure_affects_skill_signal( self, tool_event: ResourceRef, indexes: EvidenceIndexes, *, manifest_watermark: int, ) -> QualitySignal | None: tool_key = ref_tool_key(tool_event) if not tool_key: return None all_results = related_tool_results(tool_event, indexes) results = _complete_results(all_results) incidents = related_tool_incidents(tool_event, indexes) contexts = resolve_skill_contexts(tool_event, indexes) supporting_refs = [*results, *incidents] if not contexts: return self._signal( signal_type=SIGNAL_TOOL_FAILURE_AFFECTS_SKILL, subject_type="tool", subject_id=tool_key, actionability=ACTION_OBSERVE_ONLY, evidence_status=STATUS_MISSING_SKILL_CONTEXT, failure_signature=stable_failure_signature( SIGNAL_TOOL_FAILURE_AFFECTS_SKILL, tool_event=tool_event, result_refs=results, ), raw_backrefs=stable_unique( [tool_event.ref_id, *[ref.ref_id for ref in supporting_refs]] ), source_ref=tool_event, tool_key=tool_key, skill_id=None, skill_version=None, manifest_watermark=manifest_watermark, missing_refs=("skill_event", "skill_file"), metadata=_summary_metadata( tool_event, result_refs=results, incident_refs=incidents, linkage="missing_skill_context", ) | {"policy_reason": "missing_skill_context"}, ) if len(contexts) > 1: raw_backrefs = [ tool_event.ref_id, *[ref.ref_id for ref in supporting_refs], *[context.skill_event.ref_id for context in contexts], ] return self._signal( signal_type=SIGNAL_TOOL_FAILURE_AFFECTS_SKILL, subject_type="tool", subject_id=tool_key, actionability=ACTION_MANUAL_REVIEW, evidence_status=STATUS_AMBIGUOUS_SUBJECT, failure_signature=stable_failure_signature( SIGNAL_TOOL_FAILURE_AFFECTS_SKILL, tool_event=tool_event, result_refs=results, ), raw_backrefs=stable_unique(raw_backrefs), source_ref=tool_event, tool_key=tool_key, skill_id=None, skill_version=None, manifest_watermark=manifest_watermark, missing_refs=("unique_skill_context",), metadata={ **_summary_metadata( tool_event, result_refs=results, incident_refs=incidents, linkage="ambiguous_skill_context", ), "candidate_skill_ids": [context.skill_id for context in contexts], "policy_reason": "ambiguous_skill_context", }, ) context = contexts[0] skill_version = _skill_version(context) conflict_reasons = _conflicting_evidence_reasons( result_refs=all_results, incident_refs=incidents, ) if context.skill_file is None: status = STATUS_MISSING_SKILL_CONTEXT actionability = ACTION_OBSERVE_ONLY missing_refs = ("skill_file",) policy_reason = "missing_skill_file" elif conflict_reasons: status = STATUS_CONFLICTING actionability = ACTION_MANUAL_REVIEW missing_refs = ("consistent_tool_evidence",) policy_reason = "conflicting_tool_evidence" elif _is_permission_or_user_denied(tool_event): status = STATUS_EXTERNAL_ONLY actionability = ACTION_OBSERVE_ONLY missing_refs = () policy_reason = "permission_denied_without_skill_evidence" elif supporting_refs: status = STATUS_COMPLETE actionability = ACTION_TRIGGER_REVIEW missing_refs = () policy_reason = "complete_exact_failure_evidence" else: status = STATUS_ACTIONABLE_PARTIAL actionability = ACTION_TRIGGER_REVIEW missing_refs = ("tool_result", "tool_incident") policy_reason = "actionable_partial_failed_tool_event" raw_tool_refs = [*all_results, *incidents] if conflict_reasons else supporting_refs raw_backrefs = [ tool_event.ref_id, *[ref.ref_id for ref in raw_tool_refs], context.skill_event.ref_id, ] if context.skill_record is not None: raw_backrefs.append(context.skill_record.ref_id) if context.skill_file is not None: raw_backrefs.append(context.skill_file.ref_id) return self._signal( signal_type=SIGNAL_TOOL_FAILURE_AFFECTS_SKILL, subject_type="tool_skill_relation", subject_id=f"{context.skill_id}:{tool_key}", actionability=actionability, evidence_status=status, failure_signature=stable_failure_signature( SIGNAL_TOOL_FAILURE_AFFECTS_SKILL, tool_event=tool_event, skill_id=context.skill_id, result_refs=results, ), raw_backrefs=stable_unique(raw_backrefs), source_ref=tool_event, tool_key=tool_key, skill_id=context.skill_id, skill_version=skill_version, manifest_watermark=manifest_watermark, missing_refs=missing_refs, metadata={ **_summary_metadata( tool_event, result_refs=results, incident_refs=incidents, linkage=context.link_type, ), "skill_context_ref": context.skill_event.ref_id, "policy_reason": policy_reason, **({"conflict_reasons": conflict_reasons} if conflict_reasons else {}), }, ) def _aggregate_without_incident_signals( self, indexes: EvidenceIndexes, *, manifest_watermark: int, ) -> list[QualitySignal]: signals: list[QualitySignal] = [] for tool_key, records in sorted(indexes.quality_records_by_tool_key.items()): if has_representative_tool_evidence(tool_key, indexes): continue record = latest_ref(records) if record is None: continue signals.append( self._signal( signal_type=SIGNAL_AGGREGATE_WITHOUT_INCIDENT, subject_type="tool", subject_id=tool_key, actionability=ACTION_OBSERVE_ONLY, evidence_status=STATUS_AGGREGATE_ONLY, failure_signature=stable_failure_signature( SIGNAL_AGGREGATE_WITHOUT_INCIDENT, tool_key=tool_key, ), raw_backrefs=(record.ref_id,), source_ref=record, tool_key=tool_key, skill_id=None, skill_version=None, manifest_watermark=manifest_watermark, metadata={ "summary": "aggregate quality record has no representative incident in scope", "linkage": "aggregate_without_incident", "policy_reason": "aggregate_without_incident", "recent_success_rate": _safe_number( record.metadata.get("recent_success_rate") ), }, ) ) return signals def _signal( self, *, signal_type: str, subject_type: str, subject_id: str, actionability: str, evidence_status: str, failure_signature: str, raw_backrefs: Sequence[str], source_ref: ResourceRef, tool_key: str | None, skill_id: str | None, skill_version: str | None, manifest_watermark: int, missing_refs: Sequence[str] = (), metadata: Mapping[str, Any] | None = None, ) -> QualitySignal: merge_key = ( f"{subject_type}:{subject_id}:{signal_type}:" f"{failure_signature}:{skill_version or 'no_skill_version'}" ) return QualitySignal( signal_id=_signal_id(merge_key, source_ref.ref_id), signal_type=signal_type, subject_type=subject_type, subject_id=subject_id, actionability=actionability, evidence_status=evidence_status, merge_key=merge_key, failure_signature=failure_signature, raw_backrefs=tuple(raw_backrefs), session_id=source_ref.session_id, task_id=source_ref.task_id, parent_task_id=source_ref.parent_task_id, agent_id=source_ref.agent_id, tool_key=tool_key, skill_id=skill_id, skill_version=skill_version, source_watermark=manifest_watermark, missing_refs=tuple(stable_unique(missing_refs)), metadata=dict(metadata or {}), ) def stable_failure_signature( signal_type: str, *, tool_event: ResourceRef | None = None, tool_key: str | None = None, skill_id: str | None = None, result_refs: Sequence[ResourceRef] = (), ) -> str: event_metadata = tool_event.metadata if tool_event is not None else {} resolved_tool_key = tool_key or (ref_tool_key(tool_event) if tool_event else "") payload = [ signal_type, resolved_tool_key or "", skill_id or "", _normalized_error_type(event_metadata), _metadata_text(event_metadata, "permission_status"), _metadata_text(event_metadata, "exception_class", "exception_type"), _normalized_result_shape(result_refs), ] return _digest(payload) def _summary_metadata( tool_event: ResourceRef, *, result_refs: Sequence[ResourceRef], incident_refs: Sequence[ResourceRef], linkage: str, ) -> dict[str, Any]: metadata = tool_event.metadata tool_key = ref_tool_key(tool_event) error_type = _normalized_error_type(metadata) result_shape = _normalized_result_shape(result_refs) summary_parts = [ f"tool={tool_key or 'unknown'}", f"status={str(metadata.get('status') or 'unknown').lower()}", ] if error_type: summary_parts.append(f"error_type={error_type}") return _without_none( { "summary": " ".join(summary_parts), "linkage": linkage, "tool_use_id": tool_use_key(tool_event).tool_use_id or None, "error_type": error_type or None, "exception_class": _metadata_text( metadata, "exception_class", "exception_type", ) or None, "permission_status": _metadata_text(metadata, "permission_status") or None, "result_shape": result_shape or None, "result_ref_count": len(result_refs), "incident_ref_count": len(incident_refs), } ) def _complete_results(refs: Sequence[ResourceRef]) -> list[ResourceRef]: return [ ref for ref in refs if not bool(ref.metadata.get("missing")) ] _SUCCESS_STATUSES_BY_REF_TYPE: Mapping[str, frozenset[str]] = { "tool_event": frozenset({"ok", "passed", "success", "succeeded"}), "tool_result": frozenset({"ok", "passed", "success", "succeeded"}), "tool_incident": frozenset(), } def _conflicting_evidence_reasons( *, result_refs: Sequence[ResourceRef], incident_refs: Sequence[ResourceRef], ) -> list[str]: reasons: list[str] = [] for ref in sorted([*result_refs, *incident_refs], key=ref_sort_key): if _is_successish_ref(ref): reasons.append(f"{ref.ref_type}:{ref.ref_id}:success") return reasons def _is_successish_ref(ref: ResourceRef) -> bool: metadata = ref.metadata success = metadata.get("success") if success is True or _truthy_text(success): return True status = str(metadata.get("status") or "").strip().lower() if status in _SUCCESS_STATUSES_BY_REF_TYPE.get(ref.ref_type, frozenset()): return True is_error = metadata.get("is_error") return is_error is False or _falsy_text(is_error) def _is_permission_or_user_denied(ref: ResourceRef) -> bool: metadata = ref.metadata values = { str(metadata.get(key) or "").strip().lower() for key in ( "status", "permission_status", "error_type", "error_bucket", "exception_class", "exception_type", ) } values.discard("") if values.intersection({"blocked", "denied", "permission_denied", "rejected"}): return True text = " ".join( str(metadata.get(key) or "").strip().lower() for key in ("error_message", "message", "reason") ) return "permission" in text and "denied" in text def _truthy_text(value: Any) -> bool: return isinstance(value, str) and value.strip().lower() in {"1", "true", "yes"} def _falsy_text(value: Any) -> bool: return isinstance(value, str) and value.strip().lower() in {"0", "false", "no"} def _skill_version(context: SkillContextCandidate) -> str | None: record = context.skill_record if record is not None: for key in ("skill_version", "version", "generation", "lineage_generation"): value = record.metadata.get(key) if value is not None and str(value) != "": return str(value) if context.skill_file is not None: return str(context.skill_file.hash or context.skill_file.ref_id) return None def _normalized_error_type(metadata: Mapping[str, Any]) -> str: return _metadata_text( metadata, "error_type", "exception_type", "failure_mode", "error_bucket", "error_code", "exit_code", ) def _normalized_result_shape(result_refs: Sequence[ResourceRef]) -> str: shapes: list[str] = [] for ref in sorted(result_refs, key=ref_sort_key): metadata = ref.metadata source = _metadata_text(metadata, "persistence_source") missing = "missing" if metadata.get("missing") else "present" length = _metadata_text(metadata, "original_length", "persisted_output_size") shapes.append(":".join(item for item in (source, missing, length) if item)) return "|".join(shapes) def _metadata_text(metadata: Mapping[str, Any], *keys: str) -> str: values = metadata_values(metadata, *keys) if not values: return "" return sorted(values)[0] def _without_none(values: Mapping[str, Any]) -> dict[str, Any]: return {key: value for key, value in values.items() if value is not None} def _signal_id(merge_key: str, source_ref_id: str) -> str: return f"qsig_{_digest([merge_key, source_ref_id])}" def _safe_number(value: Any) -> int | float | None: if isinstance(value, (int, float)): return value try: return float(value) except (TypeError, ValueError): return None def _same_checkpoint_context(scope: EvidenceScope, ref: ResourceRef) -> bool: if scope.session_id and ref.session_id != scope.session_id: return False task_ids = { item for item in ( scope.task_id, *scope.source_task_ids, ) if item } if task_ids: ref_task_ids = { str(item) for item in ( ref.task_id, ref.parent_task_id, ref.metadata.get("task_id"), ref.metadata.get("parent_task_id"), ) if item } if not ref_task_ids.intersection(task_ids): return False if scope.agent_ids and ref.agent_id and ref.agent_id not in scope.agent_ids: return False return True def _digest(value: Any) -> str: data = json.dumps(value, sort_keys=True, separators=(",", ":"), default=str) return hashlib.sha256(data.encode("utf-8")).hexdigest()[:16] def merge_signals_by_key(signals: Iterable[QualitySignal]) -> list[QualitySignal]: merged: dict[str, QualitySignal] = {} for signal in signals: existing = merged.get(signal.merge_key) if existing is None: merged[signal.merge_key] = signal continue dominant = choose_dominant_signal(existing, signal) merged[signal.merge_key] = replace( dominant, raw_backrefs=stable_unique((*existing.raw_backrefs, *signal.raw_backrefs)), ) return list(merged.values())