From 51e2cbf3c2805509f37f81068c8a5f6c8745d51a Mon Sep 17 00:00:00 2001 From: divo12 Date: Sat, 22 Aug 2026 17:17:43 +0530 Subject: [PATCH 1/6] implement failure diagnosis and clustering --- src/ofw/__init__.py | 27 ++ src/ofw/_diagnosis_runner.py | 46 ++++ src/ofw/diagnosis.py | 469 +++++++++++++++++++++++++++++++++++ src/ofw/mine.py | 12 +- src/ofw/runtime.py | 6 +- tests/test_diagnosis.py | 266 ++++++++++++++++++++ 6 files changed, 817 insertions(+), 9 deletions(-) create mode 100644 src/ofw/_diagnosis_runner.py create mode 100644 src/ofw/diagnosis.py create mode 100644 tests/test_diagnosis.py diff --git a/src/ofw/__init__.py b/src/ofw/__init__.py index b339c02..4ab7694 100644 --- a/src/ofw/__init__.py +++ b/src/ofw/__init__.py @@ -26,6 +26,20 @@ Sha256Digest, WorkspaceFile, ) +from ofw.diagnosis import ( + 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, @@ -88,6 +102,7 @@ class _OfwNamespace: ServiceName = ServiceName MiningPolicy = MiningPolicy ScoreName = ScoreName + PythonDiagnoser = PythonDiagnoser def editable(self, path: Path) -> EditableFile: return editable(path) @@ -108,6 +123,7 @@ def collect( "AssetAccess", "CanaryCase", "CaseId", + "ClusterState", "ComponentKind", "CollectionError", "CollectionErrorCode", @@ -115,7 +131,14 @@ def collect( "CommandLoop", "CommandVerifier", "DockerCompose", + "DiagnosisResult", + "DiagnosisRun", + "DiagnosisError", + "DiagnosisErrorCode", "EditableFile", + "EvidenceAnchor", + "EvidenceAnchorKind", + "FailureCluster", "GitCommit", "Harness", "HarnessAsset", @@ -136,11 +159,13 @@ def collect( "MineError", "MineErrorCode", "MiningPolicy", + "MechanismKey", "RepositorySnapshot", "ProcessCommand", "ProcessLimits", "PythonEntrypoint", "PythonLoop", + "PythonDiagnoser", "PythonVerifier", "RunErrorCode", "RunResult", @@ -148,11 +173,13 @@ def collect( "ScoreName", "Sha256Digest", "ServiceName", + "Severity", "Subagent", "Tool", "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..b71386e --- /dev/null +++ b/src/ofw/diagnosis.py @@ -0,0 +1,469 @@ +"""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 FailureCluster: + id: ClusterId + revision: int + 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[ClusterId, ...] = () + + +@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 + 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 + + 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"{int(DiagnosisSchemaVersion.V1)}".encode() + ).hexdigest() + ) + clusters = _clusters(diagnoses, diagnoser_digest, diagnoses_digest) + result = DiagnosisResult( + DiagnosisSchemaVersion.V1, + run_id, + self.mine.id, + revision.id, + 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, + diagnoses_digest: Sha256Digest, +) -> tuple[FailureCluster, ...]: + mechanisms = tuple( + sorted( + { + diagnosis.mechanism + for diagnosis in diagnoses + if diagnosis.status is DiagnosisStatus.PROPOSED and diagnosis.mechanism is not None + }, + key=_mechanism_sort_key, + ) + ) + return tuple( + _cluster( + mechanism, + tuple(diagnosis for diagnosis in diagnoses if diagnosis.mechanism == mechanism), + diagnoser_digest, + diagnoses_digest, + ) + for mechanism in mechanisms + ) + + +def _cluster( + mechanism: MechanismKey, + diagnoses: tuple[TraceDiagnosis, ...], + diagnoser_digest: Sha256Digest, + diagnoses_digest: Sha256Digest, +) -> FailureCluster: + 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) + cluster_id = ClusterId( + "cluster_" + + hashlib.sha256( + "\0".join( + ( + mechanism.value, + str(diagnoser_digest), + str(diagnoses_digest), + *(trace_id.value for trace_id in trace_ids), + ) + ).encode() + ).hexdigest() + ) + return FailureCluster( + cluster_id, + 1, + mechanism, + first.title, + first.description, + trace_ids, + evidence, + components, + len(diagnoses), + severity, + confidence, + 0.0, + ClusterState.PROPOSED, + ) + + +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)) + return _SNAPSHOT_ADAPTER.validate_json(payload) + 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 2a7eb59..fbc5cc1 100644 --- a/src/ofw/mine.py +++ b/src/ofw/mine.py @@ -282,7 +282,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( @@ -315,9 +315,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 _classify( @@ -459,7 +459,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( @@ -482,8 +482,8 @@ 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..bfe7bc6 --- /dev/null +++ b/tests/test_diagnosis.py @@ -0,0 +1,266 @@ +"""Evidence-bound failure diagnosis and deterministic clustering.""" + +from __future__ import annotations + +import hashlib +import subprocess +import time +from datetime import UTC, datetime, timedelta +from pathlib import Path + +import pytest +from pydantic import TypeAdapter + +from ofw import ( + 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 From e7cff7aefe6a2c5cf7156f77f06095fd053f169e Mon Sep 17 00:00:00 2001 From: divo12 Date: Sat, 22 Aug 2026 17:20:01 +0530 Subject: [PATCH 2/6] record diagnoser lineage --- src/ofw/diagnosis.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/ofw/diagnosis.py b/src/ofw/diagnosis.py index b71386e..ac560a1 100644 --- a/src/ofw/diagnosis.py +++ b/src/ofw/diagnosis.py @@ -232,6 +232,7 @@ class DiagnosisResult: id: DiagnosisRunId mine_id: MineRunId revision_id: HarnessRevisionId + diagnoser_digest: Sha256Digest source_watermark: datetime diagnoses: tuple[TraceDiagnosis, ...] clusters: tuple[FailureCluster, ...] @@ -298,6 +299,7 @@ def run(self) -> DiagnosisResult: run_id, self.mine.id, revision.id, + diagnoser_digest, self.mine.window.end, diagnoses, clusters, From bfb4d0451286f08dfc312e9f4376e4075bf5756f Mon Sep 17 00:00:00 2001 From: divo12 Date: Sat, 22 Aug 2026 17:21:37 +0530 Subject: [PATCH 3/6] trim speculative cluster lifecycle fields --- src/ofw/__init__.py | 2 -- src/ofw/diagnosis.py | 15 --------------- tests/test_diagnosis.py | 2 -- 3 files changed, 19 deletions(-) diff --git a/src/ofw/__init__.py b/src/ofw/__init__.py index 4ab7694..afb3c4a 100644 --- a/src/ofw/__init__.py +++ b/src/ofw/__init__.py @@ -27,7 +27,6 @@ WorkspaceFile, ) from ofw.diagnosis import ( - ClusterState, DiagnosisError, DiagnosisErrorCode, DiagnosisResult, @@ -123,7 +122,6 @@ def collect( "AssetAccess", "CanaryCase", "CaseId", - "ClusterState", "ComponentKind", "CollectionError", "CollectionErrorCode", diff --git a/src/ofw/diagnosis.py b/src/ofw/diagnosis.py index ac560a1..dbe6a1e 100644 --- a/src/ofw/diagnosis.py +++ b/src/ofw/diagnosis.py @@ -53,14 +53,6 @@ class Severity(StrEnum): CRITICAL = "critical" -class ClusterState(StrEnum): - PROPOSED = "proposed" - CONFIRMED = "confirmed" - TARGETED = "targeted" - RESOLVED = "resolved" - REOPENED = "reopened" - - class DiagnosisStatus(StrEnum): PROPOSED = "proposed" ABSTAINED = "abstained" @@ -203,7 +195,6 @@ def __str__(self) -> str: @dataclass(frozen=True, slots=True) class FailureCluster: id: ClusterId - revision: int mechanism: MechanismKey title: str description: str @@ -213,9 +204,6 @@ class FailureCluster: recurrence: int severity: Severity confidence: float - resolution_rate: float - state: ClusterState - parents: tuple[ClusterId, ...] = () @dataclass(frozen=True, slots=True) @@ -384,7 +372,6 @@ def _cluster( ) return FailureCluster( cluster_id, - 1, mechanism, first.title, first.description, @@ -394,8 +381,6 @@ def _cluster( len(diagnoses), severity, confidence, - 0.0, - ClusterState.PROPOSED, ) diff --git a/tests/test_diagnosis.py b/tests/test_diagnosis.py index bfe7bc6..dcd0804 100644 --- a/tests/test_diagnosis.py +++ b/tests/test_diagnosis.py @@ -12,7 +12,6 @@ from pydantic import TypeAdapter from ofw import ( - ClusterState, DiagnosisError, DiagnosisErrorCode, DiagnosisRun, @@ -218,7 +217,6 @@ def test_verified_failures_form_evidence_bound_mechanism_clusters(tmp_path: Path 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) From cabcbb6e2a3a994a84c8d091743d60e2a935ce13 Mon Sep 17 00:00:00 2001 From: divo12 Date: Sat, 22 Aug 2026 17:22:17 +0530 Subject: [PATCH 4/6] stabilize per-cluster identity --- src/ofw/diagnosis.py | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/src/ofw/diagnosis.py b/src/ofw/diagnosis.py index dbe6a1e..26a0f5a 100644 --- a/src/ofw/diagnosis.py +++ b/src/ofw/diagnosis.py @@ -281,7 +281,7 @@ def run(self) -> DiagnosisResult: f"{int(DiagnosisSchemaVersion.V1)}".encode() ).hexdigest() ) - clusters = _clusters(diagnoses, diagnoser_digest, diagnoses_digest) + clusters = _clusters(diagnoses, diagnoser_digest) result = DiagnosisResult( DiagnosisSchemaVersion.V1, run_id, @@ -314,7 +314,6 @@ def _diagnose( def _clusters( diagnoses: tuple[TraceDiagnosis, ...], diagnoser_digest: Sha256Digest, - diagnoses_digest: Sha256Digest, ) -> tuple[FailureCluster, ...]: mechanisms = tuple( sorted( @@ -331,7 +330,6 @@ def _clusters( mechanism, tuple(diagnosis for diagnosis in diagnoses if diagnosis.mechanism == mechanism), diagnoser_digest, - diagnoses_digest, ) for mechanism in mechanisms ) @@ -341,7 +339,6 @@ def _cluster( mechanism: MechanismKey, diagnoses: tuple[TraceDiagnosis, ...], diagnoser_digest: Sha256Digest, - diagnoses_digest: Sha256Digest, ) -> FailureCluster: first = diagnoses[0] evidence = tuple(anchor for diagnosis in diagnoses for anchor in diagnosis.evidence) @@ -357,6 +354,7 @@ def _cluster( ) 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)) cluster_id = ClusterId( "cluster_" + hashlib.sha256( From fb9c52657aa6965c4ac4488b4412af504479282e Mon Sep 17 00:00:00 2001 From: divo12 Date: Sat, 22 Aug 2026 17:22:24 +0530 Subject: [PATCH 5/6] Revert "trim speculative cluster lifecycle fields" This reverts commit bfb4d0451286f08dfc312e9f4376e4075bf5756f. --- src/ofw/__init__.py | 2 ++ src/ofw/diagnosis.py | 15 +++++++++++++++ tests/test_diagnosis.py | 2 ++ 3 files changed, 19 insertions(+) diff --git a/src/ofw/__init__.py b/src/ofw/__init__.py index afb3c4a..4ab7694 100644 --- a/src/ofw/__init__.py +++ b/src/ofw/__init__.py @@ -27,6 +27,7 @@ WorkspaceFile, ) from ofw.diagnosis import ( + ClusterState, DiagnosisError, DiagnosisErrorCode, DiagnosisResult, @@ -122,6 +123,7 @@ def collect( "AssetAccess", "CanaryCase", "CaseId", + "ClusterState", "ComponentKind", "CollectionError", "CollectionErrorCode", diff --git a/src/ofw/diagnosis.py b/src/ofw/diagnosis.py index 26a0f5a..13c5075 100644 --- a/src/ofw/diagnosis.py +++ b/src/ofw/diagnosis.py @@ -53,6 +53,14 @@ class Severity(StrEnum): CRITICAL = "critical" +class ClusterState(StrEnum): + PROPOSED = "proposed" + CONFIRMED = "confirmed" + TARGETED = "targeted" + RESOLVED = "resolved" + REOPENED = "reopened" + + class DiagnosisStatus(StrEnum): PROPOSED = "proposed" ABSTAINED = "abstained" @@ -195,6 +203,7 @@ def __str__(self) -> str: @dataclass(frozen=True, slots=True) class FailureCluster: id: ClusterId + revision: int mechanism: MechanismKey title: str description: str @@ -204,6 +213,9 @@ class FailureCluster: recurrence: int severity: Severity confidence: float + resolution_rate: float + state: ClusterState + parents: tuple[ClusterId, ...] = () @dataclass(frozen=True, slots=True) @@ -370,6 +382,7 @@ def _cluster( ) return FailureCluster( cluster_id, + 1, mechanism, first.title, first.description, @@ -379,6 +392,8 @@ def _cluster( len(diagnoses), severity, confidence, + 0.0, + ClusterState.PROPOSED, ) diff --git a/tests/test_diagnosis.py b/tests/test_diagnosis.py index dcd0804..bfe7bc6 100644 --- a/tests/test_diagnosis.py +++ b/tests/test_diagnosis.py @@ -12,6 +12,7 @@ from pydantic import TypeAdapter from ofw import ( + ClusterState, DiagnosisError, DiagnosisErrorCode, DiagnosisRun, @@ -217,6 +218,7 @@ def test_verified_failures_form_evidence_bound_mechanism_clusters(tmp_path: Path 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) From b06cb27a68de982c673a65fd0e2e470e9abfab14 Mon Sep 17 00:00:00 2001 From: divo12 Date: Sat, 22 Aug 2026 17:28:57 +0530 Subject: [PATCH 6/6] bind diagnosis evidence and revision clusters --- src/ofw/__init__.py | 2 + src/ofw/diagnosis.py | 130 ++++++++++++++++++++++++++++++---------- tests/test_diagnosis.py | 71 ++++++++++++++++++++++ 3 files changed, 173 insertions(+), 30 deletions(-) diff --git a/src/ofw/__init__.py b/src/ofw/__init__.py index 4ab7694..5edb9a7 100644 --- a/src/ofw/__init__.py +++ b/src/ofw/__init__.py @@ -27,6 +27,7 @@ WorkspaceFile, ) from ofw.diagnosis import ( + ClusterRevisionRef, ClusterState, DiagnosisError, DiagnosisErrorCode, @@ -124,6 +125,7 @@ def collect( "CanaryCase", "CaseId", "ClusterState", + "ClusterRevisionRef", "ComponentKind", "CollectionError", "CollectionErrorCode", diff --git a/src/ofw/diagnosis.py b/src/ofw/diagnosis.py index 13c5075..60ecb76 100644 --- a/src/ofw/diagnosis.py +++ b/src/ofw/diagnosis.py @@ -200,10 +200,17 @@ 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 @@ -215,7 +222,7 @@ class FailureCluster: confidence: float resolution_rate: float state: ClusterState - parents: tuple[ClusterId, ...] = () + parents: tuple[ClusterRevisionRef, ...] = () @dataclass(frozen=True, slots=True) @@ -263,6 +270,7 @@ class DiagnosisRun: source: Harness | HarnessRevision mine: MineResult diagnoser: PythonDiagnoser + previous: DiagnosisResult | None = None def run(self) -> DiagnosisResult: revision = _resolve_revision(self.source) @@ -290,10 +298,11 @@ def run(self) -> DiagnosisResult: "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) + clusters = _clusters(diagnoses, diagnoser_digest, self.previous) result = DiagnosisResult( DiagnosisSchemaVersion.V1, run_id, @@ -326,32 +335,38 @@ def _diagnose( def _clusters( diagnoses: tuple[TraceDiagnosis, ...], diagnoser_digest: Sha256Digest, + previous: DiagnosisResult | None, ) -> tuple[FailureCluster, ...]: - mechanisms = tuple( - sorted( - { - diagnosis.mechanism - for diagnosis in diagnoses - if diagnosis.status is DiagnosisStatus.PROPOSED and diagnosis.mechanism is not None - }, - key=_mechanism_sort_key, - ) + 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( + _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( +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( @@ -367,22 +382,31 @@ def _cluster( 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)) - cluster_id = ClusterId( - "cluster_" - + hashlib.sha256( - "\0".join( - ( - mechanism.value, - str(diagnoser_digest), - str(diagnoses_digest), - *(trace_id.value for trace_id in trace_ids), - ) - ).encode() - ).hexdigest() + 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, - 1, + _cluster_id(mechanism), + revision, + content_digest, mechanism, first.title, first.description, @@ -392,8 +416,47 @@ def _cluster( len(diagnoses), severity, confidence, - 0.0, - ClusterState.PROPOSED, + 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, ) @@ -439,7 +502,14 @@ def _read_snapshot(admission: TraceAdmission, mine: MineResult) -> TraceSnapshot payload = resolved.read_bytes() if digest_bytes(payload) != digest: raise DiagnosisError(DiagnosisErrorCode.ARTIFACT_INVALID, str(path)) - return _SNAPSHOT_ADAPTER.validate_json(payload) + 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 diff --git a/tests/test_diagnosis.py b/tests/test_diagnosis.py index bfe7bc6..05a5c1a 100644 --- a/tests/test_diagnosis.py +++ b/tests/test_diagnosis.py @@ -5,6 +5,7 @@ import hashlib import subprocess import time +from dataclasses import replace from datetime import UTC, datetime, timedelta from pathlib import Path @@ -12,6 +13,7 @@ from pydantic import TypeAdapter from ofw import ( + ClusterRevisionRef, ClusterState, DiagnosisError, DiagnosisErrorCode, @@ -264,3 +266,72 @@ def test_tampered_snapshot_is_rejected_before_diagnosis(tmp_path: Path) -> None: 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