From df91bcc8ae1f44150e641b4a4985fb77ddb870c8 Mon Sep 17 00:00:00 2001 From: Duang777 Date: Wed, 7 Oct 2026 14:50:25 +0800 Subject: [PATCH] fix(goals): fence deletion against canonical writes Signed-off-by: Duang777 --- loopx/control_plane/goals/deletion_service.py | 21 ++ .../project_registry_io_manifest_v1.json | 26 +-- .../test_shadow_native_todo_update_e2e.py | 181 +++++++++++++++++- 3 files changed, 212 insertions(+), 16 deletions(-) diff --git a/loopx/control_plane/goals/deletion_service.py b/loopx/control_plane/goals/deletion_service.py index 2aee5d89e9..327a417676 100644 --- a/loopx/control_plane/goals/deletion_service.py +++ b/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: diff --git a/loopx/semantics/project_registry_io_manifest_v1.json b/loopx/semantics/project_registry_io_manifest_v1.json index eab2d56e80..99f6b45c3d 100644 --- a/loopx/semantics/project_registry_io_manifest_v1.json +++ b/loopx/semantics/project_registry_io_manifest_v1.json @@ -1263,7 +1263,7 @@ }, { "site": "loopx/control_plane/goals/deletion_service.py::._execute_deletion::codec_transaction:project_registry_transaction#1", - "line": 1101, + "line": 1122, "column": 21, "kind": "codec_transaction", "api": "project_registry_transaction", @@ -1271,7 +1271,7 @@ }, { "site": "loopx/control_plane/goals/deletion_service.py::._execute_recovery::codec_transaction:project_registry_transaction#1", - "line": 640, + "line": 654, "column": 21, "kind": "codec_transaction", "api": "project_registry_transaction", @@ -1279,7 +1279,7 @@ }, { "site": "loopx/control_plane/goals/deletion_service.py::._load_locked_payloads::direct_json_read:read_json#1", - "line": 892, + "line": 906, "column": 9, "kind": "direct_json_read", "api": "read_json", @@ -1287,7 +1287,7 @@ }, { "site": "loopx/control_plane/goals/deletion_service.py::._load_locked_payloads::codec_read:load_registry#1", - "line": 894, + "line": 908, "column": 14, "kind": "codec_read", "api": "load_registry", @@ -1295,7 +1295,7 @@ }, { "site": "loopx/control_plane/goals/deletion_service.py::._load_locked_payloads::direct_json_read:read_json#2", - "line": 898, + "line": 912, "column": 22, "kind": "direct_json_read", "api": "read_json", @@ -1303,7 +1303,7 @@ }, { "site": "loopx/control_plane/goals/deletion_service.py::._recovery_images::codec_read:decode_registry_snapshot#1", - "line": 498, + "line": 505, "column": 20, "kind": "codec_read", "api": "decode_registry_snapshot", @@ -1311,7 +1311,7 @@ }, { "site": "loopx/control_plane/goals/deletion_service.py::._recovery_states::codec_read:load_registry#1", - "line": 516, + "line": 523, "column": 23, "kind": "codec_read", "api": "load_registry", @@ -1319,7 +1319,7 @@ }, { "site": "loopx/control_plane/goals/deletion_service.py::._resolve_route::codec_read:load_registry#1", - "line": 723, + "line": 737, "column": 22, "kind": "codec_read", "api": "load_registry", @@ -1327,7 +1327,7 @@ }, { "site": "loopx/control_plane/goals/deletion_service.py::._resolve_route::direct_json_read:read_json#1", - "line": 724, + "line": 738, "column": 22, "kind": "direct_json_read", "api": "read_json", @@ -1335,7 +1335,7 @@ }, { "site": "loopx/control_plane/goals/deletion_service.py::._restore_locked_registry::direct_json_write:atomic_write_json#1", - "line": 1075, + "line": 1089, "column": 5, "kind": "direct_json_write", "api": "atomic_write_json", @@ -1343,7 +1343,7 @@ }, { "site": "loopx/control_plane/goals/deletion_service.py::._write_deletion::codec_read:load_registry#1", - "line": 1013, + "line": 1027, "column": 27, "kind": "codec_read", "api": "load_registry", @@ -1351,7 +1351,7 @@ }, { "site": "loopx/control_plane/goals/deletion_service.py::._write_deletion::direct_json_read:read_json#1", - "line": 1027, + "line": 1041, "column": 24, "kind": "direct_json_read", "api": "read_json", @@ -1359,7 +1359,7 @@ }, { "site": "loopx/control_plane/goals/deletion_service.py::._write_locked_registry::direct_json_write:atomic_write_json#1", - "line": 1062, + "line": 1076, "column": 5, "kind": "direct_json_write", "api": "atomic_write_json", diff --git a/tests/control_plane/test_shadow_native_todo_update_e2e.py b/tests/control_plane/test_shadow_native_todo_update_e2e.py index de0c1eee39..38db0a66c4 100644 --- a/tests/control_plane/test_shadow_native_todo_update_e2e.py +++ b/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)