"""Rule-based admission gates for evidence-backed skill evolution.""" from __future__ import annotations import hashlib import json import re from dataclasses import asdict, dataclass, field from datetime import datetime, timezone from typing import Any, Mapping from openspace.skill_engine.evidence import EvidencePacket, ResourceRef from openspace.skill_engine.signals.types import ( STATUS_ACTIONABLE_PARTIAL, STATUS_AGGREGATE_ONLY, TRIGGERABLE_EVIDENCE_STATUSES, ) from openspace.utils.logging import Logger logger = Logger.get_logger(__name__) _OUTCOMES = {"direct", "candidate", "reject", "noop", "needs_human_review"} _FALLBACK_CAPTURE_REF_TYPES = { "background_task_result", "memory_ref", "recording_ref", } _SIMPLE_CAPTURE_TOOLS = {"cat", "date", "ls", "pwd", "rg", "sed"} _ENVIRONMENT_FAILURE_TERMS = ( "api key", "apikey", "openai_api_key", "anthropic_api_key", "missing key", "unauthorized", "authentication", "auth failed", "401", "403", "network outage", "network error", "dns", "connection refused", "connection reset", "service unavailable", "external service", "sandbox", "permission mode", ) _EPHEMERAL_CAPTURE_TERMS = ( "api key", "apikey", "_api_key", "secret", "token", "password", "credential", "one-time", "one time", "single-use", "temporary url", "signed url", "presigned", "session-specific", "temporary environment", "/tmp/", "/private/tmp", ) _DERIVED_DIVERGENCE_TERMS = ( "divergence", "diverge", "subscenario", "sub-scenario", "specialize", "specialized", "stable sub", "scope too broad", "too broad", "too wide", "different tool", "different tools", "tool combination", "cannot stay clear", "separate workflow", ) _BUGFIX_TERMS = ("bug", "fix", "broken", "failure", "failed", "repair") _WORKFLOW_BOUNDARY_TERMS = ( "workflow", "multi-step", "multistep", "strategy", "reusable", "when to use", "when-to-use", "input", "output", "domain", "stable", ) _USER_EXPLICIT_TERMS = ( "manual", "user_requested", "user requested", "explicit", "capture_requested", ) _QUALITY_SIGNAL_EXTERNAL_ATTRIBUTIONS = { "attribution:permission", "attribution:environment", "attribution:tool_external", } @dataclass(frozen=True, slots=True) class AdmissionResult: admission_id: str decision_id: str packet_id: str outcome: str hard_failures: list[str] = field(default_factory=list) warnings: list[str] = field(default_factory=list) required_refs_checked: list[str] = field(default_factory=list) reviewed_by: str = "rule" created_at: str = "" def to_dict(self) -> dict[str, Any]: return asdict(self) @classmethod def from_mapping(cls, data: Mapping[str, Any]) -> "AdmissionResult": outcome = str(data.get("outcome") or "reject").strip().lower() if outcome not in _OUTCOMES: outcome = "reject" return cls( admission_id=str(data.get("admission_id") or ""), decision_id=str(data.get("decision_id") or ""), packet_id=str(data.get("packet_id") or ""), outcome=outcome, hard_failures=_str_list(data.get("hard_failures")), warnings=_str_list(data.get("warnings")), required_refs_checked=_str_list(data.get("required_refs_checked")), reviewed_by=str(data.get("reviewed_by") or "rule"), created_at=str(data.get("created_at") or ""), ) class EvolutionAdmission: """Deterministic hard gate between decision proposals and authoring.""" def __init__( self, *, evidence_store: Any | None = None, skill_store: Any | None = None, registry: Any | None = None, recurrence_threshold: int = 2, ) -> None: self.evidence_store = evidence_store self.skill_store = skill_store self.registry = registry self.recurrence_threshold = max(2, int(recurrence_threshold or 2)) def admit( self, decision: Any, packet: EvidencePacket, job: Any | None = None, ) -> AdmissionResult: try: result = self._admit(decision, packet, job) except Exception as exc: logger.debug("Evolution admission failed", exc_info=True) result = self._result( decision, packet, outcome="reject", hard_failures=["admission_error"], warnings=[str(exc)[:500]], required_refs_checked=_claim_ref_ids(decision), ) self._persist(result) return result def _admit( self, decision: Any, packet: EvidencePacket, job: Any | None = None, ) -> AdmissionResult: action = _action(decision) checked = _claim_ref_ids(decision) hard_failures = self._base_hard_failures(decision, packet, job) warnings: list[str] = [] if action == "NOOP": warnings.extend(_str_list(_attr(decision, "reason_tags"))) noop_reason = _none_or_str(_attr(decision, "noop_reason")) if noop_reason: warnings.append(noop_reason) return self._result( decision, packet, outcome="noop", hard_failures=hard_failures, warnings=warnings, required_refs_checked=checked, ) if _candidate_policy(decision) == "reject": hard_failures.append("decision_policy_reject") hard_failures.extend(_str_list(_attr(decision, "risks"))) if hard_failures: return self._result( decision, packet, outcome="reject", hard_failures=hard_failures, warnings=warnings, required_refs_checked=checked, ) if action == "FIX": return self._admit_fix(decision, packet, checked, job=job) if action == "DERIVED": return self._admit_derived(decision, packet, checked) if action == "CAPTURED": return self._admit_captured(decision, packet, checked) return self._result( decision, packet, outcome="reject", hard_failures=["unsupported_action"], required_refs_checked=checked, ) def _base_hard_failures( self, decision: Any, packet: EvidencePacket, job: Any | None, ) -> list[str]: failures: list[str] = [] claims = list(_attr(decision, "evidence_claims") or []) if not claims: failures.append("no_evidence_claims") valid_refs = _packet_ref_ids(packet) missing_refs: list[str] = [] for index, claim in enumerate(claims): refs = _str_list(_attr(claim, "refs")) if not refs: failures.append(f"claim_{index}_missing_refs") continue missing_refs.extend(ref_id for ref_id in refs if ref_id not in valid_refs) if missing_refs: failures.append("missing_refs") failures.extend(f"missing_ref:{ref_id}" for ref_id in missing_refs) failures.extend(_quality_signal_packet_failures(packet, job)) return list(dict.fromkeys(failures)) def _admit_fix( self, decision: Any, packet: EvidencePacket, checked: list[str], *, job: Any | None = None, ) -> AdmissionResult: target_skill_ids = _target_skill_ids(decision) hard_failures: list[str] = [] warnings: list[str] = [] if not target_skill_ids: hard_failures.append("missing_target_skill") for skill_id in target_skill_ids: if not self._skill_exists(skill_id, packet): hard_failures.append("unknown_target_skill") hard_failures.append(f"unknown_target_skill:{skill_id}") skill_file_refs = _skill_file_refs(packet, target_skill_ids) lifecycle_refs = _skill_lifecycle_refs(packet, target_skill_ids) manual_fix_refs = _manual_fix_request_refs(packet) failure_refs = _failure_or_friction_refs(packet) target_state_refs = _target_skill_state_refs(packet, target_skill_ids) tool_quality_refs = list(_refs(packet, "tool_quality_record")) tool_incident_refs = list(_refs(packet, "tool_incident")) quality_signal_refs = _quality_signal_refs(packet) tool_dependency_warning = _tool_quality_dependency_warning( packet, target_skill_ids, [*tool_quality_refs, *tool_incident_refs], ) tool_dependency_confirmed = ( bool(tool_quality_refs or tool_incident_refs) and tool_dependency_warning is None ) checked.extend(ref.ref_id for ref in skill_file_refs) checked.extend(ref.ref_id for ref in lifecycle_refs) checked.extend(ref.ref_id for ref in manual_fix_refs) checked.extend(ref.ref_id for ref in failure_refs) checked.extend(ref.ref_id for ref in target_state_refs) checked.extend(ref.ref_id for ref in tool_quality_refs) checked.extend(ref.ref_id for ref in tool_incident_refs) checked.extend(ref.ref_id for ref in quality_signal_refs) if not skill_file_refs: hard_failures.append("missing_skill_file_ref") if not lifecycle_refs and not manual_fix_refs and not tool_dependency_confirmed: hard_failures.append("missing_skill_lifecycle_ref") if not failure_refs and not manual_fix_refs: hard_failures.append("missing_failure_evidence") if ( tool_quality_refs and not tool_incident_refs and not _is_quality_signal_context(packet, None) ): hard_failures.append("tool_quality_aggregate_without_incident") hard_failures.extend( _quality_signal_fix_failures( decision, packet, target_skill_ids=target_skill_ids, ) ) if hard_failures: return self._result( decision, packet, outcome="reject", hard_failures=hard_failures, warnings=warnings, required_refs_checked=checked, ) environment_refs = [ref for ref in failure_refs if _is_environment_failure(ref)] non_environment_refs = [ref for ref in failure_refs if ref not in environment_refs] if environment_refs and not non_environment_refs: warnings.append("environment_failure") warnings.extend(_environment_reason_tags(environment_refs)) return self._result( decision, packet, outcome="noop", warnings=warnings, required_refs_checked=checked, ) if tool_dependency_warning: warnings.append(tool_dependency_warning) return self._result( decision, packet, outcome="candidate", warnings=warnings, required_refs_checked=checked, ) if not manual_fix_refs and not _fix_causality_plausible(decision, packet, failure_refs): warnings.append("causality_uncertain") return self._result( decision, packet, outcome="candidate", warnings=warnings, required_refs_checked=checked, ) return self._result( decision, packet, outcome="direct", warnings=warnings, required_refs_checked=checked, ) def _admit_derived( self, decision: Any, packet: EvidencePacket, checked: list[str], ) -> AdmissionResult: target_skill_ids = _target_skill_ids(decision) hard_failures: list[str] = [] warnings: list[str] = [] if not target_skill_ids: hard_failures.append("missing_parent_skill") for skill_id in target_skill_ids: if not self._skill_exists(skill_id, packet): hard_failures.append("unknown_parent_skill") hard_failures.append(f"unknown_parent_skill:{skill_id}") skill_file_refs = _skill_file_refs(packet, target_skill_ids) checked.extend(ref.ref_id for ref in skill_file_refs) if not skill_file_refs: hard_failures.append("missing_parent_skill_file_ref") if hard_failures: return self._result( decision, packet, outcome="reject", hard_failures=hard_failures, required_refs_checked=checked, ) text = _decision_text(decision) has_divergence = _contains_any(text, _DERIVED_DIVERGENCE_TERMS) repeated_or_manual = self._is_repeated_or_user_explicit(decision, packet) if not has_divergence: warnings.append("no_derived_divergence") if _contains_any(text, _BUGFIX_TERMS) and not has_divergence: warnings.append("bugfix_should_be_fix") if not repeated_or_manual: warnings.append("single_observation") outcome = "candidate" if has_divergence and repeated_or_manual: outcome = "direct" return self._result( decision, packet, outcome=outcome, warnings=warnings, required_refs_checked=checked, ) def _admit_captured( self, decision: Any, packet: EvidencePacket, checked: list[str], ) -> AdmissionResult: warnings: list[str] = [] if not _task_completed(packet): return self._result( decision, packet, outcome="noop", warnings=["task_incomplete"], required_refs_checked=checked, ) claim_ref_types = _ref_types_for_ids(packet, _claim_ref_ids(decision)) if claim_ref_types and claim_ref_types.issubset(_FALLBACK_CAPTURE_REF_TYPES): return self._result( decision, packet, outcome="noop", warnings=["fallback_only_capture_evidence"], required_refs_checked=checked, ) if _existing_skill_covers(decision): return self._result( decision, packet, outcome="noop", warnings=["existing_skill_covers_workflow"], required_refs_checked=checked, ) if _capture_depends_on_ephemeral_context(packet, decision): return self._result( decision, packet, outcome="noop", warnings=["ephemeral_or_secret_dependent_capture"], required_refs_checked=checked, ) if _is_simple_tool_capture(packet): return self._result( decision, packet, outcome="noop", warnings=["low_signal_capture", "simple_tool_task"], required_refs_checked=checked, ) if not _nontrivial_workflow(packet, decision): warnings.append("workflow_trivial_or_uncertain") if not _reusable_boundary(decision, packet): warnings.append("reusable_boundary_uncertain") repeated_or_manual = self._is_repeated_or_user_explicit(decision, packet) if not repeated_or_manual: warnings.append("single_observation") outcome = "candidate" if ( repeated_or_manual and "workflow_trivial_or_uncertain" not in warnings and "reusable_boundary_uncertain" not in warnings ): outcome = "direct" return self._result( decision, packet, outcome=outcome, warnings=warnings, required_refs_checked=checked, ) def _is_repeated_or_user_explicit(self, decision: Any, packet: EvidencePacket) -> bool: return _is_repeated_or_user_explicit( decision, packet, recurrence_threshold=self.recurrence_threshold, ) def _skill_exists(self, skill_id: str, packet: EvidencePacket) -> bool: if not skill_id: return False for ref_type in ("skill_file", "skill_record", "skill_event"): for ref in _refs(packet, ref_type): values = _metadata_values(ref.metadata, "skill_id", "skill_ids") if skill_id in values or skill_id in ref.ref_id: return True for source in (self.skill_store, self.registry): if source is None: continue for method_name in ("load_record", "get_skill", "get", "load"): method = getattr(source, method_name, None) if not callable(method): continue try: if method(skill_id) is not None: return True except Exception: logger.debug("Skill lookup failed for %s", skill_id, exc_info=True) break return False def _result( self, decision: Any, packet: EvidencePacket, *, outcome: str, hard_failures: list[str] | None = None, warnings: list[str] | None = None, required_refs_checked: list[str] | None = None, ) -> AdmissionResult: normalized_outcome = outcome if outcome in _OUTCOMES else "reject" failure_tags = list(dict.fromkeys(_str_list(hard_failures))) warning_tags = list(dict.fromkeys(_str_list(warnings))) refs_checked = list(dict.fromkeys(_str_list(required_refs_checked))) decision_id = str(_attr(decision, "decision_id") or "") packet_id = str(getattr(packet, "packet_id", "") or "") created_at = _utc_now() admission_id = "adm_" + _digest( { "decision_id": decision_id, "packet_id": packet_id, "outcome": normalized_outcome, "hard_failures": failure_tags, "warnings": warning_tags, "refs": refs_checked, } )[:16] return AdmissionResult( admission_id=admission_id, decision_id=decision_id, packet_id=packet_id, outcome=normalized_outcome, hard_failures=failure_tags, warnings=warning_tags, required_refs_checked=refs_checked, reviewed_by="rule", created_at=created_at, ) def _persist(self, result: AdmissionResult) -> None: persist = getattr(self.evidence_store, "persist_admission", None) if not callable(persist): return try: persist(result) except Exception: logger.debug( "Failed to persist admission result %s", result.admission_id, exc_info=True, ) def _action(decision: Any) -> str: raw = ( _attr(decision, "proposed_action") or _attr(decision, "action_type") or _attr(decision, "evolution_type") or "" ) return str(getattr(raw, "value", raw) or "").strip().upper() def _candidate_policy(decision: Any) -> str: return str(_attr(decision, "candidate_policy") or "").strip().lower() def _target_skill_ids(decision: Any) -> list[str]: return _str_list(_attr(decision, "target_skill_ids") or _attr(decision, "target_skills")) def _claim_ref_ids(decision: Any) -> list[str]: refs: list[str] = [] for claim in list(_attr(decision, "evidence_claims") or []): refs.extend(_str_list(_attr(claim, "refs"))) return list(dict.fromkeys(refs)) def _packet_ref_ids(packet: EvidencePacket) -> set[str]: refs = { ref.ref_id for group in packet.selected_refs.values() for ref in group if ref.ref_id } refs.update(path.ref_id for path in packet.readable_paths if path.ref_id) return refs def _refs(packet: EvidencePacket, ref_type: str) -> list[ResourceRef]: return list(packet.selected_refs.get(ref_type) or []) def _quality_signal_refs(packet: EvidencePacket) -> list[ResourceRef]: return _refs(packet, "quality_signal_ref") def _is_quality_signal_context( packet: EvidencePacket, job: Any | None, ) -> bool: trigger_type = str(_attr(job, "trigger_type") or "").strip().upper() if trigger_type == "QUALITY_SIGNAL": return True if str(getattr(packet, "profile_name", "") or "") == "quality_signal": return True return bool(_quality_signal_refs(packet)) def _quality_signal_packet_failures( packet: EvidencePacket, job: Any | None, ) -> list[str]: if not _is_quality_signal_context(packet, job): return [] refs = _quality_signal_refs(packet) if not refs: return ["missing_quality_signal_ref"] failures: list[str] = [] for ref in refs: metadata = ref.metadata actionability = str(metadata.get("actionability") or "").strip() evidence_status = str(metadata.get("evidence_status") or "").strip() signal_type = str(metadata.get("signal_type") or "").strip() if actionability != "trigger_review": failures.append("quality_signal_not_trigger_review") if evidence_status == STATUS_AGGREGATE_ONLY: failures.append("quality_signal_aggregate_only") failures.append("aggregate_only_quality_source") elif evidence_status not in TRIGGERABLE_EVIDENCE_STATUSES: failures.append("quality_signal_incomplete") if signal_type == "aggregate_without_incident": failures.append("quality_signal_aggregate_only") failures.append("aggregate_only_quality_source") return list(dict.fromkeys(failures)) def _quality_signal_fix_failures( decision: Any, packet: EvidencePacket, *, target_skill_ids: list[str], ) -> list[str]: if not _is_quality_signal_context(packet, None): return [] failures: list[str] = [] reason_tags = {tag.lower() for tag in _str_list(_attr(decision, "reason_tags"))} if reason_tags.intersection(_QUALITY_SIGNAL_EXTERNAL_ATTRIBUTIONS): failures.append("attribution_external_only") has_tool_event = bool(_refs(packet, "tool_event")) has_exact_tool_evidence = bool( _refs(packet, "tool_result") or _refs(packet, "tool_incident") ) if not has_tool_event or ( not has_exact_tool_evidence and not _has_actionable_partial_quality_signal( packet, target_skill_ids=target_skill_ids, ) ): failures.append("missing_representative_tool_evidence") target_set = {item for item in target_skill_ids if item} signal_skill_ids = _quality_signal_skill_ids(packet) if target_set and signal_skill_ids and not target_set.intersection(signal_skill_ids): failures.append("quality_signal_target_skill_mismatch") return list(dict.fromkeys(failures)) def _has_actionable_partial_quality_signal( packet: EvidencePacket, *, target_skill_ids: list[str], ) -> bool: failed_tool_event_ids = { ref.ref_id for ref in _refs(packet, "tool_event") if _is_failed_tool_event_ref(ref) } skill_event_ids = {ref.ref_id for ref in _refs(packet, "skill_event")} skill_file_ids = {ref.ref_id for ref in _refs(packet, "skill_file")} target_set = {item for item in target_skill_ids if item} for ref in _quality_signal_refs(packet): actionability = str(ref.metadata.get("actionability") or "").strip() evidence_status = str(ref.metadata.get("evidence_status") or "").strip() if actionability != "trigger_review" or evidence_status != STATUS_ACTIONABLE_PARTIAL: continue backrefs = {str(item) for item in (ref.raw_backrefs or []) if item} if not failed_tool_event_ids.intersection(backrefs): continue if not skill_event_ids.intersection(backrefs): continue if not skill_file_ids.intersection(backrefs): continue if target_set and not _quality_signal_ref_skill_ids(ref).intersection(target_set): continue return True return False def _is_failed_tool_event_ref(ref: ResourceRef) -> bool: if ref.ref_type != "tool_event": return False metadata = ref.metadata success = metadata.get("success") if success is False: return True if isinstance(success, str) and success.strip().lower() in {"0", "false", "no"}: return True status = str(metadata.get("status") or "").strip().lower() if status and status not in {"ok", "passed", "success", "succeeded"}: return True permission_status = str(metadata.get("permission_status") or "").strip().lower() return permission_status in {"blocked", "denied", "permission_denied", "rejected"} def _quality_signal_skill_ids(packet: EvidencePacket) -> set[str]: skill_ids: set[str] = set() for ref in _quality_signal_refs(packet): skill_ids.update(_quality_signal_ref_skill_ids(ref)) return {item for item in skill_ids if item} def _quality_signal_ref_skill_ids(ref: ResourceRef) -> set[str]: metadata = ref.metadata skill_ids = set(_metadata_values(metadata, "skill_id", "skill_ids")) subject_type = str(metadata.get("subject_type") or "") subject_id = str(metadata.get("subject_id") or "") if subject_type == "tool_skill_relation" and subject_id: skill_ids.add(subject_id.split(":", 1)[0]) return {item for item in skill_ids if item} def _skill_file_refs(packet: EvidencePacket, skill_ids: list[str]) -> list[ResourceRef]: target_set = set(skill_ids) return [ ref for ref in _refs(packet, "skill_file") if not target_set or _metadata_values(ref.metadata, "skill_id", "skill_ids").intersection(target_set) or any(skill_id in ref.ref_id for skill_id in target_set) ] def _skill_lifecycle_refs(packet: EvidencePacket, skill_ids: list[str]) -> list[ResourceRef]: target_set = set(skill_ids) allowed = {"selected", "invoked", "applied"} refs: list[ResourceRef] = [] for ref_type in ("skill_event", "skill_record"): for ref in _refs(packet, ref_type): values = _metadata_values(ref.metadata, "skill_id", "skill_ids") if target_set and not values.intersection(target_set) and not any( skill_id in ref.ref_id for skill_id in target_set ): continue lifecycle = str( ref.metadata.get("event_type") or ref.metadata.get("lifecycle_event") or ref.metadata.get("status") or "" ).strip().lower() nested = ref.metadata.get("metadata") if not lifecycle and isinstance(nested, Mapping): lifecycle = str( nested.get("event_type") or nested.get("lifecycle_event") or "" ).strip().lower() if lifecycle in allowed: refs.append(ref) return refs def _manual_fix_request_refs(packet: EvidencePacket) -> list[ResourceRef]: refs = list(_refs(packet, "manual_request_ref")) if not refs: return [] if ( str(getattr(packet, "profile_name", "") or "") == "manual_fix_or_derive" and str(getattr(packet, "subprofile", "") or "") == "fix" ): return refs selected: list[ResourceRef] = [] for ref in refs: action = str(ref.metadata.get("action") or "").strip().lower() if action in {"fix", "repair"}: selected.append(ref) return selected def _target_skill_state_refs( packet: EvidencePacket, skill_ids: list[str], ) -> list[ResourceRef]: target_set = set(skill_ids) refs: list[ResourceRef] = [] for ref in _refs(packet, "skill_record"): values = _metadata_values(ref.metadata, "skill_id", "skill_ids") if ( not target_set or values.intersection(target_set) or any(skill_id in ref.ref_id for skill_id in target_set) ): refs.append(ref) return refs def _failure_or_friction_refs(packet: EvidencePacket) -> list[ResourceRef]: refs: list[ResourceRef] = [] for ref in _refs(packet, "runtime_snapshot"): metadata = ref.metadata status = str(metadata.get("status") or "").strip().lower() stop_reason = str(metadata.get("stop_reason") or "").strip().lower() if status and status not in {"success", "completed", "ok", "passed"}: refs.append(ref) elif stop_reason in {"max_iterations", "error", "cancelled", "incomplete"}: refs.append(ref) refs.extend(_tool_failure_refs(packet)) for ref in _refs(packet, "skill_event"): lifecycle = str(ref.metadata.get("event_type") or ref.metadata.get("status") or "").lower() if lifecycle in {"fallback", "failed", "error", "permission_denied"}: refs.append(ref) for ref in _refs(packet, "transcript_message"): text = _ref_text(ref).lower() if any(term in text for term in ("correction", "not what i asked", "wrong", "failed")): refs.append(ref) return _dedupe_refs(refs) def _tool_failure_refs(packet: EvidencePacket) -> list[ResourceRef]: refs: list[ResourceRef] = [] for ref_type in ("tool_event", "tool_result", "tool_incident"): for ref in _refs(packet, ref_type): metadata = ref.metadata status = str( metadata.get("status") or metadata.get("outcome") or metadata.get("result") or "" ).strip().lower() success = metadata.get("success") if ref_type == "tool_incident": refs.append(ref) elif status and status not in {"success", "ok", "completed", "passed"}: refs.append(ref) elif success is False: refs.append(ref) elif ( metadata.get("error_type") or metadata.get("error_message") or metadata.get("error_bucket") or metadata.get("permission_status") in {"denied", "rejected"} ): refs.append(ref) return _dedupe_refs(refs) def _is_environment_failure(ref: ResourceRef) -> bool: return _contains_any(_ref_text(ref), _ENVIRONMENT_FAILURE_TERMS) def _environment_reason_tags(refs: list[ResourceRef]) -> list[str]: tags: list[str] = [] text = "\n".join(_ref_text(ref).lower() for ref in refs) if "api key" in text or "apikey" in text or "_api_key" in text: tags.append("api_key") if "network" in text or "dns" in text or "connection" in text: tags.append("network") if "sandbox" in text or "permission mode" in text: tags.append("sandbox_or_permission_mode") if "external service" in text or "service unavailable" in text: tags.append("external_service") return tags or ["environment"] def _fix_causality_plausible( decision: Any, packet: EvidencePacket, failure_refs: list[ResourceRef], ) -> bool: if not failure_refs: return False text = _decision_text(decision).lower() if any(term in text for term in ("no_causality", "unrelated_to_skill", "external_only")): return False if any(term in text for term in ("skill", "instruction", "step", "workflow", "fix", "repair")): return True return bool(_skill_lifecycle_refs(packet, _target_skill_ids(decision))) def _tool_quality_dependency_warning( packet: EvidencePacket, target_skill_ids: list[str], tool_refs: list[ResourceRef], ) -> str | None: if not tool_refs: return None tool_variants = _tool_identity_variants_from_refs(tool_refs) if not tool_variants: return "tool_dependency_uncertain" dependency_variants = _target_skill_dependency_variants(packet, target_skill_ids) if dependency_variants and dependency_variants.intersection(tool_variants): return None if dependency_variants: return "tool_dependency_mismatch" if _target_skill_text_mentions_tool(packet, target_skill_ids, tool_variants): return None return "tool_dependency_uncertain" def _target_skill_dependency_variants( packet: EvidencePacket, target_skill_ids: list[str], ) -> set[str]: variants: set[str] = set() for ref in _target_skill_state_refs(packet, target_skill_ids): for value in _metadata_tool_values(ref.metadata): variants.update(_tool_key_variants(value)) return variants def _metadata_tool_values(metadata: Mapping[str, Any]) -> set[str]: keys = { "tool_dependency", "tool_dependencies", "tool_key", "tool_keys", "critical_tool", "critical_tools", "allowed-tools", "allowed_tools", } values: set[str] = set() for key, value in metadata.items(): if key in keys: values.update(_split_tool_values(value)) if isinstance(value, Mapping): values.update(_metadata_tool_values(value)) return values def _split_tool_values(value: Any) -> set[str]: if value is None: return set() if isinstance(value, str): raw = value.strip() if not raw: return set() parts = { item.strip() for item in _split_tool_value_tokens(raw) if item.strip() } parts.add(raw) return parts if isinstance(value, Mapping): return _metadata_tool_values(value) if isinstance(value, (list, tuple, set)): values: set[str] = set() for item in value: values.update(_split_tool_values(item)) return values text = str(value).strip() return {text} if text else set() def _split_tool_value_tokens(value: str) -> list[str]: return re.split(r"[\s,;\n]+", value) def _tool_identity_variants_from_refs(refs: list[ResourceRef]) -> set[str]: variants: set[str] = set() for ref in refs: metadata = ref.metadata raw_key = metadata.get("tool_key") if raw_key: variants.update(_tool_key_variants(raw_key)) tool_name = metadata.get("tool_name") backend = metadata.get("backend") server = metadata.get("server") or metadata.get("server_name") if tool_name: variants.update(_tool_key_variants(tool_name)) if backend: variants.update( _tool_key_variants( f"{backend}:{server or 'default'}:{tool_name}" ) ) variants.update(_tool_key_variants(_tool_key_from_ref_id(ref.ref_id))) return {item for item in variants if item} def _tool_key_from_ref_id(ref_id: str) -> str: parts = str(ref_id or "").split(":") if len(parts) >= 4 and parts[0] in {"tool_quality_record", "tool_incident"}: return ":".join(parts[1:4]) return "" def _tool_key_variants(value: Any) -> set[str]: text = str(value or "").strip().lower() if not text: return set() variants = {text} parts = [part for part in text.split(":") if part] if len(parts) >= 3: backend, server, tool_name = parts[0], parts[1], parts[2] variants.add(f"{backend}:{server}:{tool_name}") variants.add(f"{backend}:{tool_name}") variants.add(tool_name) elif len(parts) == 2: backend, tool_name = parts variants.add(f"{backend}:default:{tool_name}") variants.add(f"{backend}:{tool_name}") variants.add(tool_name) elif len(parts) == 1: variants.add(parts[0]) return variants def _target_skill_text_mentions_tool( packet: EvidencePacket, target_skill_ids: list[str], tool_variants: set[str], ) -> bool: names = {variant.split(":")[-1] for variant in tool_variants if variant} names.update(tool_variants) names = {name for name in names if len(name) >= 3} if not names: return False text_parts: list[str] = [] for ref in _skill_file_refs(packet, target_skill_ids): text_parts.append(_ref_text(ref).lower()) for ref in _target_skill_state_refs(packet, target_skill_ids): text_parts.append(_ref_text(ref).lower()) text = "\n".join(text_parts) return any(name in text for name in names) def _task_completed(packet: EvidencePacket) -> bool: saw_status = False for ref in _refs(packet, "runtime_snapshot"): metadata = ref.metadata status = str(metadata.get("status") or "").strip().lower() final_response = str(metadata.get("final_response_preview") or "").strip() if status: saw_status = True if status in {"success", "completed", "ok", "passed"} and final_response: return True if status in {"error", "failed", "incomplete", "cancelled"}: return False if saw_status: return False for ref in _refs(packet, "transcript_message"): if str(ref.metadata.get("role") or "").lower() == "assistant" and ( ref.metadata.get("is_final_response") or ref.metadata.get("final_response") or ref.metadata.get("message_kind") in {"final", "final_response"} ): return bool(str(ref.preview or "").strip()) return False def _is_simple_tool_capture(packet: EvidencePacket) -> bool: tool_refs = _refs(packet, "tool_event") if len(tool_refs) != 1: return False ref = tool_refs[0] name = str(ref.metadata.get("tool_name") or "").strip().lower() input_preview = str(ref.metadata.get("input_preview") or ref.preview or "").strip().lower() command = input_preview.split(None, 1)[0] if input_preview else name if name in _SIMPLE_CAPTURE_TOOLS or command in _SIMPLE_CAPTURE_TOOLS: return True if name == "bash" and command in _SIMPLE_CAPTURE_TOOLS: return True return False def _nontrivial_workflow(packet: EvidencePacket, decision: Any) -> bool: if len(_refs(packet, "tool_event")) >= 2: return True if _refs(packet, "file_history"): return True return _contains_any(_decision_text(decision), _WORKFLOW_BOUNDARY_TERMS) def _reusable_boundary(decision: Any, packet: EvidencePacket) -> bool: if _contains_any(_decision_text(decision), _WORKFLOW_BOUNDARY_TERMS): return True return bool(_refs(packet, "manual_request_ref")) and len(_refs(packet, "tool_event")) >= 2 def _capture_depends_on_ephemeral_context( packet: EvidencePacket, decision: Any, ) -> bool: parts = [_decision_text(decision)] for ref_type in ( "runtime_snapshot", "transcript_message", "tool_event", "tool_result", "file_history", "manual_request_ref", ): parts.extend(_ref_text(ref) for ref in _refs(packet, ref_type)) return _contains_any("\n".join(parts), _EPHEMERAL_CAPTURE_TERMS) def _is_repeated_or_user_explicit( decision: Any, packet: EvidencePacket, *, recurrence_threshold: int = 2, ) -> bool: text = _decision_text(decision) if _contains_any(text, _USER_EXPLICIT_TERMS): return True if _refs(packet, "manual_request_ref"): return True recurrence = str(_attr(decision, "recurrence") or "").lower() if recurrence in {"repeated", "user_explicit"}: return True tags = " ".join(_str_list(_attr(decision, "reason_tags"))).lower() if any(term in tags for term in ("repeated", "recurrence", "min_observations")): return True threshold = max(2, int(recurrence_threshold or 2)) scope = getattr(packet, "scope", None) source_task_ids = getattr(scope, "source_task_ids", ()) if scope is not None else () if len([item for item in source_task_ids if item]) >= threshold: return True representative_ids = ( getattr(scope, "representative_execution_ids", ()) if scope is not None else () ) return len([item for item in representative_ids if item]) >= threshold def _existing_skill_covers(decision: Any) -> bool: text = _decision_text(decision).lower() return "existing_skill_covers" in text or "covered_by_existing_skill" in text def _ref_types_for_ids(packet: EvidencePacket, ref_ids: list[str]) -> set[str]: wanted = set(ref_ids) return { ref.ref_type for group in packet.selected_refs.values() for ref in group if ref.ref_id in wanted } def _dedupe_refs(refs: list[ResourceRef]) -> list[ResourceRef]: seen: set[str] = set() result: list[ResourceRef] = [] for ref in refs: key = ref.ref_id if key in seen: continue seen.add(key) result.append(ref) return result def _decision_text(decision: Any) -> str: parts: list[str] = [] for name in ("reason_summary", "noop_reason"): value = _attr(decision, name) if value: parts.append(str(value)) parts.extend(_str_list(_attr(decision, "reason_tags"))) parts.extend(_str_list(_attr(decision, "risks"))) for claim in list(_attr(decision, "evidence_claims") or []): value = _attr(claim, "claim") if value: parts.append(str(value)) return "\n".join(parts) def _contains_any(text: str, terms: tuple[str, ...]) -> bool: normalized = str(text or "").lower() return any(term in normalized for term in terms) def _ref_text(ref: ResourceRef) -> str: return "\n".join( [ str(ref.preview or ""), json.dumps(ref.metadata, sort_keys=True, default=str), ] ) def _metadata_values(metadata: Mapping[str, Any], *keys: str) -> set[str]: values: set[str] = set() for key in keys: value = metadata.get(key) if isinstance(value, str) and value: values.add(value) elif isinstance(value, (list, tuple, set)): values.update(str(item) for item in value if str(item)) nested = metadata.get("metadata") if isinstance(nested, Mapping): values.update(_metadata_values(nested, *keys)) return values def _attr(value: Any, name: str) -> Any: if isinstance(value, Mapping): return value.get(name) return getattr(value, name, None) def _none_or_str(value: Any) -> str | None: if value is None: return None text = str(value) return text if text else None def _str_list(value: Any) -> list[str]: if value is None: return [] if isinstance(value, str): return [value] if value else [] if isinstance(value, (list, tuple, set)): return [str(item) for item in value if str(item)] return [] def _digest(payload: Any) -> str: return hashlib.sha256( json.dumps(payload, sort_keys=True, default=str).encode("utf-8") ).hexdigest() def _utc_now() -> str: return datetime.now(timezone.utc).isoformat()