diff --git a/loopx/control_plane/work_items/task_lease_acquire_adapter.py b/loopx/control_plane/work_items/task_lease_acquire_adapter.py index 2b73f076ac..5b6ad0b6ea 100644 --- a/loopx/control_plane/work_items/task_lease_acquire_adapter.py +++ b/loopx/control_plane/work_items/task_lease_acquire_adapter.py @@ -8,16 +8,12 @@ from typing import Any from ..projects.registry_codec import load_registry -from ...paths import resolve_runtime_root from ..coordination.coordination_state_contract_generated import ( TASK_LEASE_CANONICAL_ACQUIRE_REQUEST_SCHEMA, TASK_LEASE_CANONICAL_LIFECYCLE_REQUEST_SCHEMA, TASK_LEASE_CANONICAL_CLAIM_TRANSFER_REQUEST_SCHEMA, TASK_LEASE_LIFECYCLE_REQUEST_SCHEMA, ) -from ..goals.legacy_event_source import ( - state_event_log_candidates as _state_event_log_candidates, -) from ..todos.handoff_mode import HANDOFF_MODE_LEGACY from .local_lease_record import TASK_LEASE_SCHEMA_VERSION, TaskLeaseError @@ -57,57 +53,6 @@ def _registry_goal(registry: dict[str, Any], goal_id: str) -> dict[str, Any] | N ) -def _task_lease_authority_source_paths( - *, - registry_path: Path, - registry: dict[str, Any], - goal_id: str, -) -> tuple[dict[str, Any] | None, Path | None, list[tuple[str, Path]]]: - from ...rollout_event_log import rollout_event_log_path - from ...state_refresh import resolve_goal_state - sources: list[tuple[str, Path]] = [("registry", registry_path)] - goal = _registry_goal(registry, goal_id) - if goal is None: - return None, None, sources - try: - _goal, _project, state_file = resolve_goal_state( - registry=registry, - goal_id=goal_id, - project_override=None, - state_file_override=None, - ) - except (OSError, ValueError): - return goal, None, sources - sources.append(("active_state", state_file)) - for index, path in enumerate( - _state_event_log_candidates( - goal, - state_path=state_file, - ) - ): - sources.append((f"state_event_{index}", path)) - projection_runtime_root = resolve_runtime_root( - registry, - None, - registry_path=registry_path, - ) - sources.append( - ( - "rollout_events", - rollout_event_log_path(projection_runtime_root, goal_id), - ) - ) - unique: list[tuple[str, Path]] = [] - seen: set[str] = set() - for source_id, path in sources: - key = str(path.expanduser().resolve(strict=False)) - if key in seen: - continue - seen.add(key) - unique.append((source_id, path)) - return goal, state_file, unique - - def _raw_registered_agent_candidates(goal: dict[str, Any]) -> list[Any]: coordination = goal.get("coordination") spawn_policy = goal.get("spawn_policy") @@ -122,101 +67,6 @@ def _raw_registered_agent_candidates(goal: dict[str, Any]) -> list[Any]: ] -def _task_lease_authority_projection( - *, - goal: dict[str, Any] | None, - state_file: Path | None, -) -> tuple[Any, list[Any]]: - from ..goals.active_state_metadata import parse_state_frontmatter - - if goal is None: - return HANDOFF_MODE_LEGACY, [] - - handoff_mode: Any = HANDOFF_MODE_LEGACY - if state_file is not None and state_file.exists(): - handoff_mode = parse_state_frontmatter( - state_file.read_text(encoding="utf-8") - ).get("handoff_mode") - return handoff_mode, _raw_registered_agent_candidates(goal) - - -def _authority_source_identity( - sources: list[tuple[str, Path]], -) -> list[tuple[str, str]]: - return [ - (source_id, str(path.expanduser().resolve(strict=False))) - for source_id, path in sources - ] - - -def _task_lease_authority_snapshot_attempt( - *, - registry_path: Path, - goal_id: str, - todo_id: str, -) -> dict[str, Any] | None: - registry_receipt_before = _authority_source_receipt("registry", registry_path) - registry = load_registry(registry_path) - goal, state_file, sources = _task_lease_authority_source_paths( - registry_path=registry_path, - registry=registry, - goal_id=goal_id, - ) - receipts_before = [ - _authority_source_receipt(source_id, path) for source_id, path in sources - ] - if receipts_before[0] != registry_receipt_before: - return None - - handoff_mode, registered_candidates = _task_lease_authority_projection( - goal=goal, state_file=state_file, - ) - - registry_after = load_registry(registry_path) - _goal_after, _state_file_after, sources_after = _task_lease_authority_source_paths( - registry_path=registry_path, - registry=registry_after, - goal_id=goal_id, - ) - if _authority_source_identity(sources_after) != _authority_source_identity(sources): - return None - receipts_after = [ - _authority_source_receipt(source_id, path) for source_id, path in sources_after - ] - if receipts_before != receipts_after: - return None - return { - "handoff_mode": handoff_mode, - "registered_agent_candidates": registered_candidates, - "todos": [], - "todo_projection_error": None, - "source_receipts": receipts_after, - } - - -def task_lease_acquire_authority_facts( - *, - registry_path: Path, - goal_id: str, - todo_id: str, -) -> dict[str, Any]: - """Read source-stable history metadata without reviving Todo admission.""" - - for _attempt in range(TASK_LEASE_AUTHORITY_SNAPSHOT_ATTEMPTS): - facts = _task_lease_authority_snapshot_attempt( - registry_path=registry_path, - goal_id=goal_id, - todo_id=todo_id, - ) - if facts is not None: - return facts - raise TaskLeaseError( - "task-lease authority sources changed while preparing acquire; retry", - code="authority_source_changed", - payload={"goal_id": goal_id, "todo_id": todo_id}, - ) - - def _require_native_acquire_shape(payload: object) -> dict[str, Any]: if ( not isinstance(payload, dict) diff --git a/tests/control_plane/test_canonical_lease_renew.py b/tests/control_plane/test_canonical_lease_renew.py index 43725c4b9e..ff4ff6d74f 100644 --- a/tests/control_plane/test_canonical_lease_renew.py +++ b/tests/control_plane/test_canonical_lease_renew.py @@ -23,7 +23,6 @@ from loopx.control_plane.coordination.runtime_shadow import ( build_todo_runtime_shadow_projection, ) -from loopx.control_plane.work_items import task_lease_acquire_adapter from loopx.control_plane.work_items.task_lease import renew_task_lease REPO = Path(__file__).resolve().parents[2] @@ -101,10 +100,15 @@ def renew(version, ttl=600, expected_exit=0): assert first["lease"]["write_scopes"] == lease["write_scopes"] second = renew(2) assert second["lease"]["version"] == 3 - def forbidden_legacy_path(*args, **kwargs): - raise AssertionError("canonical renew used a legacy capture/write path") + # Guard the actual source IO, not the name of a retired capture helper. + # Both read_bytes and read_text enter Path.open, even for missing files. + display_path, original_open = state.resolve(), Path.open + def forbidden_display_open(path, *args, **kwargs): + if path.resolve() == display_path: + raise AssertionError("canonical renew accessed the Goal display") + return original_open(path, *args, **kwargs) with monkeypatch.context() as guard: - guard.setattr(task_lease_acquire_adapter, "task_lease_acquire_authority_facts", forbidden_legacy_path) + guard.setattr(Path, "open", forbidden_display_open) third = renew_task_lease(registry_path=registry, runtime_root=runtime, goal_id=goal, todo_id="todo_renew", owner="agent-a", idempotency_key="execution-a", expected_version=3, ttl_seconds=600) assert third["lease"]["version"] == 4