From f3e90cd72d97e214891b7f95e31875707c1fd106 Mon Sep 17 00:00:00 2001 From: Duang777 Date: Wed, 7 Oct 2026 16:02:14 +0800 Subject: [PATCH 1/2] fix(runtime): serialize archive with artifact writers Signed-off-by: Duang777 --- loopx/control_plane/quota/spend_commit.py | 60 +-- .../turn_driver/journal_store.py | 37 +- loopx/runtime.py | 88 ++++- .../project_registry_io_manifest_v1.json | 10 +- tests/test_runtime_archive_concurrency.py | 370 ++++++++++++++++++ 5 files changed, 532 insertions(+), 33 deletions(-) create mode 100644 tests/test_runtime_archive_concurrency.py diff --git a/loopx/control_plane/quota/spend_commit.py b/loopx/control_plane/quota/spend_commit.py index 077b05185c..00ebf58ad7 100644 --- a/loopx/control_plane/quota/spend_commit.py +++ b/loopx/control_plane/quota/spend_commit.py @@ -5,7 +5,8 @@ from pathlib import Path from typing import Any -from ...file_lock import exclusive_file_lock +from ...file_lock import exclusive_cross_runtime_file_lock +from ..coordination.shadow_management import shadow_maintenance_lock_target from ..effect_runtime import ( EffectRuntimeConflict, EffectRuntimeRejected, @@ -226,20 +227,38 @@ def record_quota_slot_spend_from_preview( ): raise ValueError("quota spend preview index basis must be a string or null") if execute: - index_path = runtime_root / "goals" / safe_goal_id / "runs" / "index.jsonl" - if registry_path is None and goal_ref is None: - with exclusive_file_lock( - index_path, - operation="quota_spend_commit", + maintenance_target = shadow_maintenance_lock_target( + runtime_root.expanduser().resolve(), + safe_goal_id, + ) + + def commit(source_admission: Mapping[str, Any] | None) -> Mapping[str, Any]: + return _quota_spend_commit_result( + preview, + source=source, + generated_at=None, + execute=True, + runtime_root=runtime_root, + expected_index_digest=expected_index_digest, + goal_ref=goal_ref, + source_admission=source_admission, + ) + + # Legacy commits have no source-lifetime guard, so maintenance must be + # outermost. Exact commits preserve the established index -> source -> M order. + if goal_ref is None: + with exclusive_cross_runtime_file_lock( + maintenance_target, + operation="quota_spend_runtime_artifact_commit", ): - result = _quota_spend_commit_result( - preview, - source=source, - generated_at=None, - execute=True, + with quota_accounting_admission( runtime_root=runtime_root, - expected_index_digest=expected_index_digest, - ) + registry_path=registry_path, + goal_id=safe_goal_id, + goal_ref=None, + operation="quota_spend_commit", + ) as source_admission: + result = commit(source_admission) else: with quota_accounting_admission( runtime_root=runtime_root, @@ -248,16 +267,11 @@ def record_quota_slot_spend_from_preview( goal_ref=goal_ref, operation="quota_spend_commit", ) as source_admission: - result = _quota_spend_commit_result( - preview, - source=source, - generated_at=None, - execute=True, - runtime_root=runtime_root, - expected_index_digest=expected_index_digest, - goal_ref=goal_ref, - source_admission=source_admission, - ) + with exclusive_cross_runtime_file_lock( + maintenance_target, + operation="quota_spend_runtime_artifact_commit", + ): + result = commit(source_admission) else: result = _quota_spend_commit_result( preview, diff --git a/loopx/control_plane/turn_driver/journal_store.py b/loopx/control_plane/turn_driver/journal_store.py index f67860f8e9..4390bba3c2 100644 --- a/loopx/control_plane/turn_driver/journal_store.py +++ b/loopx/control_plane/turn_driver/journal_store.py @@ -8,7 +8,8 @@ from pathlib import Path from typing import Any -from ...file_lock import exclusive_file_lock +from ...file_lock import exclusive_cross_runtime_file_lock, exclusive_file_lock +from ..coordination.shadow_management import shadow_maintenance_lock_target from ..effect_runtime import EffectRuntimeConflict, effect_runtime_result from .turn_journal_runtime import ( write_turn_journal, @@ -57,18 +58,40 @@ def journal_committed_effect_id(journal: Mapping[str, Any]) -> str | None: return effect_id or None +def _turn_journal_goal_route(path: Path) -> tuple[Path, str] | None: + resolved = path.expanduser().resolve() + turns_dir = resolved.parent + goal_dir = turns_dir.parent + goals_dir = goal_dir.parent + if turns_dir.name != "turns" or goals_dir.name != "goals": + return None + return goals_dir.parent, goal_dir.name + + def write_turn_journal_checkpoint( path: Path, journal: Mapping[str, Any], *, source_admission: Mapping[str, Any] | None = None, ) -> None: - write_turn_journal( - str(path), - journal, - expected_effect_id=journal_committed_effect_id(journal), - source_admission=source_admission, - ) + def write() -> None: + write_turn_journal( + str(path), + journal, + expected_effect_id=journal_committed_effect_id(journal), + source_admission=source_admission, + ) + + route = _turn_journal_goal_route(path) + if route is None: + write() + return + runtime_root, goal_id = route + with exclusive_cross_runtime_file_lock( + shadow_maintenance_lock_target(runtime_root, goal_id), + operation="turn_journal_runtime_artifact_commit", + ): + write() def load_loopx_turn_plan_from_journal( diff --git a/loopx/runtime.py b/loopx/runtime.py index a3205f13fc..fffe17cc3f 100644 --- a/loopx/runtime.py +++ b/loopx/runtime.py @@ -1,6 +1,7 @@ from __future__ import annotations import shutil +from contextlib import ExitStack from datetime import datetime, timezone from pathlib import Path from typing import Any @@ -31,6 +32,26 @@ def unique_archive_path(archive_root: Path, goal_id: str, timestamp: str) -> Pat return candidate +def _archive_source_registry_paths( + registry: dict[str, Any], + *, + registry_path: Path, + goal_id: str, +) -> tuple[Path, ...]: + paths: set[Path] = set() + for goal in registry_goals(registry): + if str(goal.get("id")) != goal_id: + continue + raw_path = goal.get("source_registry") + if not isinstance(raw_path, str) or not raw_path.strip(): + continue + path = Path(raw_path).expanduser() + if not path.is_absolute(): + path = registry_path.expanduser().resolve().parent / path + paths.add(path.resolve()) + return tuple(sorted(paths, key=str)) + + def archive_runtime_goal( *, registry_path: Path, @@ -63,8 +84,71 @@ def archive_runtime_goal( destination = unique_archive_path(archive_base, safe_goal_id, timestamp) if execute: - destination.parent.mkdir(parents=True, exist_ok=True) - shutil.move(str(source), str(destination)) + from .control_plane.coordination.shadow_management import ( + shadow_maintenance_lock_target, + ) + from .control_plane.effect_runtime import ( + CANONICAL_AUTHORITY_WRITE_TIMEOUT_SECONDS, + ) + from .control_plane.goals.source_session_registry_state import guard_path + from .file_lock import exclusive_cross_runtime_file_lock + + with ExitStack() as locks: + # Exact writers enter source before maintenance. Match that order so + # archive cannot overtake a writer between its admission locks. + for source_registry in _archive_source_registry_paths( + registry, + registry_path=registry_path, + goal_id=safe_goal_id, + ): + locks.enter_context( + exclusive_cross_runtime_file_lock( + guard_path(source_registry, safe_goal_id), + operation="archive_runtime_goal_source", + ) + ) + locks.enter_context( + exclusive_cross_runtime_file_lock( + shadow_maintenance_lock_target( + runtime_root.resolve(), + safe_goal_id, + ), + operation="archive_runtime_goal", + timeout_seconds=CANONICAL_AUTHORITY_WRITE_TIMEOUT_SECONDS, + ) + ) + locks.enter_context( + exclusive_cross_runtime_file_lock( + registry_path.expanduser().resolve(), + operation="archive_runtime_goal_registry", + ) + ) + locked_registry = load_registry(registry_path) + registry_member = safe_goal_id in { + str(goal.get("id")) for goal in registry_goals(locked_registry) + } + if registry_member and not allow_registered: + raise ValueError( + "goal exists in registry; pass --allow-registered only " + "after confirming it is obsolete" + ) + if not source.exists(): + raise FileNotFoundError( + f"runtime goal directory does not exist: {source}" + ) + if not source.is_dir(): + raise ValueError(f"runtime goal path is not a directory: {source}") + if source.resolve().parent != goals_root: + raise ValueError( + "runtime goal directory resolved outside the goals root" + ) + destination = unique_archive_path( + archive_base, + safe_goal_id, + timestamp, + ) + destination.parent.mkdir(parents=True, exist_ok=True) + shutil.move(str(source), str(destination)) return { "ok": True, diff --git a/loopx/semantics/project_registry_io_manifest_v1.json b/loopx/semantics/project_registry_io_manifest_v1.json index eab2d56e80..40d59a2673 100644 --- a/loopx/semantics/project_registry_io_manifest_v1.json +++ b/loopx/semantics/project_registry_io_manifest_v1.json @@ -2159,12 +2159,20 @@ }, { "site": "loopx/runtime.py::.archive_runtime_goal::codec_read:load_registry#1", - "line": 43, + "line": 64, "column": 16, "kind": "codec_read", "api": "load_registry", "classification": "codec_api" }, + { + "site": "loopx/runtime.py::.archive_runtime_goal::codec_read:load_registry#2", + "line": 126, + "column": 31, + "kind": "codec_read", + "api": "load_registry", + "classification": "codec_api" + }, { "site": "loopx/state_backup.py::._registry_goals::direct_json_read:json.loads#1", "line": 54, diff --git a/tests/test_runtime_archive_concurrency.py b/tests/test_runtime_archive_concurrency.py new file mode 100644 index 0000000000..3fa087f0cb --- /dev/null +++ b/tests/test_runtime_archive_concurrency.py @@ -0,0 +1,370 @@ +from __future__ import annotations + +import json +import threading +from collections.abc import Callable, Iterator +from contextlib import contextmanager +from pathlib import Path +from typing import Any + +import pytest + +from loopx import runtime as runtime_module +from loopx.control_plane.coordination.shadow_management import ( + shadow_maintenance_lock_target, +) +from loopx.control_plane.effect_runtime import ( + effect_runtime_result as actual_effect_runtime_result, +) +from loopx.control_plane.projects.registry_codec import ( + source_session_registry_transaction, +) +from loopx.control_plane.quota import spend_commit +from loopx.control_plane.turn_driver import journal_store +from loopx.file_lock import ( + exclusive_cross_runtime_file_lock as actual_cross_runtime_file_lock, +) + + +GOAL_ID = "archive-writer-race" +GOAL_INSTANCE_ID = "ginst_aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" + + +def _decision(spent_slots: int) -> dict[str, Any]: + return { + "should_run": True, + "normal_delivery_allowed": True, + "recovery_delivery_allowed": False, + "effective_action": "advance", + "self_repair_allowed": False, + "capability_repair_allowed": False, + "workspace_repair_allowed": False, + "state": "eligible", + "safe_bypass_allowed": False, + "safe_bypass_kind": None, + "blocked_action_scope": None, + "quota": { + "compute": 1, + "window_hours": 24, + "slot_minutes": 1, + "spent_slots": spent_slots, + "allowed_slots": 1440, + }, + } + + +def _preview() -> dict[str, Any]: + return { + "ok": True, + "mode": "spend-slot", + "dry_run": True, + "goal_id": GOAL_ID, + "slots": 1, + "agent_id": "codex-main-control", + "appended": False, + "registry_mutated": False, + "before": _decision(0), + "after": _decision(1), + "after_recommended_action": "inspect next quota should-run decision", + "would_throttle": False, + "delivery_completion_spend": False, + "safe_bypass_spend": False, + "delivery_workspace_validated": False, + "expected_index_digest": None, + } + + +def _write_legacy_registry(registry_path: Path, runtime_root: Path) -> None: + registry_path.parent.mkdir(parents=True, exist_ok=True) + registry_path.write_text( + json.dumps( + { + "schema_version": 1, + "common_runtime_root": str(runtime_root), + "goals": [], + } + ), + encoding="utf-8", + ) + + +def _write_source_registry(registry_path: Path, runtime_root: Path) -> None: + payload = { + "schema_version": "0.2", + "registry_role": "project-local", + "profile_id": "source_session_v1", + "common_runtime_root": str(runtime_root), + "projects": [], + "goals": [ + { + "id": GOAL_ID, + "goal_instance_id": GOAL_INSTANCE_ID, + "status": "active", + "execution_authority": False, + } + ], + "session_bindings": [], + "session_receipts": [], + "lifetime_receipts": [], + "retired_goal_instances": [], + } + with source_session_registry_transaction( + registry_path, + operation="runtime_archive_writer_test", + create=lambda: payload, + ) as transaction: + transaction.commit(payload) + + +def _run_archive_race( + *, + registry_path: Path, + writer: Callable[[], Any], + writer_paused: threading.Event, + continue_writer: threading.Event, + allow_registered: bool, +) -> tuple[Any, dict[str, Any]]: + archive_finished = threading.Event() + writer_results: list[Any] = [] + archive_results: list[dict[str, Any]] = [] + errors: list[BaseException] = [] + + def run_writer() -> None: + try: + writer_results.append(writer()) + except BaseException as error: + errors.append(error) + + def archive() -> None: + try: + archive_results.append( + runtime_module.archive_runtime_goal( + registry_path=registry_path, + runtime_root_override=None, + goal_id=GOAL_ID, + archive_root=None, + allow_registered=allow_registered, + execute=True, + ) + ) + except BaseException as error: + errors.append(error) + finally: + archive_finished.set() + + writer_thread = threading.Thread(target=run_writer) + archive_thread = threading.Thread(target=archive) + writer_thread.start() + blocked = False + try: + assert writer_paused.wait(timeout=5), "writer did not reach its commit barrier" + archive_thread.start() + blocked = not archive_finished.wait(timeout=0.2) + finally: + continue_writer.set() + writer_thread.join(timeout=20) + if archive_thread.ident is not None: + archive_thread.join(timeout=20) + + assert blocked, f"archive completed while the writer was committing: {errors!r}" + assert not writer_thread.is_alive() + assert not archive_thread.is_alive() + assert errors == [] + assert len(writer_results) == 1 + assert len(archive_results) == 1 + return writer_results[0], archive_results[0] + + +@pytest.mark.parametrize("exact_source", [False, True], ids=["legacy", "exact"]) +def test_archive_waits_for_quota_writer_before_moving_goal( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + exact_source: bool, +) -> None: + runtime_root = tmp_path / "runtime" + source = runtime_root / "goals" / GOAL_ID + (source / "runs").mkdir(parents=True) + (source / "before.txt").write_text("before", encoding="utf-8") + writer_paused = threading.Event() + continue_writer = threading.Event() + goal_ref = None + + if exact_source: + source_registry = tmp_path / "project" / ".loopx" / "registry.json" + _write_source_registry(source_registry, runtime_root) + registry_path = runtime_root / "registry.global.json" + registry_path.write_text( + json.dumps( + { + "schema_version": "0.1", + "registry_role": "global-local", + "common_runtime_root": str(runtime_root), + "goals": [ + { + "id": GOAL_ID, + "source_registry": str(source_registry), + } + ], + } + ), + encoding="utf-8", + ) + goal_ref = { + "goal_id": GOAL_ID, + "goal_instance_id": GOAL_INSTANCE_ID, + } + + @contextmanager + def paused_maintenance_lock( + path: Path, + **kwargs: Any, + ) -> Iterator[Path]: + writer_paused.set() + if not continue_writer.wait(timeout=10): + raise TimeoutError("exact quota writer barrier was not released") + with actual_cross_runtime_file_lock(path, **kwargs) as lock_path: + yield lock_path + + monkeypatch.setattr( + spend_commit, + "exclusive_cross_runtime_file_lock", + paused_maintenance_lock, + ) + else: + source_registry = None + registry_path = tmp_path / "registry.json" + _write_legacy_registry(registry_path, runtime_root) + + def paused_effect(method: str, params: dict[str, Any]) -> Any: + writer_paused.set() + if not continue_writer.wait(timeout=10): + raise TimeoutError("legacy quota writer barrier was not released") + return actual_effect_runtime_result(method, params) + + monkeypatch.setattr(spend_commit, "effect_runtime_result", paused_effect) + + writer_result, archive_result = _run_archive_race( + registry_path=registry_path, + writer=lambda: spend_commit.record_quota_slot_spend_from_preview( + _preview(), + {"runtime_root": str(runtime_root)}, + goal_id=GOAL_ID, + execute=True, + registry_path=source_registry, + goal_ref=goal_ref, + ), + writer_paused=writer_paused, + continue_writer=continue_writer, + allow_registered=exact_source, + ) + + assert writer_result["appended"] is True + assert archive_result["archived"] is True + archive_path = Path(archive_result["archive_path"]) + assert not source.exists() + assert (archive_path / "before.txt").read_text(encoding="utf-8") == "before" + assert (archive_path / "runs" / "index.jsonl").is_file() + + +def test_archive_waits_for_turn_journal_writer_before_moving_goal( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + runtime_root = tmp_path / "runtime" + source = runtime_root / "goals" / GOAL_ID + source.mkdir(parents=True) + journal_path = source / "turns" / f"{'a' * 64}.json" + registry_path = tmp_path / "registry.json" + _write_legacy_registry(registry_path, runtime_root) + writer_paused = threading.Event() + continue_writer = threading.Event() + + def paused_write( + path: str, + journal: dict[str, Any], + **_: Any, + ) -> None: + writer_paused.set() + if not continue_writer.wait(timeout=10): + raise TimeoutError("Turn journal writer barrier was not released") + output = Path(path) + output.parent.mkdir(parents=True, exist_ok=True) + output.write_text(json.dumps(journal), encoding="utf-8") + + monkeypatch.setattr(journal_store, "write_turn_journal", paused_write) + _, archive_result = _run_archive_race( + registry_path=registry_path, + writer=lambda: journal_store.write_turn_journal_checkpoint( + journal_path, + {"schema_version": "loopx_turn_journal_v0", "goal_id": GOAL_ID}, + ), + writer_paused=writer_paused, + continue_writer=continue_writer, + allow_registered=False, + ) + + archive_path = Path(archive_result["archive_path"]) + assert not source.exists() + assert json.loads( + (archive_path / "turns" / journal_path.name).read_text(encoding="utf-8") + )["goal_id"] == GOAL_ID + + +def test_archive_rechecks_registry_membership_after_waiting_for_writer( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + runtime_root = tmp_path / "runtime" + source = runtime_root / "goals" / GOAL_ID + source.mkdir(parents=True) + registry_path = tmp_path / "registry.json" + _write_legacy_registry(registry_path, runtime_root) + archive_ready = threading.Event() + archive_finished = threading.Event() + errors: list[BaseException] = [] + actual_source_paths = runtime_module._archive_source_registry_paths + + def observed_source_paths(*args: Any, **kwargs: Any) -> tuple[Path, ...]: + result = actual_source_paths(*args, **kwargs) + archive_ready.set() + return result + + monkeypatch.setattr( + runtime_module, + "_archive_source_registry_paths", + observed_source_paths, + ) + + def archive() -> None: + try: + runtime_module.archive_runtime_goal( + registry_path=registry_path, + runtime_root_override=None, + goal_id=GOAL_ID, + archive_root=None, + allow_registered=False, + execute=True, + ) + except BaseException as error: + errors.append(error) + finally: + archive_finished.set() + + worker = threading.Thread(target=archive) + with actual_cross_runtime_file_lock( + shadow_maintenance_lock_target(runtime_root, GOAL_ID), + operation="runtime_archive_registration_test", + ): + worker.start() + assert archive_ready.wait(timeout=5), "archive did not finish preflight" + assert not archive_finished.wait(timeout=0.2) + registry = json.loads(registry_path.read_text(encoding="utf-8")) + registry["goals"] = [{"id": GOAL_ID}] + registry_path.write_text(json.dumps(registry), encoding="utf-8") + worker.join(timeout=10) + + assert not worker.is_alive() + assert len(errors) == 1 + assert isinstance(errors[0], ValueError) + assert "goal exists in registry" in str(errors[0]) + assert source.is_dir() From ef7b418c19ad512b686cdf69c7d06ab3846fed78 Mon Sep 17 00:00:00 2001 From: Duang777 Date: Wed, 7 Oct 2026 18:03:38 +0800 Subject: [PATCH 2/2] fix(runtime): fence unrouted artifact writers The earlier archive lock relied on global source routes, so an exact writer could hold index/source admission while an unrouted archive moved its goal directory. Acquire a stable runtime artifact guard before writer admission and archive source discovery, while preserving the existing inner lock order. Signed-off-by: Duang777 --- .../coordination/shadow_management.py | 4 + .../goals/first_party_host_admission.py | 71 +++-- loopx/control_plane/quota/spend_commit.py | 60 ++-- loopx/control_plane/turn_driver/executor.py | 4 +- .../turn_driver/journal_store.py | 22 +- loopx/runtime.py | 14 +- .../project_registry_io_manifest_v1.json | 8 +- .../test_turn_journal_goal_instance.py | 4 +- tests/test_loopx_turn_executor.py | 4 +- tests/test_runtime_archive_concurrency.py | 273 +++++++++++++++++- 10 files changed, 402 insertions(+), 62 deletions(-) diff --git a/loopx/control_plane/coordination/shadow_management.py b/loopx/control_plane/coordination/shadow_management.py index 0964d525b8..c64707a6d6 100644 --- a/loopx/control_plane/coordination/shadow_management.py +++ b/loopx/control_plane/coordination/shadow_management.py @@ -52,6 +52,10 @@ def shadow_management_directory(runtime_root: Path, goal_id: str) -> Path: return runtime_root / "authority-transition" / "file-v0" / f"shadow-management-{digest}" +def runtime_artifact_lock_target(runtime_root: Path, goal_id: str) -> Path: + return shadow_management_directory(runtime_root, goal_id) / "runtime-artifacts" + + def shadow_maintenance_lock_target(runtime_root: Path, goal_id: str) -> Path: return shadow_management_directory(runtime_root, goal_id) / "maintenance" diff --git a/loopx/control_plane/goals/first_party_host_admission.py b/loopx/control_plane/goals/first_party_host_admission.py index 8bdf0ff940..6de6df6360 100644 --- a/loopx/control_plane/goals/first_party_host_admission.py +++ b/loopx/control_plane/goals/first_party_host_admission.py @@ -10,6 +10,7 @@ cross_runtime_lock_witness, exclusive_cross_runtime_file_lock, ) +from ..coordination.shadow_management import runtime_artifact_lock_target from ..effect_runtime import effect_runtime_result from ..projects.registry_codec import ( SOURCE_SESSION_PROFILE_ID, @@ -68,6 +69,27 @@ def _effect( journal_path=self.journal_path, ) + @contextmanager + def _runtime_artifact_lifetime(self, *, operation: str) -> Iterator[None]: + path = self.journal_path.expanduser().resolve() + turns_dir = path.parent + goal_dir = turns_dir.parent + goals_dir = goal_dir.parent + if ( + turns_dir.name != "turns" + or goal_dir.name != self.goal_admission.goal_id + or goals_dir.name != "goals" + ): + raise FirstPartyHostRuntimeRejected("journal_path_mismatch") + with exclusive_cross_runtime_file_lock( + runtime_artifact_lock_target( + goals_dir.parent, + self.goal_admission.goal_id, + ), + operation=operation, + ): + yield + def prepare( self, step_kind: TurnProviderStepKind, @@ -75,13 +97,16 @@ def prepare( persist_journal: JournalPersist, ) -> None: try: - prepare_source_turn_effect( - registry_path=self.goal_admission.registry_path, - goal_id=self.goal_admission.goal_id, - effect=self._effect(step_kind, effect_ref), - source_admission=self.goal_admission.source_journal_admission_locked, - persist_journal=persist_journal, - ) + with self._runtime_artifact_lifetime( + operation="source_turn_effect_runtime_artifact_admit", + ): + prepare_source_turn_effect( + registry_path=self.goal_admission.registry_path, + goal_id=self.goal_admission.goal_id, + effect=self._effect(step_kind, effect_ref), + source_admission=self.goal_admission.source_journal_admission_locked, + persist_journal=persist_journal, + ) except SourceTurnEffectRejected as exc: raise FirstPartyHostRuntimeRejected(exc.code) from exc @@ -100,13 +125,16 @@ def release( persist_journal: JournalPersist, ) -> None: try: - release_source_turn_effect( - registry_path=self.goal_admission.registry_path, - goal_id=self.goal_admission.goal_id, - effect=self._effect(step_kind, effect_ref), - source_admission=self.goal_admission.source_journal_admission_locked, - persist_journal=persist_journal, - ) + with self._runtime_artifact_lifetime( + operation="source_turn_effect_runtime_artifact_release", + ): + release_source_turn_effect( + registry_path=self.goal_admission.registry_path, + goal_id=self.goal_admission.goal_id, + effect=self._effect(step_kind, effect_ref), + source_admission=self.goal_admission.source_journal_admission_locked, + persist_journal=persist_journal, + ) except SourceTurnEffectRejected as exc: raise FirstPartyHostRuntimeRejected(exc.code) from exc @@ -310,6 +338,8 @@ def current_lifetime(self, *, operation: str) -> Iterator[None]: @contextmanager def source_journal_admission( self, + *, + runtime_root: Path, ) -> Iterator[dict[str, Any] | None]: """Hand one journal mutation to the TS owner under the source guard.""" @@ -318,10 +348,17 @@ def source_journal_admission( return target = guard_path(self.registry_path, self.goal_id) with exclusive_cross_runtime_file_lock( - target, - operation="first_party_host_journal_commit", + runtime_artifact_lock_target( + runtime_root.expanduser().resolve(), + self.goal_id, + ), + operation="first_party_host_journal_runtime_artifact_guard", ): - yield self.source_journal_admission_locked() + with exclusive_cross_runtime_file_lock( + target, + operation="first_party_host_journal_commit", + ): + yield self.source_journal_admission_locked() def source_journal_admission_locked(self) -> dict[str, Any]: """Build a TS handoff while the caller holds this Goal's source guard.""" diff --git a/loopx/control_plane/quota/spend_commit.py b/loopx/control_plane/quota/spend_commit.py index 00ebf58ad7..9aea9f539d 100644 --- a/loopx/control_plane/quota/spend_commit.py +++ b/loopx/control_plane/quota/spend_commit.py @@ -6,7 +6,10 @@ from typing import Any from ...file_lock import exclusive_cross_runtime_file_lock -from ..coordination.shadow_management import shadow_maintenance_lock_target +from ..coordination.shadow_management import ( + runtime_artifact_lock_target, + shadow_maintenance_lock_target, +) from ..effect_runtime import ( EffectRuntimeConflict, EffectRuntimeRejected, @@ -227,8 +230,13 @@ def record_quota_slot_spend_from_preview( ): raise ValueError("quota spend preview index basis must be a string or null") if execute: + resolved_runtime_root = runtime_root.resolve() + artifact_target = runtime_artifact_lock_target( + resolved_runtime_root, + safe_goal_id, + ) maintenance_target = shadow_maintenance_lock_target( - runtime_root.expanduser().resolve(), + resolved_runtime_root, safe_goal_id, ) @@ -244,34 +252,38 @@ def commit(source_admission: Mapping[str, Any] | None) -> Mapping[str, Any]: source_admission=source_admission, ) - # Legacy commits have no source-lifetime guard, so maintenance must be - # outermost. Exact commits preserve the established index -> source -> M order. - if goal_ref is None: - with exclusive_cross_runtime_file_lock( - maintenance_target, - operation="quota_spend_runtime_artifact_commit", - ): + with exclusive_cross_runtime_file_lock( + artifact_target, + operation="quota_spend_runtime_artifact_guard", + ): + # Legacy commits have no source-lifetime guard, so maintenance must + # precede their index lock. Exact commits retain index -> source -> M. + if goal_ref is None: + with exclusive_cross_runtime_file_lock( + maintenance_target, + operation="quota_spend_runtime_artifact_commit", + ): + with quota_accounting_admission( + runtime_root=runtime_root, + registry_path=registry_path, + goal_id=safe_goal_id, + goal_ref=None, + operation="quota_spend_commit", + ) as source_admission: + result = commit(source_admission) + else: with quota_accounting_admission( runtime_root=runtime_root, registry_path=registry_path, goal_id=safe_goal_id, - goal_ref=None, + goal_ref=goal_ref, operation="quota_spend_commit", ) as source_admission: - result = commit(source_admission) - else: - with quota_accounting_admission( - runtime_root=runtime_root, - registry_path=registry_path, - goal_id=safe_goal_id, - goal_ref=goal_ref, - operation="quota_spend_commit", - ) as source_admission: - with exclusive_cross_runtime_file_lock( - maintenance_target, - operation="quota_spend_runtime_artifact_commit", - ): - result = commit(source_admission) + with exclusive_cross_runtime_file_lock( + maintenance_target, + operation="quota_spend_runtime_artifact_commit", + ): + result = commit(source_admission) else: result = _quota_spend_commit_result( preview, diff --git a/loopx/control_plane/turn_driver/executor.py b/loopx/control_plane/turn_driver/executor.py index 3b1649b50a..a7d93fd263 100644 --- a/loopx/control_plane/turn_driver/executor.py +++ b/loopx/control_plane/turn_driver/executor.py @@ -1400,7 +1400,9 @@ def persist_journal(snapshot: Mapping[str, Any]) -> None: _write_journal(journal_path, snapshot) return try: - with goal_admission.source_journal_admission() as source_admission: + with goal_admission.source_journal_admission( + runtime_root=runtime_root, + ) as source_admission: if source_admission is None: raise RuntimeError("source journal admission was not produced") _write_journal( diff --git a/loopx/control_plane/turn_driver/journal_store.py b/loopx/control_plane/turn_driver/journal_store.py index 4390bba3c2..80a9dd4339 100644 --- a/loopx/control_plane/turn_driver/journal_store.py +++ b/loopx/control_plane/turn_driver/journal_store.py @@ -9,7 +9,10 @@ from typing import Any from ...file_lock import exclusive_cross_runtime_file_lock, exclusive_file_lock -from ..coordination.shadow_management import shadow_maintenance_lock_target +from ..coordination.shadow_management import ( + runtime_artifact_lock_target, + shadow_maintenance_lock_target, +) from ..effect_runtime import EffectRuntimeConflict, effect_runtime_result from .turn_journal_runtime import ( write_turn_journal, @@ -87,11 +90,22 @@ def write() -> None: write() return runtime_root, goal_id = route + + def write_under_maintenance() -> None: + with exclusive_cross_runtime_file_lock( + shadow_maintenance_lock_target(runtime_root, goal_id), + operation="turn_journal_runtime_artifact_commit", + ): + write() + + if source_admission is not None: + write_under_maintenance() + return with exclusive_cross_runtime_file_lock( - shadow_maintenance_lock_target(runtime_root, goal_id), - operation="turn_journal_runtime_artifact_commit", + runtime_artifact_lock_target(runtime_root, goal_id), + operation="turn_journal_runtime_artifact_guard", ): - write() + write_under_maintenance() def load_loopx_turn_plan_from_journal( diff --git a/loopx/runtime.py b/loopx/runtime.py index fffe17cc3f..36d40d17f3 100644 --- a/loopx/runtime.py +++ b/loopx/runtime.py @@ -85,6 +85,7 @@ def archive_runtime_goal( if execute: from .control_plane.coordination.shadow_management import ( + runtime_artifact_lock_target, shadow_maintenance_lock_target, ) from .control_plane.effect_runtime import ( @@ -94,6 +95,17 @@ def archive_runtime_goal( from .file_lock import exclusive_cross_runtime_file_lock with ExitStack() as locks: + resolved_runtime_root = runtime_root.resolve() + locks.enter_context( + exclusive_cross_runtime_file_lock( + runtime_artifact_lock_target( + resolved_runtime_root, + safe_goal_id, + ), + operation="archive_runtime_goal_artifacts", + timeout_seconds=CANONICAL_AUTHORITY_WRITE_TIMEOUT_SECONDS, + ) + ) # Exact writers enter source before maintenance. Match that order so # archive cannot overtake a writer between its admission locks. for source_registry in _archive_source_registry_paths( @@ -110,7 +122,7 @@ def archive_runtime_goal( locks.enter_context( exclusive_cross_runtime_file_lock( shadow_maintenance_lock_target( - runtime_root.resolve(), + resolved_runtime_root, safe_goal_id, ), operation="archive_runtime_goal", diff --git a/loopx/semantics/project_registry_io_manifest_v1.json b/loopx/semantics/project_registry_io_manifest_v1.json index 40d59a2673..7833741e13 100644 --- a/loopx/semantics/project_registry_io_manifest_v1.json +++ b/loopx/semantics/project_registry_io_manifest_v1.json @@ -1367,7 +1367,7 @@ }, { "site": "loopx/control_plane/goals/first_party_host_admission.py::.FirstPartyHostGoalAdmission.for_plan::codec_read:load_project_registry#1", - "line": 216, + "line": 244, "column": 17, "kind": "codec_read", "api": "load_project_registry", @@ -1375,7 +1375,7 @@ }, { "site": "loopx/control_plane/goals/first_party_host_admission.py::.capture_first_party_host_goal_ref::codec_read:load_project_registry#1", - "line": 167, + "line": 195, "column": 16, "kind": "codec_read", "api": "load_project_registry", @@ -1383,7 +1383,7 @@ }, { "site": "loopx/control_plane/goals/first_party_host_admission.py::.source_goal_authority::codec_read:load_project_registry#1", - "line": 132, + "line": 160, "column": 20, "kind": "codec_read", "api": "load_project_registry", @@ -2167,7 +2167,7 @@ }, { "site": "loopx/runtime.py::.archive_runtime_goal::codec_read:load_registry#2", - "line": 126, + "line": 138, "column": 31, "kind": "codec_read", "api": "load_registry", diff --git a/tests/control_plane/test_turn_journal_goal_instance.py b/tests/control_plane/test_turn_journal_goal_instance.py index f2cb96e55c..11d86eeac4 100644 --- a/tests/control_plane/test_turn_journal_goal_instance.py +++ b/tests/control_plane/test_turn_journal_goal_instance.py @@ -123,7 +123,9 @@ def _commit_source( journal: dict[str, Any], admission: FirstPartyHostGoalAdmission, ) -> None: - with admission.source_journal_admission() as source_admission: + with admission.source_journal_admission( + runtime_root=path.parents[3], + ) as source_admission: assert source_admission is not None write_turn_journal_checkpoint( path, diff --git a/tests/test_loopx_turn_executor.py b/tests/test_loopx_turn_executor.py index 3020f85520..26528e3d1f 100644 --- a/tests/test_loopx_turn_executor.py +++ b/tests/test_loopx_turn_executor.py @@ -669,7 +669,9 @@ def test_cached_host_result_cannot_resume_after_goal_recreation( ) def write_source_journal() -> None: - with admission.source_journal_admission() as source_admission: + with admission.source_journal_admission( + runtime_root=runtime_root, + ) as source_admission: assert source_admission is not None turn_executor._write_journal( path, diff --git a/tests/test_runtime_archive_concurrency.py b/tests/test_runtime_archive_concurrency.py index 3fa087f0cb..cac9a2b0e4 100644 --- a/tests/test_runtime_archive_concurrency.py +++ b/tests/test_runtime_archive_concurrency.py @@ -16,12 +16,17 @@ from loopx.control_plane.effect_runtime import ( effect_runtime_result as actual_effect_runtime_result, ) +from loopx.control_plane import effect_runtime as effect_runtime_module +from loopx.control_plane.goals.first_party_host_admission import ( + FirstPartyHostGoalAdmission, +) from loopx.control_plane.projects.registry_codec import ( source_session_registry_transaction, ) from loopx.control_plane.quota import spend_commit from loopx.control_plane.turn_driver import journal_store from loopx.file_lock import ( + LockAcquireTimeoutError, exclusive_cross_runtime_file_lock as actual_cross_runtime_file_lock, ) @@ -74,6 +79,46 @@ def _preview() -> dict[str, Any]: } +def _turn_journal() -> dict[str, Any]: + turn_key = f"sha256:{'a' * 64}" + goal_ref = { + "goal_id": GOAL_ID, + "goal_instance_id": GOAL_INSTANCE_ID, + } + return { + "schema_version": "loopx_turn_journal_v0", + "goal_id": GOAL_ID, + "turn_key": turn_key, + "status": "in_progress", + "completed_phases": [], + "plan": { + "goal_ref": goal_ref, + "turn_envelope": { + "goal_id": GOAL_ID, + "agent_id": "codex-main-control", + "action": {"selected_todo": {"todo_id": "todo_archive_race"}}, + }, + "transaction": { + "turn_key": turn_key, + "goal_ref": goal_ref, + "settlement_plan": { + "schema_version": "quota_settlement_plan_v1", + "identity": { + "schema_version": "quota_settlement_identity_v0", + "effect_id": ( + f"{GOAL_ID}:codex-main-control:todo_archive_race:{turn_key}" + ), + "goal_id": GOAL_ID, + "agent_id": "codex-main-control", + "todo_id": "todo_archive_race", + "turn_instance_id": turn_key, + }, + }, + }, + }, + } + + def _write_legacy_registry(registry_path: Path, runtime_root: Path) -> None: registry_path.parent.mkdir(parents=True, exist_ok=True) registry_path.write_text( @@ -175,11 +220,16 @@ def archive() -> None: return writer_results[0], archive_results[0] -@pytest.mark.parametrize("exact_source", [False, True], ids=["legacy", "exact"]) +@pytest.mark.parametrize( + ("exact_source", "global_route"), + [(False, False), (True, True), (True, False)], + ids=["legacy", "exact-routed", "exact-unrouted"], +) def test_archive_waits_for_quota_writer_before_moving_goal( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, exact_source: bool, + global_route: bool, ) -> None: runtime_root = tmp_path / "runtime" source = runtime_root / "goals" / GOAL_ID @@ -199,12 +249,16 @@ def test_archive_waits_for_quota_writer_before_moving_goal( "schema_version": "0.1", "registry_role": "global-local", "common_runtime_root": str(runtime_root), - "goals": [ - { - "id": GOAL_ID, - "source_registry": str(source_registry), - } - ], + "goals": ( + [ + { + "id": GOAL_ID, + "source_registry": str(source_registry), + } + ] + if global_route + else [] + ), } ), encoding="utf-8", @@ -219,6 +273,10 @@ def paused_maintenance_lock( path: Path, **kwargs: Any, ) -> Iterator[Path]: + if path != shadow_maintenance_lock_target(runtime_root, GOAL_ID): + with actual_cross_runtime_file_lock(path, **kwargs) as lock_path: + yield lock_path + return writer_paused.set() if not continue_writer.wait(timeout=10): raise TimeoutError("exact quota writer barrier was not released") @@ -255,7 +313,7 @@ def paused_effect(method: str, params: dict[str, Any]) -> Any: ), writer_paused=writer_paused, continue_writer=continue_writer, - allow_registered=exact_source, + allow_registered=exact_source and global_route, ) assert writer_result["appended"] is True @@ -263,7 +321,125 @@ def paused_effect(method: str, params: dict[str, Any]) -> Any: archive_path = Path(archive_result["archive_path"]) assert not source.exists() assert (archive_path / "before.txt").read_text(encoding="utf-8") == "before" - assert (archive_path / "runs" / "index.jsonl").is_file() + rows = [ + json.loads(line) + for line in (archive_path / "runs" / "index.jsonl") + .read_text(encoding="utf-8") + .splitlines() + ] + assert len(rows) == 1 + assert rows[0]["classification"] == "quota_slot_spent" + assert rows[0].get("goal_ref") == ( + { + "goal_id": GOAL_ID, + "goal_instance_id": GOAL_INSTANCE_ID, + } + if exact_source + else None + ) + + +def test_archive_timeout_keeps_unrouted_exact_quota_settlement_together( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + runtime_root = tmp_path / "runtime" + source = runtime_root / "goals" / GOAL_ID + (source / "runs").mkdir(parents=True) + source_registry = tmp_path / "project" / ".loopx" / "registry.json" + _write_source_registry(source_registry, runtime_root) + registry_path = runtime_root / "registry.global.json" + registry_path.write_text( + json.dumps( + { + "schema_version": "0.1", + "registry_role": "global-local", + "common_runtime_root": str(runtime_root), + "goals": [], + } + ), + encoding="utf-8", + ) + writer_paused = threading.Event() + continue_writer = threading.Event() + writer_results: list[dict[str, Any]] = [] + writer_errors: list[BaseException] = [] + + @contextmanager + def paused_maintenance_lock( + path: Path, + **kwargs: Any, + ) -> Iterator[Path]: + if path != shadow_maintenance_lock_target(runtime_root, GOAL_ID): + with actual_cross_runtime_file_lock(path, **kwargs) as lock_path: + yield lock_path + return + writer_paused.set() + if not continue_writer.wait(timeout=10): + raise TimeoutError("exact quota writer barrier was not released") + with actual_cross_runtime_file_lock(path, **kwargs) as lock_path: + yield lock_path + + monkeypatch.setattr( + spend_commit, + "exclusive_cross_runtime_file_lock", + paused_maintenance_lock, + ) + monkeypatch.setattr( + effect_runtime_module, + "CANONICAL_AUTHORITY_WRITE_TIMEOUT_SECONDS", + 0.0, + ) + + def write() -> None: + try: + writer_results.append( + spend_commit.record_quota_slot_spend_from_preview( + _preview(), + {"runtime_root": str(runtime_root)}, + goal_id=GOAL_ID, + execute=True, + registry_path=source_registry, + goal_ref={ + "goal_id": GOAL_ID, + "goal_instance_id": GOAL_INSTANCE_ID, + }, + ) + ) + except BaseException as error: + writer_errors.append(error) + + writer = threading.Thread(target=write) + writer.start() + try: + assert writer_paused.wait(timeout=5), "writer did not reach maintenance" + with pytest.raises(LockAcquireTimeoutError): + runtime_module.archive_runtime_goal( + registry_path=registry_path, + runtime_root_override=None, + goal_id=GOAL_ID, + archive_root=None, + allow_registered=False, + execute=True, + ) + assert source.is_dir() + assert not (runtime_root / "archived-goals").exists() + finally: + continue_writer.set() + writer.join(timeout=20) + + assert not writer.is_alive() + assert writer_errors == [] + assert len(writer_results) == 1 + assert writer_results[0]["appended"] is True + rows = [ + json.loads(line) + for line in (source / "runs" / "index.jsonl") + .read_text(encoding="utf-8") + .splitlines() + ] + assert len(rows) == 1 + assert rows[0]["classification"] == "quota_slot_spent" def test_archive_waits_for_turn_journal_writer_before_moving_goal( @@ -310,6 +486,85 @@ def paused_write( )["goal_id"] == GOAL_ID +def test_archive_waits_for_unrouted_exact_turn_journal_writer( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + runtime_root = tmp_path / "runtime" + source = runtime_root / "goals" / GOAL_ID + source.mkdir(parents=True) + journal_path = source / "turns" / f"{'a' * 64}.json" + source_registry = tmp_path / "project" / ".loopx" / "registry.json" + _write_source_registry(source_registry, runtime_root) + registry_path = runtime_root / "registry.global.json" + registry_path.write_text( + json.dumps( + { + "schema_version": "0.1", + "registry_role": "global-local", + "common_runtime_root": str(runtime_root), + "goals": [], + } + ), + encoding="utf-8", + ) + admission = FirstPartyHostGoalAdmission.for_plan( + registry_path=source_registry, + goal_id=GOAL_ID, + planned_goal_ref={ + "goal_id": GOAL_ID, + "goal_instance_id": GOAL_INSTANCE_ID, + }, + ) + writer_paused = threading.Event() + continue_writer = threading.Event() + + @contextmanager + def paused_maintenance_lock( + path: Path, + **kwargs: Any, + ) -> Iterator[Path]: + writer_paused.set() + if not continue_writer.wait(timeout=10): + raise TimeoutError("exact Turn writer barrier was not released") + with actual_cross_runtime_file_lock(path, **kwargs) as lock_path: + yield lock_path + + monkeypatch.setattr( + journal_store, + "exclusive_cross_runtime_file_lock", + paused_maintenance_lock, + ) + + def write() -> None: + with admission.source_journal_admission( + runtime_root=runtime_root, + ) as source_admission: + assert source_admission is not None + journal_store.write_turn_journal_checkpoint( + journal_path, + _turn_journal(), + source_admission=source_admission, + ) + + _, archive_result = _run_archive_race( + registry_path=registry_path, + writer=write, + writer_paused=writer_paused, + continue_writer=continue_writer, + allow_registered=False, + ) + + archive_path = Path(archive_result["archive_path"]) + assert not source.exists() + assert ( + json.loads( + (archive_path / "turns" / journal_path.name).read_text(encoding="utf-8") + )["goal_id"] + == GOAL_ID + ) + + def test_archive_rechecks_registry_membership_after_waiting_for_writer( tmp_path: Path, monkeypatch: pytest.MonkeyPatch,