feat(experimental): export Focus–Field recovery operator result

cml/experimental/focus_field_export.py

@@ -0,0 +1,63 @@
+"""Portable serializer for CML Focus–Field Recovery v0.2 decisions.
+
+This module exports the recovery decision as data without granting authority and
+without importing RESONANCE or any downstream consumer.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from cml.experimental.focus_field import RecoveryDecision
+
+SOURCE_CONTRACT = "cml.focus-field.recovery-decision.v0.2"
+
+
+class FocusFieldExportError(ValueError):
+    pass
+
+
+def export_recovery_decision(
+    decision: RecoveryDecision,
+    *,
+    source_revision: str,
+) -> dict[str, Any]:
+    """Serialize one recovery decision with an explicit advisory boundary."""
+
+    if not isinstance(source_revision, str) or len(source_revision) != 40:
+        raise FocusFieldExportError("source_revision must be a full 40-character Git SHA")
+    try:
+        int(source_revision, 16)
+    except ValueError as exc:
+        raise FocusFieldExportError("source_revision must be hexadecimal") from exc
+
+    ranked = [
+        {
+            "anchor_id": item.anchor_id,
+            "total": item.total,
+            "concept_overlap": item.concept_overlap,
+            "value_overlap": item.value_overlap,
+            "goal_overlap": item.goal_overlap,
+            "causal_overlap": item.causal_overlap,
+            "phase_match": item.phase_match,
+            "temporal_proximity": item.temporal_proximity,
+            "unresolved_bonus": item.unresolved_bonus,
+            "evidence_quality": item.evidence_quality,
+            "verification_ready": item.verification_ready,
+            "rewind_steps_saved": item.rewind_steps_saved,
+        }
+        for item in decision.ranked_candidates
+    ]
+
+    return {
+        "source_contract": SOURCE_CONTRACT,
+        "source_revision": source_revision,
+        "state": decision.state,
+        "selected_anchor_id": decision.selected_anchor_id,
+        "score": decision.score,
+        "rewind_steps_saved": decision.rewind_steps_saved,
+        "trusted_continuation": decision.trusted_continuation,
+        "ranked_candidates": ranked,
+        "mode": "ADVISORY_ONLY",
+        "authority_granted": False,
+    }

tests/test_focus_field_export.py

@@ -0,0 +1,79 @@
+import unittest
+
+from cml.experimental.focus_field import CandidateScore, RecoveryDecision
+from cml.experimental.focus_field_export import (
+    FocusFieldExportError,
+    SOURCE_CONTRACT,
+    export_recovery_decision,
+)
+
+
+class FocusFieldExportTests(unittest.TestCase):
+    def decision(self):
+        candidate = CandidateScore(
+            anchor_id="node-8",
+            total=0.81,
+            concept_overlap=1.0,
+            value_overlap=1.0,
+            goal_overlap=1.0,
+            causal_overlap=1.0,
+            phase_match=1.0,
+            temporal_proximity=0.5,
+            unresolved_bonus=1.0,
+            evidence_quality=1.0,
+            verification_ready=True,
+            rewind_steps_saved=16,
+        )
+        return RecoveryDecision(
+            state="reanchored",
+            selected_anchor_id="node-8",
+            score=0.81,
+            ranked_candidates=(candidate,),
+            rewind_steps_saved=16,
+            trusted_continuation=True,
+        )
+
+    def test_export_is_portable_and_non_authorizing(self):
+        result = export_recovery_decision(
+            self.decision(),
+            source_revision="2a649903693fc61a560ee056834127ada3120206",
+        )
+
+        self.assertEqual(SOURCE_CONTRACT, result["source_contract"])
+        self.assertEqual("reanchored", result["state"])
+        self.assertEqual("node-8", result["selected_anchor_id"])
+        self.assertTrue(result["trusted_continuation"])
+        self.assertEqual("ADVISORY_ONLY", result["mode"])
+        self.assertFalse(result["authority_granted"])
+        self.assertEqual("node-8", result["ranked_candidates"][0]["anchor_id"])
+        self.assertTrue(result["ranked_candidates"][0]["verification_ready"])
+
+    def test_export_preserves_exploratory_state_without_promoting_trust(self):
+        decision = RecoveryDecision(
+            state="reanchored_exploratory",
+            selected_anchor_id="old-node",
+            score=0.52,
+            ranked_candidates=(),
+            rewind_steps_saved=9,
+            trusted_continuation=False,
+        )
+        result = export_recovery_decision(
+            decision,
+            source_revision="2a649903693fc61a560ee056834127ada3120206",
+        )
+
+        self.assertEqual("reanchored_exploratory", result["state"])
+        self.assertFalse(result["trusted_continuation"])
+        self.assertFalse(result["authority_granted"])
+
+    def test_export_rejects_non_full_revision(self):
+        with self.assertRaises(FocusFieldExportError):
+            export_recovery_decision(self.decision(), source_revision="2a6499")
+
+    def test_export_rejects_non_hex_revision(self):
+        with self.assertRaises(FocusFieldExportError):
+            export_recovery_decision(self.decision(), source_revision="z" * 40)
+
+
+if __name__ == "__main__":
+    unittest.main()