Add local state write shadow validation

examples/catalog-canary-planner-smoke.py

@@ -62,6 +62,11 @@ def assert_profiles_come_from_catalog_matrix() -> None:
         "catalog-canary-contract",
         "benchmark-adapter-readiness",
     } <= domain_profile_ids, payload
+    domain_profiles = {profile["id"]: profile for profile in payload["domain_profiles"]}
+    state_write_commands = [
+        check["command"] for check in domain_profiles["state-write-correctness"]["checks"]
+    ]
+    assert "python3 examples/todo-write-correctness-smoke.py" in state_write_commands
 
 
 def assert_plan_selects_minimal_profiles_from_changed_surfaces() -> None:

examples/todo-write-correctness-smoke.py

@@ -12,6 +12,13 @@
 
 
 REPO_ROOT = Path(__file__).resolve().parents[1]
+if str(REPO_ROOT) not in sys.path:
+    sys.path.insert(0, str(REPO_ROOT))
+
+from loopx.local_state_write_correctness import (  # noqa: E402
+    shadow_validate_local_state_write_correctness_packet,
+)
+
 GOAL_ID = "todo-write-correctness-goal"
 AGENT_ID = "codex-product-capability"
 TODO_TEXT = "Preview a todo write correctness packet before mutating state."
@@ -187,6 +194,51 @@ def main() -> int:
             todo_id=add_dry_run["todo_id"],
             claimed_by=AGENT_ID,
         )
+        clean_shadow = shadow_validate_local_state_write_correctness_packet(
+            add_packet,
+            current_state_text=state_file.read_text(encoding="utf-8"),
+            observed_lease_ref=add_packet["write_intent"]["lease_ref"],
+        )
+        assert clean_shadow["apply_result"]["status"] == "preview_only", clean_shadow
+        assert clean_shadow["apply_result"]["conflict"] is None, clean_shadow
+
+        revision_conflict = shadow_validate_local_state_write_correctness_packet(
+            add_packet,
+            current_state_text=(
+                state_file.read_text(encoding="utf-8")
+                + "\n<!-- concurrent change -->\n"
+            ),
+        )
+        assert (
+            revision_conflict["apply_result"]["status"] == "revision_conflict"
+        ), revision_conflict
+        revision_conflict_detail = revision_conflict["apply_result"]["conflict"]
+        assert revision_conflict_detail["kind"] == "revision_conflict", revision_conflict
+        assert revision_conflict_detail["expected_revision"] == add_packet["write_intent"][
+            "expected_revision"
+        ], revision_conflict
+        assert revision_conflict_detail["current_revision"] != add_packet["write_intent"][
+            "expected_revision"
+        ], revision_conflict
+        assert state_file.read_text(encoding="utf-8") == original
+
+        foreign_lease = dict(add_packet["write_intent"]["lease_ref"])
+        foreign_lease["claimed_by"] = "codex-main-control"
+        foreign_lease["lease_id"] = f"lease_{add_dry_run['todo_id']}_codex_main_control"
+        lease_conflict = shadow_validate_local_state_write_correctness_packet(
+            add_packet,
+            current_state_text=state_file.read_text(encoding="utf-8"),
+            observed_lease_ref=foreign_lease,
+        )
+        assert lease_conflict["apply_result"]["status"] == "lease_conflict", lease_conflict
+        lease_conflict_detail = lease_conflict["apply_result"]["conflict"]
+        assert lease_conflict_detail["kind"] == "lease_conflict", lease_conflict
+        assert lease_conflict_detail["expected_lease_ref"] == add_packet["write_intent"][
+            "lease_ref"
+        ], lease_conflict
+        assert lease_conflict_detail["observed_lease_ref"] == foreign_lease, lease_conflict
+        assert state_file.read_text(encoding="utf-8") == original
+
         repeat_packet = add_dry_run_repeat["local_state_write_correctness"]
         assert (
             add_packet["write_intent"]["idempotency_key"]

loopx/canary/planner.py

@@ -573,6 +573,11 @@
                 "tier": "default",
                 "reason": "guards refresh-state update behavior and projection writes",
             },
+            {
+                "command": "python3 examples/todo-write-correctness-smoke.py",
+                "tier": "default",
+                "reason": "guards todo dry-run write correctness and shadow revision/lease validation",
+            },
             {
                 "command": "python3 examples/todo-concurrent-write-lock-smoke.py",
                 "tier": "deep",

loopx/local_state_write_correctness.py

@@ -100,6 +100,79 @@ def build_local_state_write_correctness_dry_run_packet(
     }
 
 
+def shadow_validate_local_state_write_correctness_packet(
+    packet: dict[str, Any],
+    *,
+    current_state_text: str,
+    observed_lease_ref: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    """Preview revision and lease conflicts without changing write behavior."""
+
+    validated = json.loads(json.dumps(packet, ensure_ascii=False))
+    intent = validated.get("write_intent") if isinstance(validated, dict) else {}
+    if not isinstance(intent, dict):
+        return validated
+    apply_result = validated.setdefault("apply_result", {})
+    if not isinstance(apply_result, dict):
+        apply_result = {}
+        validated["apply_result"] = apply_result
+    apply_result.setdefault("status", "preview_only")
+    apply_result.setdefault("applied_revision", None)
+    apply_result.setdefault("duplicate_of", None)
+    apply_result.setdefault("conflict", None)
+
+    expected_revision = intent.get("expected_revision")
+    current_revision = active_state_revision(current_state_text)
+    if (
+        isinstance(expected_revision, dict)
+        and expected_revision.get("value")
+        and expected_revision.get("value") != current_revision["value"]
+    ):
+        apply_result.update(
+            {
+                "status": "revision_conflict",
+                "applied_revision": None,
+                "duplicate_of": None,
+                "conflict": {
+                    "kind": "revision_conflict",
+                    "expected_revision": expected_revision,
+                    "current_revision": current_revision,
+                },
+            }
+        )
+        return validated
+
+    expected_lease_ref = intent.get("lease_ref")
+    if _lease_refs_conflict(expected_lease_ref, observed_lease_ref):
+        apply_result.update(
+            {
+                "status": "lease_conflict",
+                "applied_revision": None,
+                "duplicate_of": None,
+                "conflict": {
+                    "kind": "lease_conflict",
+                    "expected_lease_ref": expected_lease_ref,
+                    "observed_lease_ref": observed_lease_ref,
+                },
+            }
+        )
+    return validated
+
+
+def _lease_refs_conflict(expected: Any, observed: Any) -> bool:
+    if not isinstance(expected, dict) or not isinstance(observed, dict):
+        return False
+    if str(expected.get("kind") or "") != str(observed.get("kind") or ""):
+        return False
+    if str(expected.get("goal_id") or "") != str(observed.get("goal_id") or ""):
+        return False
+    if str(expected.get("todo_id") or "") != str(observed.get("todo_id") or ""):
+        return False
+    expected_holder = str(expected.get("claimed_by") or "")
+    observed_holder = str(observed.get("claimed_by") or "")
+    return bool(expected_holder and observed_holder and expected_holder != observed_holder)
+
+
 def _todo_lease_ref(
     *,
     goal_id: str,