diff --git a/src/ofw/__init__.py b/src/ofw/__init__.py index bda3cea..1a515e3 100644 --- a/src/ofw/__init__.py +++ b/src/ofw/__init__.py @@ -26,6 +26,21 @@ Sha256Digest, WorkspaceFile, ) +from ofw.diagnosis import ( + ClusterRevisionRef, + ClusterState, + DiagnosisError, + DiagnosisErrorCode, + DiagnosisResult, + DiagnosisRun, + EvidenceAnchor, + EvidenceAnchorKind, + FailureCluster, + MechanismKey, + PythonDiagnoser, + Severity, + TraceDiagnosis, +) from ofw.harness import EditableFile, Harness, Subagent, Tool, editable from ofw.mine import ( Mine, @@ -109,6 +124,7 @@ class _OfwNamespace: ServiceName = ServiceName MiningPolicy = MiningPolicy ScoreName = ScoreName + PythonDiagnoser = PythonDiagnoser def editable(self, path: Path) -> EditableFile: return editable(path) @@ -158,6 +174,8 @@ def read_snapshot_content( "AssetAccess", "CanaryCase", "CaseId", + "ClusterState", + "ClusterRevisionRef", "ComponentKind", "CollectionError", "CollectionErrorCode", @@ -166,7 +184,14 @@ def read_snapshot_content( "CommandVerifier", "ContentCaptureMode", "DockerCompose", + "DiagnosisResult", + "DiagnosisRun", + "DiagnosisError", + "DiagnosisErrorCode", "EditableFile", + "EvidenceAnchor", + "EvidenceAnchorKind", + "FailureCluster", "GitCommit", "Harness", "HarnessAsset", @@ -187,6 +212,7 @@ def read_snapshot_content( "MineError", "MineErrorCode", "MiningPolicy", + "MechanismKey", "ObservationContent", "ObservationContentField", "ObservationContentHit", @@ -199,6 +225,7 @@ def read_snapshot_content( "ProcessLimits", "PythonEntrypoint", "PythonLoop", + "PythonDiagnoser", "PythonVerifier", "RunErrorCode", "RunResult", @@ -206,6 +233,7 @@ def read_snapshot_content( "ScoreName", "Sha256Digest", "ServiceName", + "Severity", "SecretEnvironmentVariable", "SnapshotContentReference", "Subagent", @@ -213,6 +241,7 @@ def read_snapshot_content( "TraceWindow", "TracePartition", "TraceQualityThreshold", + "TraceDiagnosis", "VerifierResult", "VerifierVerdict", "WorkspaceFile", diff --git a/src/ofw/_diagnosis_runner.py b/src/ofw/_diagnosis_runner.py new file mode 100644 index 0000000..d808425 --- /dev/null +++ b/src/ofw/_diagnosis_runner.py @@ -0,0 +1,46 @@ +"""Timed child-process entrypoint for a file-backed trace diagnoser.""" + +from __future__ import annotations + +import importlib +import inspect +import sys +from collections.abc import Callable +from typing import cast + +from pydantic import TypeAdapter, ValidationError + +from ofw.diagnosis import TraceDiagnosis +from ofw.mine import TraceSnapshot + +_SNAPSHOT_ADAPTER: TypeAdapter[TraceSnapshot] = TypeAdapter(TraceSnapshot) +_DIAGNOSIS_ADAPTER: TypeAdapter[TraceDiagnosis] = TypeAdapter(TraceDiagnosis) + + +def main() -> int: + if len(sys.argv) != 3: + return 2 + payload: str = sys.stdin.read() + try: + snapshot = _SNAPSHOT_ADAPTER.validate_json(payload) + except ValidationError: + return 2 + module = importlib.import_module(sys.argv[1]) + functions = tuple( + function + for name, function in inspect.getmembers(module, inspect.isfunction) + if name == sys.argv[2] + ) + if len(functions) != 1: + return 2 + function: Callable[[TraceSnapshot], TraceDiagnosis] = cast( + Callable[[TraceSnapshot], TraceDiagnosis], + functions[0], + ) + diagnosis: TraceDiagnosis = function(snapshot) + sys.stdout.write(_DIAGNOSIS_ADAPTER.dump_json(diagnosis).decode()) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/ofw/diagnosis.py b/src/ofw/diagnosis.py new file mode 100644 index 0000000..60ecb76 --- /dev/null +++ b/src/ofw/diagnosis.py @@ -0,0 +1,539 @@ +"""Evidence-bound failure diagnosis and deterministic cluster revisions.""" + +from __future__ import annotations + +import hashlib +import math +import sys +from dataclasses import dataclass +from datetime import datetime +from enum import IntEnum, StrEnum +from pathlib import Path + +from pydantic import TypeAdapter, ValidationError + +from ofw.contracts import ComponentKind, HarnessRevision, HarnessRevisionId, Sha256Digest +from ofw.harness import Harness +from ofw.mine import ( + MineResult, + MineRunId, + TraceAdmission, + TracePartition, + TraceSnapshot, + digest_bytes, + write_artifact, +) +from ofw.observability.langfuse.domain import TraceId +from ofw.runtime import ( + CanaryCase, + CaseId, + LocalProcess, + ModelFingerprint, + PreparedEnvironment, + ProcessCommand, + ProcessLimits, + PythonEntrypoint, + resolve_python_source, +) + + +class DiagnosisSchemaVersion(IntEnum): + V1 = 1 + + +class EvidenceAnchorKind(StrEnum): + OBSERVATION = "observation" + SCORE = "score" + + +class Severity(StrEnum): + LOW = "low" + MEDIUM = "medium" + HIGH = "high" + CRITICAL = "critical" + + +class ClusterState(StrEnum): + PROPOSED = "proposed" + CONFIRMED = "confirmed" + TARGETED = "targeted" + RESOLVED = "resolved" + REOPENED = "reopened" + + +class DiagnosisStatus(StrEnum): + PROPOSED = "proposed" + ABSTAINED = "abstained" + + +class DiagnosisErrorCode(StrEnum): + STALE_HARNESS = "stale_harness" + REVISION_MISMATCH = "revision_mismatch" + ARTIFACT_INVALID = "artifact_invalid" + + +class DiagnosisError(Exception): + __slots__ = ("code", "subject") + + def __init__(self, code: DiagnosisErrorCode, subject: str) -> None: + self.code = code + self.subject = subject + super().__init__(f"{code.value}: {subject}") + + +@dataclass(frozen=True, slots=True) +class MechanismKey: + value: str + + def __post_init__(self) -> None: + if not self.value or "\0" in self.value: + raise ValueError("invalid mechanism key") + + +@dataclass(frozen=True, slots=True) +class EvidenceAnchor: + kind: EvidenceAnchorKind + id: str + + def __post_init__(self) -> None: + if not self.id: + raise ValueError("evidence id is required") + + +@dataclass(frozen=True, slots=True) +class TraceDiagnosis: + trace_id: TraceId + status: DiagnosisStatus + mechanism: MechanismKey | None + title: str + description: str + evidence: tuple[EvidenceAnchor, ...] + components: tuple[ComponentKind, ...] + severity: Severity | None + confidence: float | None + + @classmethod + def proposed( + cls, + trace_id: TraceId, + mechanism: MechanismKey, + title: str, + description: str, + evidence: tuple[EvidenceAnchor, ...], + components: tuple[ComponentKind, ...], + severity: Severity, + confidence: float, + ) -> TraceDiagnosis: + if ( + not title + or not description + or not evidence + or not components + or not math.isfinite(confidence) + or confidence < 0 + or confidence > 1 + ): + raise ValueError("invalid diagnosis") + return cls( + trace_id, + DiagnosisStatus.PROPOSED, + mechanism, + title, + description, + evidence, + components, + severity, + confidence, + ) + + @classmethod + def abstained(cls, trace_id: TraceId) -> TraceDiagnosis: + return cls(trace_id, DiagnosisStatus.ABSTAINED, None, "", "", (), (), None, None) + + +@dataclass(frozen=True, slots=True) +class PythonDiagnoser: + entrypoint: PythonEntrypoint + limits: ProcessLimits + model: ModelFingerprint | None = None + + def fingerprint(self, root: Path) -> Sha256Digest: + source = resolve_python_source(root, self.entrypoint) + model = ( + "none" + if self.model is None + else f"{self.model.provider}:{self.model.model}:{self.model.reasoning}" + ) + return _digest_text( + f"{self.entrypoint.module.value}\0{self.entrypoint.function.value}\0" + f"{digest_bytes(source.read_bytes())}\0{self.limits.timeout.total_seconds()}\0{model}" + ) + + def diagnose( + self, + snapshot: TraceSnapshot, + prepared: PreparedEnvironment, + ) -> TraceDiagnosis: + command = ProcessCommand( + ( + sys.executable, + "-m", + "ofw._diagnosis_runner", + self.entrypoint.module.value, + self.entrypoint.function.value, + ) + ) + process = prepared.run(command, _SNAPSHOT_ADAPTER.dump_json(snapshot).decode()) + if process.timed_out or process.exit_code != 0: + return TraceDiagnosis.abstained(snapshot.trace.id) + try: + return _DIAGNOSIS_ADAPTER.validate_json(process.stdout) + except ValidationError: + return TraceDiagnosis.abstained(snapshot.trace.id) + + +@dataclass(frozen=True, slots=True) +class ClusterId: + value: str + + def __str__(self) -> str: + return self.value + + +@dataclass(frozen=True, slots=True) +class ClusterRevisionRef: + id: ClusterId + revision: int + + +@dataclass(frozen=True, slots=True) +class FailureCluster: + id: ClusterId + revision: int + content_digest: Sha256Digest + mechanism: MechanismKey + title: str + description: str + source_trace_ids: tuple[TraceId, ...] + evidence: tuple[EvidenceAnchor, ...] + components: tuple[ComponentKind, ...] + recurrence: int + severity: Severity + confidence: float + resolution_rate: float + state: ClusterState + parents: tuple[ClusterRevisionRef, ...] = () + + +@dataclass(frozen=True, slots=True) +class DiagnosisRunId: + value: str + + def __str__(self) -> str: + return self.value + + +@dataclass(frozen=True, slots=True) +class DiagnosisResult: + schema_version: DiagnosisSchemaVersion + id: DiagnosisRunId + mine_id: MineRunId + revision_id: HarnessRevisionId + diagnoser_digest: Sha256Digest + source_watermark: datetime + diagnoses: tuple[TraceDiagnosis, ...] + clusters: tuple[FailureCluster, ...] + root: Path + + @property + def abstained_count(self) -> int: + return sum(diagnosis.status is DiagnosisStatus.ABSTAINED for diagnosis in self.diagnoses) + + @property + def manifest_path(self) -> Path: + return self.root / ".ofw" / "mine" / str(self.mine_id) / "diagnosis" / f"{self.id}.json" + + def to_json(self) -> str: + return _RESULT_ADAPTER.dump_json(self).decode() + + +_SNAPSHOT_ADAPTER: TypeAdapter[TraceSnapshot] = TypeAdapter(TraceSnapshot) +_DIAGNOSIS_ADAPTER: TypeAdapter[TraceDiagnosis] = TypeAdapter(TraceDiagnosis) +_DIAGNOSES_ADAPTER: TypeAdapter[tuple[TraceDiagnosis, ...]] = TypeAdapter( + tuple[TraceDiagnosis, ...] +) +_RESULT_ADAPTER: TypeAdapter[DiagnosisResult] = TypeAdapter(DiagnosisResult) + + +@dataclass(frozen=True, slots=True) +class DiagnosisRun: + source: Harness | HarnessRevision + mine: MineResult + diagnoser: PythonDiagnoser + previous: DiagnosisResult | None = None + + def run(self) -> DiagnosisResult: + revision = _resolve_revision(self.source) + if revision.id != self.mine.revision_id: + raise DiagnosisError(DiagnosisErrorCode.REVISION_MISMATCH, str(revision.id)) + diagnoser_digest = self.diagnoser.fingerprint(revision.root) + failures = tuple( + admission + for admission in self.mine.admissions + if admission.partition is TracePartition.VERIFIED_FAILURE + and admission.snapshot_path is not None + ) + environment = LocalProcess(self.diagnoser.limits) + prepared = environment.prepare(revision, CanaryCase(CaseId("diagnosis"), "")) + try: + diagnoses = tuple( + self._diagnose(_read_snapshot(admission, self.mine), prepared) + for admission in failures + if admission.snapshot_path is not None + ) + finally: + environment.destroy(prepared) + diagnoses_digest = digest_bytes(_DIAGNOSES_ADAPTER.dump_json(diagnoses)) + run_id = DiagnosisRunId( + "diagnosis_" + + hashlib.sha256( + f"{self.mine.id}\0{diagnoser_digest}\0{diagnoses_digest}\0" + f"{'' if self.previous is None else self.previous.id}\0" + f"{int(DiagnosisSchemaVersion.V1)}".encode() + ).hexdigest() + ) + clusters = _clusters(diagnoses, diagnoser_digest, self.previous) + result = DiagnosisResult( + DiagnosisSchemaVersion.V1, + run_id, + self.mine.id, + revision.id, + diagnoser_digest, + self.mine.window.end, + diagnoses, + clusters, + revision.root, + ) + write_artifact(result.manifest_path, f"{result.to_json()}\n".encode()) + return result + + def _diagnose( + self, + snapshot: TraceSnapshot, + prepared: PreparedEnvironment, + ) -> TraceDiagnosis: + diagnosis = self.diagnoser.diagnose(snapshot, prepared) + if ( + diagnosis.trace_id != snapshot.trace.id + or not _diagnosis_valid(diagnosis) + or not _anchors_exist(diagnosis, snapshot) + ): + return TraceDiagnosis.abstained(snapshot.trace.id) + return diagnosis + + +def _clusters( + diagnoses: tuple[TraceDiagnosis, ...], + diagnoser_digest: Sha256Digest, + previous: DiagnosisResult | None, +) -> tuple[FailureCluster, ...]: + current_mechanisms = { + diagnosis.mechanism + for diagnosis in diagnoses + if diagnosis.status is DiagnosisStatus.PROPOSED and diagnosis.mechanism is not None + } + previous_mechanisms = ( + set() if previous is None else {cluster.mechanism for cluster in previous.clusters} + ) + mechanisms = tuple(sorted(current_mechanisms | previous_mechanisms, key=_mechanism_sort_key)) + return tuple( + _cluster_revision( + mechanism, + tuple(diagnosis for diagnosis in diagnoses if diagnosis.mechanism == mechanism), + diagnoser_digest, + _previous_cluster(previous, mechanism), + ) + for mechanism in mechanisms + ) + + +def _cluster_revision( + mechanism: MechanismKey, + diagnoses: tuple[TraceDiagnosis, ...], + diagnoser_digest: Sha256Digest, + previous: FailureCluster | None, +) -> FailureCluster: + if not diagnoses: + if previous is None: + raise ValueError("cluster requires diagnosis or prior revision") + return _resolved_cluster(previous) + first = diagnoses[0] + evidence = tuple(anchor for diagnosis in diagnoses for anchor in diagnosis.evidence) + components = tuple( + sorted( + {component for diagnosis in diagnoses for component in diagnosis.components}, + key=_component_sort_key, + ) + ) + severity = max( + (diagnosis.severity for diagnosis in diagnoses if diagnosis.severity is not None), + key=_severity_rank, + ) + confidence = sum(diagnosis.confidence or 0 for diagnosis in diagnoses) / len(diagnoses) + trace_ids = tuple(diagnosis.trace_id for diagnosis in diagnoses) + diagnoses_digest = digest_bytes(_DIAGNOSES_ADAPTER.dump_json(diagnoses)) + content_digest = _digest_text( + f"{diagnoser_digest}\0{diagnoses_digest}\0" + + "\0".join(trace_id.value for trace_id in trace_ids) + ) + if previous is not None and previous.content_digest == content_digest: + return previous + revision = 1 if previous is None else previous.revision + 1 + parents = () if previous is None else (ClusterRevisionRef(previous.id, previous.revision),) + removed = ( + 0 + if previous is None + else sum(trace_id not in trace_ids for trace_id in previous.source_trace_ids) + ) + resolution_rate = ( + 0.0 if previous is None or not previous.source_trace_ids else removed / previous.recurrence + ) + state = ( + ClusterState.PROPOSED + if previous is None + else (ClusterState.REOPENED if previous.state is ClusterState.RESOLVED else previous.state) + ) + return FailureCluster( + _cluster_id(mechanism), + revision, + content_digest, + mechanism, + first.title, + first.description, + trace_ids, + evidence, + components, + len(diagnoses), + severity, + confidence, + resolution_rate, + state, + parents, + ) + + +def _resolved_cluster(previous: FailureCluster) -> FailureCluster: + if previous.state is ClusterState.RESOLVED: + return previous + return FailureCluster( + previous.id, + previous.revision + 1, + _digest_text(f"resolved\0{previous.content_digest}"), + previous.mechanism, + previous.title, + previous.description, + (), + (), + previous.components, + 0, + previous.severity, + previous.confidence, + 1.0, + ClusterState.RESOLVED, + (ClusterRevisionRef(previous.id, previous.revision),), + ) + + +def _cluster_id(mechanism: MechanismKey) -> ClusterId: + return ClusterId("cluster_" + hashlib.sha256(mechanism.value.encode()).hexdigest()) + + +def _previous_cluster( + previous: DiagnosisResult | None, + mechanism: MechanismKey, +) -> FailureCluster | None: + if previous is None: + return None + return next( + (cluster for cluster in previous.clusters if cluster.mechanism == mechanism), + None, + ) + + +def _anchors_exist(diagnosis: TraceDiagnosis, snapshot: TraceSnapshot) -> bool: + if diagnosis.status is DiagnosisStatus.ABSTAINED: + return True + return bool(diagnosis.evidence) and all( + _anchor_exists(anchor, snapshot) for anchor in diagnosis.evidence + ) + + +def _diagnosis_valid(diagnosis: TraceDiagnosis) -> bool: + if diagnosis.status is DiagnosisStatus.ABSTAINED: + return diagnosis.mechanism is None + return ( + diagnosis.mechanism is not None + and bool(diagnosis.title) + and bool(diagnosis.description) + and bool(diagnosis.evidence) + and bool(diagnosis.components) + and diagnosis.severity is not None + and diagnosis.confidence is not None + and math.isfinite(diagnosis.confidence) + and 0 <= diagnosis.confidence <= 1 + ) + + +def _anchor_exists(anchor: EvidenceAnchor, snapshot: TraceSnapshot) -> bool: + if anchor.kind is EvidenceAnchorKind.OBSERVATION: + return any(observation.id.value == anchor.id for observation in snapshot.observations) + return any(score.id.value == anchor.id for score in snapshot.scores) + + +def _read_snapshot(admission: TraceAdmission, mine: MineResult) -> TraceSnapshot: + path = admission.snapshot_path + digest = admission.snapshot_digest + if path is None or digest is None: + raise DiagnosisError(DiagnosisErrorCode.ARTIFACT_INVALID, admission.trace_id.value) + try: + root = (mine.root / ".ofw" / "mine" / str(mine.id) / "traces").resolve(strict=True) + resolved = path.resolve(strict=True) + resolved.relative_to(root) + payload = resolved.read_bytes() + if digest_bytes(payload) != digest: + raise DiagnosisError(DiagnosisErrorCode.ARTIFACT_INVALID, str(path)) + snapshot = _SNAPSHOT_ADAPTER.validate_json(payload) + if ( + snapshot.trace.id != admission.trace_id + or snapshot.revision_id != mine.revision_id + or snapshot.collection_digest != mine.collection_digest + ): + raise DiagnosisError(DiagnosisErrorCode.ARTIFACT_INVALID, str(path)) + return snapshot + except (OSError, ValueError, ValidationError) as error: + raise DiagnosisError(DiagnosisErrorCode.ARTIFACT_INVALID, str(path)) from error + + +def _resolve_revision(source: Harness | HarnessRevision) -> HarnessRevision: + if isinstance(source, HarnessRevision): + return source + revision = source.current_revision + if revision is None: + raise DiagnosisError(DiagnosisErrorCode.STALE_HARNESS, source.name) + return revision + + +def _severity_rank(severity: Severity) -> int: + return (Severity.LOW, Severity.MEDIUM, Severity.HIGH, Severity.CRITICAL).index(severity) + + +def _mechanism_sort_key(mechanism: MechanismKey) -> str: + return mechanism.value + + +def _component_sort_key(component: ComponentKind) -> str: + return component.value + + +def _digest_text(value: str) -> Sha256Digest: + return digest_bytes(value.encode()) diff --git a/src/ofw/mine.py b/src/ofw/mine.py index 5a3b080..07bbbc5 100644 --- a/src/ofw/mine.py +++ b/src/ofw/mine.py @@ -293,7 +293,7 @@ def run(self) -> MineResult: admissions, revision.root, ) - _write_artifact(result.manifest_path, f"{result.to_json()}\n".encode()) + write_artifact(result.manifest_path, f"{result.to_json()}\n".encode()) return result def _admit( @@ -330,9 +330,9 @@ def _admit( tuple(_snapshot_score(score) for score in snapshot_scores), ) payload = _SNAPSHOT_ADAPTER.dump_json(snapshot) - digest = _digest_bytes(payload) + digest = digest_bytes(payload) path = revision.root / ".ofw" / "mine" / str(run_id) / "traces" / f"{digest.value[7:]}.json" - _write_artifact(path, payload + b"\n") + write_artifact(path, payload + b"\n") return TraceAdmission(trace.id, partition, reason, evidence, digest, path) def _snapshot_observation( @@ -369,7 +369,7 @@ def _snapshot_content( / "content" / f"{reference.digest.value[7:]}.txt" ) - _write_artifact(path, content.text.encode()) + write_artifact(path, content.text.encode()) return SnapshotContentReference(reference, path) def _classify( @@ -517,7 +517,7 @@ def _resolve_revision(source: Harness | HarnessRevision) -> HarnessRevision: return revision -def _write_artifact(path: Path, payload: bytes) -> None: +def write_artifact(path: Path, payload: bytes) -> None: try: path.parent.mkdir(parents=True, exist_ok=True) descriptor, temporary_name = tempfile.mkstemp( @@ -540,10 +540,10 @@ def _write_artifact(path: Path, payload: bytes) -> None: def _digest_text(value: str) -> Sha256Digest: - return _digest_bytes(value.encode()) + return digest_bytes(value.encode()) -def _digest_bytes(value: bytes) -> Sha256Digest: +def digest_bytes(value: bytes) -> Sha256Digest: return Sha256Digest(f"sha256:{hashlib.sha256(value).hexdigest()}") diff --git a/src/ofw/runtime.py b/src/ofw/runtime.py index e5fe1f1..088b896 100644 --- a/src/ofw/runtime.py +++ b/src/ofw/runtime.py @@ -322,7 +322,7 @@ class PythonLoop: models: tuple[ModelFingerprint, ...] = () def fingerprint(self, root: Path) -> Sha256Digest: - module_path = _python_source(root, self.entrypoint) + module_path = resolve_python_source(root, self.entrypoint) return _digest_text( f"python-loop\0{self.entrypoint.module.value}\0" f"{self.entrypoint.function.value}\0{_digest_file(module_path)}\0" @@ -361,7 +361,7 @@ def __post_init__(self) -> None: raise ValueError("invalid verifier name") def fingerprint(self, root: Path) -> Sha256Digest: - source = _python_source(root, self.entrypoint) + source = resolve_python_source(root, self.entrypoint) return _digest_text( f"python-verifier\0{self.name}\0{self.entrypoint.module.value}\0" f"{self.entrypoint.function.value}\0{_digest_file(source)}" @@ -594,7 +594,7 @@ def _command_fingerprint( return _digest_text("\0".join((kind, *command.arguments, *sources, _models_text(models)))) -def _python_source(root: Path, entrypoint: PythonEntrypoint) -> Path: +def resolve_python_source(root: Path, entrypoint: PythonEntrypoint) -> Path: path = root / Path(*entrypoint.module.value.split(".")).with_suffix(".py") if not path.is_file(): raise ValueError("python module is missing") diff --git a/tests/test_diagnosis.py b/tests/test_diagnosis.py new file mode 100644 index 0000000..05a5c1a --- /dev/null +++ b/tests/test_diagnosis.py @@ -0,0 +1,337 @@ +"""Evidence-bound failure diagnosis and deterministic clustering.""" + +from __future__ import annotations + +import hashlib +import subprocess +import time +from dataclasses import replace +from datetime import UTC, datetime, timedelta +from pathlib import Path + +import pytest +from pydantic import TypeAdapter + +from ofw import ( + ClusterRevisionRef, + ClusterState, + DiagnosisError, + DiagnosisErrorCode, + DiagnosisRun, + FunctionName, + Harness, + MechanismKey, + ModuleName, + ProcessLimits, + PythonDiagnoser, + PythonEntrypoint, +) +from ofw.contracts import HarnessRevisionId, Sha256Digest +from ofw.mine import ( + AdmissionReason, + MineResult, + MineRunId, + MineSchemaVersion, + SnapshotObservation, + SnapshotTrace, + TraceAdmission, + TracePartition, + TraceSnapshot, +) +from ofw.observability.langfuse.contracts import TraceWindow +from ofw.observability.langfuse.domain import ( + AttributionLevel, + ObservationId, + ObservationType, + TraceId, +) + +_SNAPSHOT_ADAPTER: TypeAdapter[TraceSnapshot] = TypeAdapter(TraceSnapshot) + + +def _run_git(root: Path, *arguments: str) -> None: + subprocess.run( + ("git", "-C", str(root), *arguments), + check=True, + capture_output=True, + text=True, + ) + + +def _repository(tmp_path: Path) -> tuple[Path, HarnessRevisionId]: + root = tmp_path / "diagnosis-agent" + root.mkdir() + (root / "prompt.md").write_text("Be accurate.\n", encoding="utf-8") + (root / "diagnoser.py").write_text( + "from __future__ import annotations\n" + "import time\n" + "from ofw import (ComponentKind, EvidenceAnchor, EvidenceAnchorKind, MechanismKey, " + "Severity, TraceDiagnosis)\n" + "from ofw.mine import TraceSnapshot\n" + "def diagnose(snapshot: TraceSnapshot) -> TraceDiagnosis:\n" + " observation = snapshot.observations[0]\n" + " if observation.name == 'abstain':\n" + " return TraceDiagnosis.abstained(snapshot.trace.id)\n" + " mechanism = 'tool-schema' if observation.name == 'tool' else 'prompt-gap'\n" + " component = (\n" + " ComponentKind.TOOL if observation.name == 'tool' else ComponentKind.PROMPT\n" + " )\n" + " return TraceDiagnosis.proposed(\n" + " snapshot.trace.id, MechanismKey(mechanism), mechanism, 'fixture diagnosis',\n" + " (EvidenceAnchor(EvidenceAnchorKind.OBSERVATION, observation.id.value),),\n" + " (component,), Severity.HIGH, 0.9,\n" + " )\n" + "def invalid_anchor(snapshot: TraceSnapshot) -> TraceDiagnosis:\n" + " return TraceDiagnosis.proposed(\n" + " snapshot.trace.id, MechanismKey('invalid-anchor'), 'invalid', 'invalid',\n" + " (EvidenceAnchor(EvidenceAnchorKind.OBSERVATION, 'missing'),),\n" + " (ComponentKind.PROMPT,), Severity.LOW, 0.5,\n" + " )\n" + "def slow(snapshot: TraceSnapshot) -> TraceDiagnosis:\n" + " del snapshot\n" + " time.sleep(2)\n" + " raise RuntimeError('late')\n", + encoding="utf-8", + ) + _run_git(root, "init", "-q") + _run_git(root, "config", "user.email", "fixture@example.test") + _run_git(root, "config", "user.name", "FixtureCo") + _run_git(root, "add", ".") + _run_git(root, "commit", "-qm", "fixture baseline") + harness = Harness("diagnosis-agent", root=root) + harness.connect_prompt(Path("prompt.md")) + return root, harness.process().id + + +def _snapshot( + root: Path, + revision_id: HarnessRevisionId, + mine_id: MineRunId, + trace: str, + name: str, +) -> tuple[Path, Sha256Digest]: + observation_id = ObservationId(f"observation-{trace}") + snapshot = TraceSnapshot( + MineSchemaVersion.V1, + revision_id, + Sha256Digest("sha256:collection"), + SnapshotTrace( + TraceId(trace), + (observation_id,), + (observation_id,), + (), + AttributionLevel.EXACT, + (), + Sha256Digest(f"sha256:trace-{trace}"), + ), + ( + SnapshotObservation( + observation_id, + TraceId(trace), + datetime(2026, 8, 22, tzinfo=UTC), + datetime(2026, 8, 22, 0, 1, tzinfo=UTC), + None, + ObservationType.AGENT, + True, + name, + None, + None, + Sha256Digest(f"sha256:observation-{trace}"), + ), + ), + (), + ) + payload = _SNAPSHOT_ADAPTER.dump_json(snapshot) + digest = Sha256Digest(f"sha256:{hashlib.sha256(payload).hexdigest()}") + path = root / ".ofw" / "mine" / str(mine_id) / "traces" / f"{digest.value[7:]}.json" + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(payload) + return path, digest + + +def _admission( + root: Path, + revision_id: HarnessRevisionId, + mine_id: MineRunId, + trace: str, + name: str, + partition: TracePartition, +) -> TraceAdmission: + path, digest = _snapshot(root, revision_id, mine_id, trace, name) + return TraceAdmission( + TraceId(trace), + partition, + ( + AdmissionReason.VERIFIED_FAIL + if partition is TracePartition.VERIFIED_FAILURE + else AdmissionReason.VERIFIED_PASS + ), + (), + digest, + path, + ) + + +def _mine_result(tmp_path: Path) -> tuple[MineResult, Harness]: + root, revision_id = _repository(tmp_path) + cases = ( + ("tool-one", "tool", TracePartition.VERIFIED_FAILURE), + ("tool-two", "tool", TracePartition.VERIFIED_FAILURE), + ("prompt-one", "prompt", TracePartition.VERIFIED_FAILURE), + ("unknown", "abstain", TracePartition.VERIFIED_FAILURE), + ("good", "tool", TracePartition.VERIFIED_GOOD), + ) + mine_id = MineRunId("mine_fixture") + admissions = tuple( + _admission(root, revision_id, mine_id, trace, name, partition) + for trace, name, partition in cases + ) + start = datetime(2026, 8, 22, tzinfo=UTC) + result = MineResult( + MineSchemaVersion.V1, + mine_id, + revision_id, + TraceWindow(start, start + timedelta(hours=1)), + Sha256Digest("sha256:collection"), + Sha256Digest("sha256:policy"), + admissions, + root, + ) + harness = Harness("diagnosis-agent", root=root) + harness.connect_prompt(Path("prompt.md")) + harness.process() + return result, harness + + +def _diagnoser(function: str, timeout: timedelta = timedelta(seconds=1)) -> PythonDiagnoser: + return PythonDiagnoser( + PythonEntrypoint(ModuleName("diagnoser"), FunctionName(function)), + ProcessLimits(timeout), + ) + + +def test_verified_failures_form_evidence_bound_mechanism_clusters(tmp_path: Path) -> None: + mine, harness = _mine_result(tmp_path) + + result = DiagnosisRun(harness, mine, _diagnoser("diagnose")).run() + + assert tuple(cluster.mechanism for cluster in result.clusters) == ( + MechanismKey("prompt-gap"), + MechanismKey("tool-schema"), + ) + assert tuple(cluster.recurrence for cluster in result.clusters) == (1, 2) + assert all(cluster.state is ClusterState.PROPOSED for cluster in result.clusters) + assert all(cluster.evidence for cluster in result.clusters) + assert result.abstained_count == 1 + assert TraceId("good") not in tuple(diagnosis.trace_id for diagnosis in result.diagnoses) + + +def test_invalid_evidence_anchor_becomes_abstention(tmp_path: Path) -> None: + mine, harness = _mine_result(tmp_path) + + result = DiagnosisRun(harness, mine, _diagnoser("invalid_anchor")).run() + + assert not result.clusters + assert result.abstained_count == 4 + + +def test_diagnoser_timeout_is_bounded_and_abstains(tmp_path: Path) -> None: + mine, harness = _mine_result(tmp_path) + started = time.monotonic() + + result = DiagnosisRun( + harness, + mine, + _diagnoser("slow", timedelta(milliseconds=50)), + ).run() + + assert result.abstained_count == 4 + assert time.monotonic() - started < 1 + + +def test_diagnosis_run_is_deterministic(tmp_path: Path) -> None: + mine, harness = _mine_result(tmp_path) + run = DiagnosisRun(harness, mine, _diagnoser("diagnose")) + + assert run.run() == run.run() + + +def test_tampered_snapshot_is_rejected_before_diagnosis(tmp_path: Path) -> None: + mine, harness = _mine_result(tmp_path) + snapshot_path = mine.admissions[0].snapshot_path + assert snapshot_path is not None + snapshot_path.write_text("{}", encoding="utf-8") + + with pytest.raises(DiagnosisError) as raised: + DiagnosisRun(harness, mine, _diagnoser("diagnose")).run() + + assert raised.value.code is DiagnosisErrorCode.ARTIFACT_INVALID + + +def test_snapshot_lineage_must_match_admission_revision_and_collection(tmp_path: Path) -> None: + mine, harness = _mine_result(tmp_path) + admission = mine.admissions[0] + path = admission.snapshot_path + assert path is not None + original = _SNAPSHOT_ADAPTER.validate_json(path.read_bytes()) + mismatches = ( + replace(original, trace=replace(original.trace, id=TraceId("swapped"))), + replace(original, revision_id=HarnessRevisionId("other-revision")), + replace(original, collection_digest=Sha256Digest("sha256:other-collection")), + ) + + for snapshot in mismatches: + payload = _SNAPSHOT_ADAPTER.dump_json(snapshot) + digest = Sha256Digest(f"sha256:{hashlib.sha256(payload).hexdigest()}") + path.write_bytes(payload) + changed_admission = replace(admission, snapshot_digest=digest) + changed_mine = replace(mine, admissions=(changed_admission, *mine.admissions[1:])) + with pytest.raises(DiagnosisError) as raised: + DiagnosisRun(harness, changed_mine, _diagnoser("diagnose")).run() + assert raised.value.code is DiagnosisErrorCode.ARTIFACT_INVALID + + +def test_cluster_identity_is_stable_across_revisions_and_tracks_parent(tmp_path: Path) -> None: + mine, harness = _mine_result(tmp_path) + first_mine = replace( + mine, + admissions=tuple( + admission for admission in mine.admissions if admission.trace_id != TraceId("tool-two") + ), + ) + first = DiagnosisRun(harness, first_mine, _diagnoser("diagnose")).run() + second = DiagnosisRun(harness, mine, _diagnoser("diagnose"), previous=first).run() + first_tool = next( + cluster for cluster in first.clusters if cluster.mechanism == MechanismKey("tool-schema") + ) + second_tool = next( + cluster for cluster in second.clusters if cluster.mechanism == MechanismKey("tool-schema") + ) + + assert second_tool.id == first_tool.id + assert second_tool.revision == 2 + assert second_tool.parents == (ClusterRevisionRef(first_tool.id, first_tool.revision),) + assert second_tool.recurrence == 2 + + +def test_cluster_without_current_failures_is_revisioned_as_resolved(tmp_path: Path) -> None: + mine, harness = _mine_result(tmp_path) + diagnoser = _diagnoser("diagnose") + first = DiagnosisRun(harness, mine, diagnoser).run() + without_tools = replace( + mine, + admissions=tuple( + admission + for admission in mine.admissions + if admission.trace_id not in (TraceId("tool-one"), TraceId("tool-two")) + ), + ) + + second = DiagnosisRun(harness, without_tools, diagnoser, previous=first).run() + resolved = next( + cluster for cluster in second.clusters if cluster.mechanism == MechanismKey("tool-schema") + ) + + assert resolved.state is ClusterState.RESOLVED + assert resolved.revision == 2 + assert resolved.resolution_rate == 1.0