fix(goals): fence deletion against canonical writes

loopx/control_plane/goals/deletion_service.py

@@ -27,6 +27,8 @@
 from ...history import load_registry
 from ...registry import atomic_write_json, read_json
 from ...registry_writability import probe_registry_write_path
+from ..coordination.shadow_management import shadow_maintenance_lock_target
+from ..effect_runtime import CANONICAL_AUTHORITY_WRITE_TIMEOUT_SECONDS
 from ..runtime.time import now_local_iso
 from .activation import GoalActivationState, goal_activation_state
 from .activation_service import (
@@ -267,6 +269,11 @@ def _recovery_receipt_path(registry_path: Path, goal_id: str) -> Path:
     )
 
 
+def _canonical_writer_guard_path(target_registry: Path, goal_id: str) -> Path:
+    runtime_root = target_registry.expanduser().resolve().parent
+    return shadow_maintenance_lock_target(runtime_root, goal_id)
+
+
 def _fsync_parent(path: Path) -> None:
     if os.name != "posix":
         return
@@ -633,6 +640,13 @@ def _execute_recovery(record: Mapping[str, Any]) -> dict[str, Any]:
         source_registry if source_available else declared_source_registry
     )
     with ExitStack() as stack:
+        stack.enter_context(
+            exclusive_cross_runtime_file_lock(
+                _canonical_writer_guard_path(target_registry, str(record["goal_id"])),
+                operation="recover_delete_stopped_goal_canonical",
+                timeout_seconds=CANONICAL_AUTHORITY_WRITE_TIMEOUT_SECONDS,
+            )
+        )
         source_transaction = None
         if not _same_path(locked_source_registry, target_registry):
             if source_available and not same_registry:
@@ -1094,6 +1108,13 @@ def _execute_deletion(
         source_registry if source_available else declared_source_registry
     )
     with ExitStack() as stack:
+        stack.enter_context(
+            exclusive_cross_runtime_file_lock(
+                _canonical_writer_guard_path(target_registry, goal_id),
+                operation="delete_stopped_goal_canonical",
+                timeout_seconds=CANONICAL_AUTHORITY_WRITE_TIMEOUT_SECONDS,
+            )
+        )
         source_transaction = None
         if not _same_path(locked_source_registry, target_registry):
             if source_available and not same_registry:

loopx/semantics/project_registry_io_manifest_v1.json

@@ -1263,103 +1263,103 @@
     },
     {
       "site": "loopx/control_plane/goals/deletion_service.py::<module>._execute_deletion::codec_transaction:project_registry_transaction#1",
-      "line": 1101,
+      "line": 1122,
       "column": 21,
       "kind": "codec_transaction",
       "api": "project_registry_transaction",
       "classification": "codec_api"
     },
     {
       "site": "loopx/control_plane/goals/deletion_service.py::<module>._execute_recovery::codec_transaction:project_registry_transaction#1",
-      "line": 640,
+      "line": 654,
       "column": 21,
       "kind": "codec_transaction",
       "api": "project_registry_transaction",
       "classification": "codec_api"
     },
     {
       "site": "loopx/control_plane/goals/deletion_service.py::<module>._load_locked_payloads::direct_json_read:read_json#1",
-      "line": 892,
+      "line": 906,
       "column": 9,
       "kind": "direct_json_read",
       "api": "read_json",
       "classification": "global_registry_io"
     },
     {
       "site": "loopx/control_plane/goals/deletion_service.py::<module>._load_locked_payloads::codec_read:load_registry#1",
-      "line": 894,
+      "line": 908,
       "column": 14,
       "kind": "codec_read",
       "api": "load_registry",
       "classification": "codec_api"
     },
     {
       "site": "loopx/control_plane/goals/deletion_service.py::<module>._load_locked_payloads::direct_json_read:read_json#2",
-      "line": 898,
+      "line": 912,
       "column": 22,
       "kind": "direct_json_read",
       "api": "read_json",
       "classification": "global_registry_io"
     },
     {
       "site": "loopx/control_plane/goals/deletion_service.py::<module>._recovery_images::codec_read:decode_registry_snapshot#1",
-      "line": 498,
+      "line": 505,
       "column": 20,
       "kind": "codec_read",
       "api": "decode_registry_snapshot",
       "classification": "codec_api"
     },
     {
       "site": "loopx/control_plane/goals/deletion_service.py::<module>._recovery_states::codec_read:load_registry#1",
-      "line": 516,
+      "line": 523,
       "column": 23,
       "kind": "codec_read",
       "api": "load_registry",
       "classification": "codec_api"
     },
     {
       "site": "loopx/control_plane/goals/deletion_service.py::<module>._resolve_route::codec_read:load_registry#1",
-      "line": 723,
+      "line": 737,
       "column": 22,
       "kind": "codec_read",
       "api": "load_registry",
       "classification": "codec_api"
     },
     {
       "site": "loopx/control_plane/goals/deletion_service.py::<module>._resolve_route::direct_json_read:read_json#1",
-      "line": 724,
+      "line": 738,
       "column": 22,
       "kind": "direct_json_read",
       "api": "read_json",
       "classification": "global_registry_io"
     },
     {
       "site": "loopx/control_plane/goals/deletion_service.py::<module>._restore_locked_registry::direct_json_write:atomic_write_json#1",
-      "line": 1075,
+      "line": 1089,
       "column": 5,
       "kind": "direct_json_write",
       "api": "atomic_write_json",
       "classification": "global_registry_io"
     },
     {
       "site": "loopx/control_plane/goals/deletion_service.py::<module>._write_deletion::codec_read:load_registry#1",
-      "line": 1013,
+      "line": 1027,
       "column": 27,
       "kind": "codec_read",
       "api": "load_registry",
       "classification": "codec_api"
     },
     {
       "site": "loopx/control_plane/goals/deletion_service.py::<module>._write_deletion::direct_json_read:read_json#1",
-      "line": 1027,
+      "line": 1041,
       "column": 24,
       "kind": "direct_json_read",
       "api": "read_json",
       "classification": "global_registry_io"
     },
     {
       "site": "loopx/control_plane/goals/deletion_service.py::<module>._write_locked_registry::direct_json_write:atomic_write_json#1",
-      "line": 1062,
+      "line": 1076,
       "column": 5,
       "kind": "direct_json_write",
       "api": "atomic_write_json",

tests/control_plane/test_shadow_native_todo_update_e2e.py

@@ -33,13 +33,16 @@
     shadow_management_state_path,
 )
 from loopx.control_plane.effect_runtime import effect_runtime_result
-from loopx.control_plane.goals import source_session_recreation
+from loopx.control_plane.goals import deletion_service, source_session_recreation
+from loopx.control_plane.goals.activation_service import set_goal_activation_state
+from loopx.control_plane.goals.deletion_service import delete_stopped_goal
 from loopx.control_plane.goals.source_session_recreation import RecreateGoalRequest
 from loopx.control_plane.projects.registry_codec import (
     load_project_registry,
     source_session_registry_transaction,
 )
 from loopx.control_plane.todos import provider_update
+from loopx.global_registry import sync_project_registry_to_global
 
 REPO = Path(__file__).resolve().parents[2]
 GOAL, TODO = "goal-update", "todo_update_probe"
@@ -97,7 +100,12 @@ def workspace(tmp_path: Path, request: pytest.FixtureRequest) -> Workspace:
             "archive_state": "active", "claimed_by": "agent-a", "note": "Keep the note",
             "required_capabilities": ["code_review"], "excluded_agents": ["agent-b"],
             "evidence": "Complete provider metadata"}
-    if request.param in {"native", "source_sqlite"}:
+    if request.param in {
+        "native",
+        "source_sqlite",
+        "witnessed_file",
+        "witnessed_sqlite",
+    }:
         projection = {"goal_id": GOAL, "handoff_mode": "soft_claim", "todos": [todo], "leases": [],
             "todo_read_model": {"schema_version": TODO_DOMAIN_READ_RECORD_SCHEMA_VERSION, "todo_count": 1,
                 "records_sha256": hashlib.sha256(canonical_bytes([todo])).hexdigest(),
@@ -107,7 +115,11 @@ def workspace(tmp_path: Path, request: pytest.FixtureRequest) -> Workspace:
         projection = build_todo_runtime_shadow_projection(goal_id=GOAL, todos=[todo], handoff_mode="soft_claim")
     initialize_canonical_authority(
         runtime, GOAL, projection, state_path=state,
-        provider="sqlite" if source_profile else "file",
+        provider=(
+            "sqlite"
+            if request.param in {"source_sqlite", "witnessed_sqlite"}
+            else "file"
+        ),
     )
     state.unlink()  # Neither update route may require or recreate Markdown.
     return Workspace(runtime, registry, state, projection)
@@ -458,3 +470,166 @@ def recreate() -> None:
     finally:
         stop(writer)
         thread.join(timeout=10)
+
+
+@pytest.mark.parametrize(
+    ("workspace", "recover"),
+    [
+        pytest.param("witnessed_file", False, id="file-delete"),
+        pytest.param("witnessed_sqlite", False, id="sqlite-delete"),
+        pytest.param("witnessed_sqlite", True, id="sqlite-recovery"),
+    ],
+    indirect=["workspace"],
+)
+def test_goal_deletion_waits_for_an_admitted_canonical_update(
+    workspace: Workspace,
+    recover: bool,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    synced = sync_project_registry_to_global(
+        registry_path=workspace.registry,
+        runtime_root_override=str(workspace.runtime),
+        goal_id=GOAL,
+        dry_run=False,
+    )
+    assert synced["ok"] is True
+    global_registry = workspace.runtime / "registry.global.json"
+    preview: dict
+    source_witness = workspace.registry
+    if recover:
+        stopped = set_goal_activation_state(
+            registry_path=global_registry,
+            goal_id=GOAL,
+            state="stopped",
+            actor_kind="owner",
+            execute=True,
+        )
+        assert stopped["ok"] is True
+        preview = delete_stopped_goal(
+            registry_path=global_registry,
+            goal_id=GOAL,
+            execute=False,
+        )
+        actual_write = deletion_service._write_locked_registry
+        write_count = 0
+
+        def interrupt_after_source_write(**kwargs: object) -> None:
+            nonlocal write_count
+            actual_write(**kwargs)
+            write_count += 1
+            if write_count == 1:
+                raise KeyboardInterrupt("simulated deletion interruption")
+
+        monkeypatch.setattr(
+            deletion_service,
+            "_write_locked_registry",
+            interrupt_after_source_write,
+        )
+        with pytest.raises(KeyboardInterrupt, match="simulated deletion interruption"):
+            delete_stopped_goal(
+                registry_path=global_registry,
+                goal_id=GOAL,
+                execute=True,
+                expected_state_fingerprint=preview["observed_state_fingerprint"],
+                expected_source_basis=preview["source_basis"],
+            )
+        monkeypatch.setattr(deletion_service, "_write_locked_registry", actual_write)
+        assert load_project_registry(workspace.registry)["goals"] == []
+        assert len(load_project_registry(global_registry)["goals"]) == 1
+        source_witness = global_registry
+
+    request = workspace.request(
+        schema_version="loopx_local_coordination_todo_update_request_v2",
+        registry_source={
+            "path": str(source_witness.resolve()),
+            "sha256": hashlib.sha256(source_witness.read_bytes()).hexdigest(),
+        },
+        lifecycle_grants=[],
+        authority_reason=None,
+    )
+    writer = start(node_command("update_paused", request))
+    thread = None
+    try:
+        expect_barrier(writer, "commit")
+        if not recover:
+            stopped = set_goal_activation_state(
+                registry_path=global_registry,
+                goal_id=GOAL,
+                state="stopped",
+                actor_kind="owner",
+                execute=True,
+            )
+            assert stopped["ok"] is True
+            preview = delete_stopped_goal(
+                registry_path=global_registry,
+                goal_id=GOAL,
+                execute=False,
+            )
+        operation_started = threading.Event()
+        operation_finished = threading.Event()
+        results: list[dict] = []
+        errors: list[BaseException] = []
+        if recover:
+            actual_execute = deletion_service._execute_recovery
+
+            def observed_execute(record: dict) -> dict:
+                operation_started.set()
+                return actual_execute(record)
+
+            monkeypatch.setattr(
+                deletion_service,
+                "_execute_recovery",
+                observed_execute,
+            )
+        else:
+            actual_execute = deletion_service._execute_deletion
+
+            def observed_execute(**kwargs: object) -> None:
+                operation_started.set()
+                actual_execute(**kwargs)
+
+            monkeypatch.setattr(
+                deletion_service,
+                "_execute_deletion",
+                observed_execute,
+            )
+
+        def delete() -> None:
+            try:
+                results.append(delete_stopped_goal(
+                    registry_path=global_registry,
+                    goal_id=GOAL,
+                    execute=True,
+                    expected_state_fingerprint=preview["observed_state_fingerprint"],
+                    expected_source_basis=preview["source_basis"],
+                ))
+            except BaseException as error:
+                errors.append(error)
+            finally:
+                operation_finished.set()
+
+        thread = threading.Thread(target=delete)
+        thread.start()
+        assert operation_started.wait(timeout=5), "Goal deletion did not start"
+        assert not operation_finished.wait(timeout=0.2), (
+            "Goal deletion returned while an admitted canonical update "
+            "was paused before provider commit"
+        )
+        output, error = writer.communicate("continue\n", timeout=20)
+        assert writer.returncode == 0, output + error
+        update = json.loads(output)
+        assert update["status"] == "applied", update
+        thread.join(timeout=10)
+        assert not thread.is_alive()
+        assert errors == []
+        assert results and results[0]["ok"] is True
+        assert results[0]["recovered"] is recover
+        assert load_project_registry(workspace.registry)["goals"] == []
+        assert load_project_registry(global_registry)["goals"] == []
+        assert inspect(workspace)["head"]["head"]["todos"][0]["text"] == (
+            "Updated through native RPC"
+        )
+    finally:
+        stop(writer)
+        if thread is not None:
+            thread.join(timeout=10)