From 2161ea4376909bbb923f8f7df695e8994064e6e7 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Tue, 27 Oct 2026 14:11:29 +0800 Subject: [PATCH 1/9] test: align release fixtures with canonical authority and replan contracts Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- examples/backlog-hygiene-smoke.py | 4 ++ ...actual_default_model_behavior_portfolio.py | 53 +++++++---------- ...capability_monitor_repair_tool_behavior.py | 6 ++ .../replan_semantic_action_behavior.py | 6 +- .../replan_vision_closeout_behavior.py | 1 + .../scoped_gate_successor_tool_behavior.py | 8 ++- .../testing/selected_todo_tool_behavior.py | 6 ++ .../canonical_authority_fixture.py | 32 +++++++++++ ...actual_default_model_behavior_portfolio.py | 7 ++- .../test_agent_capability_memory.py | 14 ++--- tests/control_plane/test_agent_context.py | 10 ++-- .../test_agent_lifecycle_state.py | 57 +++++++++++++------ tests/control_plane/test_agent_preferences.py | 48 +++++++++------- .../test_agent_scope_frontier_contract.py | 6 ++ .../control_plane/test_capability_gate_cli.py | 17 +++++- ...capability_monitor_repair_tool_behavior.py | 7 ++- .../test_causal_blocked_closeout_cli.py | 9 ++- tests/control_plane/test_contract_health.py | 4 ++ .../test_dated_vision_wait_cli.py | 24 ++++---- .../test_declared_terminal_settlement_cli.py | 6 +- .../test_deferred_original_turn_binding.py | 21 ++++--- .../test_delivery_semantics_cli.py | 3 + .../test_host_result_semantics.py | 8 +-- .../test_in_flight_turn_replay_cli.py | 4 +- .../test_interrupted_turn_continuation.py | 3 + .../test_quota_settlement_cli.py | 6 +- .../test_replan_semantic_action_behavior.py | 2 +- .../protocol_action_packet_compat.test.ts | 24 +++++++- 28 files changed, 263 insertions(+), 133 deletions(-) diff --git a/examples/backlog-hygiene-smoke.py b/examples/backlog-hygiene-smoke.py index 498ab56caf..a6d82a9366 100644 --- a/examples/backlog-hygiene-smoke.py +++ b/examples/backlog-hygiene-smoke.py @@ -11,6 +11,9 @@ REPO_ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(REPO_ROOT / "tests" / "control_plane")) +from canonical_authority_fixture import initialize_canonical_source_fixture # noqa: E402 + GOAL_ID = "backlog-hygiene-goal" @@ -112,6 +115,7 @@ def write_fixture( + "\n", encoding="utf-8", ) + initialize_canonical_source_fixture(runtime, GOAL_ID, state_path=state_path, provider="sqlite") return registry_path, runtime diff --git a/loopx/control_plane/testing/actual_default_model_behavior_portfolio.py b/loopx/control_plane/testing/actual_default_model_behavior_portfolio.py index b792cb3591..fd5a6b0d14 100644 --- a/loopx/control_plane/testing/actual_default_model_behavior_portfolio.py +++ b/loopx/control_plane/testing/actual_default_model_behavior_portfolio.py @@ -336,11 +336,11 @@ class _ContrastSpec: "decision", "user_action_required", "must_attempt_work", - "delivery_allowed", "quiet_noop_allowed", "external_write_requested", ), - ("selected_todo_id",), + # The Host must attempt the replan, without ordinary delivery authority. + ("selected_todo_id", "delivery_allowed"), ), ) ACTUAL_DEFAULT_MODEL_BEHAVIOR_CONTRAST_COUNT = len(_CONTRASTS) @@ -411,42 +411,29 @@ def actual_default_model_behavior_scenario_catalog() -> dict[str, Any]: def _write_scenario_project(root: Path) -> tuple[Path, Path]: + root = root.resolve() + from ..coordination.new_goal_storage_transport import initialize_goal_storage_target + from ..goals.source_session_registration import configure_source_creation_goal + project = root / "project" - state_file = ( - project - / ".codex" - / "goals" - / ACTUAL_DEFAULT_MODEL_BEHAVIOR_FIXTURE_GOAL_ID - / "ACTIVE_GOAL_STATE.md" - ) + state_file = project / ".codex" / "goals" / ACTUAL_DEFAULT_MODEL_BEHAVIOR_FIXTURE_GOAL_ID / "ACTIVE_GOAL_STATE.md" state_file.parent.mkdir(parents=True) state_file.write_text("# Active Goal State\n", encoding="utf-8") registry_path = project / ".loopx" / "registry.json" registry_path.parent.mkdir(parents=True) - registry_path.write_text( - json.dumps( - { - "schema_version": "0.1", - "goals": [ - { - "id": ACTUAL_DEFAULT_MODEL_BEHAVIOR_FIXTURE_GOAL_ID, - "status": "active", - "repo": str(project), - "state_file": str(state_file.relative_to(project)), - "coordination": { - "agent_model": "peer_v1", - "registered_agents": [ - ACTUAL_DEFAULT_MODEL_BEHAVIOR_FIXTURE_AGENT_ID - ], - }, - } - ], - }, - indent=2, - ) - + "\n", - encoding="utf-8", - ) + runtime_root = root / "runtime" + goal = { + "id": ACTUAL_DEFAULT_MODEL_BEHAVIOR_FIXTURE_GOAL_ID, + "status": "active", "repo": str(project), + "state_file": str(state_file.relative_to(project)), + } + # This is a new empty synthetic Goal, not an imported legacy authority. + # Reuse the first-party typed creation owner before requesting its frontier. + configure_source_creation_goal(goal, [ACTUAL_DEFAULT_MODEL_BEHAVIOR_FIXTURE_AGENT_ID], + "sqlite", "portfolio-entry-fixture") + registry_path.write_text(json.dumps({"schema_version": "0.1", + "common_runtime_root": str(runtime_root), "goals": [goal]}, indent=2) + "\n", encoding="utf-8") + initialize_goal_storage_target(runtime_root, goal, registry_path=registry_path) return project, registry_path diff --git a/loopx/control_plane/testing/capability_monitor_repair_tool_behavior.py b/loopx/control_plane/testing/capability_monitor_repair_tool_behavior.py index f38ec73b11..3b1a75bcc0 100644 --- a/loopx/control_plane/testing/capability_monitor_repair_tool_behavior.py +++ b/loopx/control_plane/testing/capability_monitor_repair_tool_behavior.py @@ -156,6 +156,12 @@ def _build_capability_repair_fixture(root: Path) -> _SelectedTodoToolFixture: registry_path.parent.mkdir(parents=True, exist_ok=True) registry_path.write_text(registry_text + "\n", encoding="utf-8") + from .canary_harness import import_fixture_authority + + import_fixture_authority( + global_registry_path, SELECTED_TODO_TOOL_FIXTURE_GOAL_ID, provider="sqlite", + ) + prompt = build_heartbeat_prompt( goal_id=SELECTED_TODO_TOOL_FIXTURE_GOAL_ID, thin=True, diff --git a/loopx/control_plane/testing/replan_semantic_action_behavior.py b/loopx/control_plane/testing/replan_semantic_action_behavior.py index 52b0181e2c..7a300f6b86 100644 --- a/loopx/control_plane/testing/replan_semantic_action_behavior.py +++ b/loopx/control_plane/testing/replan_semantic_action_behavior.py @@ -642,16 +642,16 @@ def _quota_behavior_observation(packet: Mapping[str, Any]) -> dict[str, Any]: raise ValueError("quota does not expose the semantic replan contract") if not ( action.get("must_attempt") is True - and action.get("delivery_allowed") is True + and action.get("delivery_allowed") is False and action.get("quiet_noop_allowed") is False ): - raise ValueError("semantic replan must require one executable action") + raise ValueError("semantic replan must require its Host action without ordinary delivery authority") return { "decision": "execute", "selected_todo_id": None, "user_action_required": bool(user.get("action_required")), "must_attempt_work": True, - "delivery_allowed": True, + "delivery_allowed": False, "quiet_noop_allowed": False, "external_write_requested": False, "replan_obligation_id": obligation.get("obligation_id"), diff --git a/loopx/control_plane/testing/replan_vision_closeout_behavior.py b/loopx/control_plane/testing/replan_vision_closeout_behavior.py index 4249dfa03f..c25c639537 100644 --- a/loopx/control_plane/testing/replan_vision_closeout_behavior.py +++ b/loopx/control_plane/testing/replan_vision_closeout_behavior.py @@ -72,6 +72,7 @@ def required_vision_scenario_contract( "selected_todo_id": None, "user_action_required": False, "must_attempt_work": True, + "delivery_allowed": False, "quiet_noop_allowed": False, } if any(contract.get(field) != value for field, value in required.items()): diff --git a/loopx/control_plane/testing/scoped_gate_successor_tool_behavior.py b/loopx/control_plane/testing/scoped_gate_successor_tool_behavior.py index b275d42eca..b1db9cec49 100644 --- a/loopx/control_plane/testing/scoped_gate_successor_tool_behavior.py +++ b/loopx/control_plane/testing/scoped_gate_successor_tool_behavior.py @@ -131,7 +131,7 @@ def _build_scoped_gate_fixture(root: Path) -> _SelectedTodoToolFixture: "## Completed Work Archive\n\n" "- [x] [P0] Complete the quality prerequisite.\n" " \n", encoding="utf-8", ) @@ -168,6 +168,12 @@ def _build_scoped_gate_fixture(root: Path) -> _SelectedTodoToolFixture: registry_path.parent.mkdir(parents=True, exist_ok=True) registry_path.write_text(registry_text + "\n", encoding="utf-8") + from .canary_harness import import_fixture_authority + + import_fixture_authority( + global_registry_path, SELECTED_TODO_TOOL_FIXTURE_GOAL_ID, provider="sqlite", + ) + prompt = build_heartbeat_prompt( goal_id=SELECTED_TODO_TOOL_FIXTURE_GOAL_ID, thin=True, diff --git a/loopx/control_plane/testing/selected_todo_tool_behavior.py b/loopx/control_plane/testing/selected_todo_tool_behavior.py index aa39a9d66e..14e7ba977d 100644 --- a/loopx/control_plane/testing/selected_todo_tool_behavior.py +++ b/loopx/control_plane/testing/selected_todo_tool_behavior.py @@ -220,6 +220,12 @@ def _build_fixture( registry_path.parent.mkdir(parents=True, exist_ok=True) registry_path.write_text(registry_text + "\n", encoding="utf-8") + from .canary_harness import import_fixture_authority + + import_fixture_authority( + global_registry_path, SELECTED_TODO_TOOL_FIXTURE_GOAL_ID, provider="sqlite", + ) + if prior_in_flight_progress: run = { "generated_at": "2026-08-12T00:01:00+00:00", diff --git a/tests/control_plane/canonical_authority_fixture.py b/tests/control_plane/canonical_authority_fixture.py index f07b7758d0..057d3f14ff 100644 --- a/tests/control_plane/canonical_authority_fixture.py +++ b/tests/control_plane/canonical_authority_fixture.py @@ -99,6 +99,38 @@ def read_fixture_todos(state_path: Path) -> list[dict]: ] +def advance_canonical_fixture_authority( + runtime_root: Path, goal_id: str, projection: dict, *, operation_id: str, +) -> dict: + """Advance an already initialized disposable store through its real CAS. + + Consumer fixtures may author a later synthetic state, but must not replace + storage bytes, reset lineage, re-import Markdown or re-engage its fence. + """ + module = Path(__file__).resolve().parents[2] / "loopx/control_plane/coordination/local_authority_provider.ts" + script = ( + f"import {{openLocalAuthorityStore}} from {json.dumps(module.as_uri())};" + "let input='';for await(const chunk of process.stdin)input+=chunk;" + "const request=JSON.parse(input);" + "const store=await openLocalAuthorityStore(request.root,request.goal);" + "const head=await store.loadAuthority();" + "if(head.status!=='loaded')throw new Error(JSON.stringify(head));" + "const result=await store.commitAuthority({expected_provider_revision:head.provider_revision," + "operation_id:request.operation_id,events:[],next_projection:request.projection,receipts:[]});" + "process.stdout.write(JSON.stringify(result));" + ) + process = subprocess.run( + ["node", "--no-warnings", "--experimental-strip-types", "--input-type=module", "-e", script], + input=json.dumps({"root": str(runtime_root), "goal": goal_id, + "projection": projection, "operation_id": operation_id}), + capture_output=True, text=True, check=False, timeout=45, + ) + assert process.returncode == 0, process.stderr + result = json.loads(process.stdout) + assert result["status"] == "applied", result + return result + + def initialize_canonical_source_fixture( runtime_root: Path, goal_id: str, *, state_path: Path, provider: str = "file", handoff_mode: str = "soft_claim", diff --git a/tests/control_plane/test_actual_default_model_behavior_portfolio.py b/tests/control_plane/test_actual_default_model_behavior_portfolio.py index 320627da86..d54e44fe7c 100644 --- a/tests/control_plane/test_actual_default_model_behavior_portfolio.py +++ b/tests/control_plane/test_actual_default_model_behavior_portfolio.py @@ -348,7 +348,7 @@ def _replan_semantic_action_actor(_: str) -> dict[str, Any]: "replan_semantic_action_behavior_receipt_v0", selected_todo_id=None, user_action_required=False, - delivery_allowed=True, + delivery_allowed=False, semantic_action_accepted=True, context_delivery="host_projected", selected_semantic_outcomes=["fresh_vision_path_outcome"], @@ -1659,7 +1659,10 @@ def test_portfolio_real_tool_scenarios_choose_from_latest_quota_result( **_real_tool_actors(tmp_path / "real-tool-actors"), ) - assert result["qualification_passed"] is True + assert result["qualification_passed"] is True, [ + (row["scenario_id"], row.get("failure_codes"), row.get("repeat_diagnostics")) + for row in result["scenarios"] if row["status"] != "passed" + ] boundary = result["boundary"] assert boundary["tools_enabled"] is True assert boundary["tool_enabled_scenario_count"] == 5 diff --git a/tests/control_plane/test_agent_capability_memory.py b/tests/control_plane/test_agent_capability_memory.py index 3282734842..39e9f9849b 100644 --- a/tests/control_plane/test_agent_capability_memory.py +++ b/tests/control_plane/test_agent_capability_memory.py @@ -9,7 +9,7 @@ def test_next_turn_reuses_declared_runtime_capability(tmp_path): - project, runtime, registry = _write_fixture(tmp_path, required_capability="network") + project, runtime, registry = _write_fixture(tmp_path, required_capability="network", provider="sqlite") before = registry.read_bytes() args = ("quota", "should-run", "--goal-id", GOAL_ID, "--agent-id", AGENT_ID) rc, first = _run_cli(registry, runtime, *args, "--available-capability", "network") @@ -29,7 +29,7 @@ def test_next_turn_reuses_declared_runtime_capability(tmp_path): def test_goal_inheritance_local_failure_recovery_and_forget(tmp_path): - _, runtime, registry = _write_fixture(tmp_path, required_capability="network", claimed=False) + _, runtime, registry = _write_fixture(tmp_path, required_capability="network", claimed=False, provider="sqlite") data = json.loads(registry.read_text()) data["goals"][0]["coordination"]["available_capabilities"] = ["network"] data["goals"][0]["coordination"]["registered_agents"].append("agent-b") @@ -58,7 +58,7 @@ def test_goal_inheritance_local_failure_recovery_and_forget(tmp_path): def test_owner_authority_and_optional_enablement_are_not_remembered(tmp_path): - _, runtime, registry = _write_fixture(tmp_path, required_capability="credentials") + _, runtime, registry = _write_fixture(tmp_path, required_capability="credentials", provider="sqlite") args = ("quota", "should-run", "--goal-id", GOAL_ID, "--agent-id", AGENT_ID) rc, first = _run_cli(registry, runtime, *args, "--available-capability", "credentials", "--available-capability", "material_lifecycle") @@ -69,7 +69,7 @@ def test_owner_authority_and_optional_enablement_are_not_remembered(tmp_path): def test_observation_does_not_cross_registry_or_agent(tmp_path): - _, runtime, registry = _write_fixture(tmp_path, required_capability="network", claimed=False) + _, runtime, registry = _write_fixture(tmp_path, required_capability="network", claimed=False, provider="sqlite") data = json.loads(registry.read_text()) data["goals"][0]["coordination"]["registered_agents"].append("agent-b") registry.write_text(json.dumps(data)) @@ -85,7 +85,7 @@ def test_observation_does_not_cross_registry_or_agent(tmp_path): def test_turn_plan_stays_read_only_and_settlement_rechecks_memory(tmp_path): - _, runtime, registry = _write_fixture(tmp_path, required_capability="network") + _, runtime, registry = _write_fixture(tmp_path, required_capability="network", provider="sqlite") scope = ("--goal-id", GOAL_ID, "--agent-id", AGENT_ID) rc, plan = _run_cli(registry, runtime, "turn", "plan", *scope, "--available-capability", "network") assert rc == 0, plan @@ -109,7 +109,7 @@ def test_executing_managed_turn_remembers_through_turn_start_hook( tmp_path, capsys, monkeypatch ): project, runtime, registry = _write_fixture( - tmp_path, required_capability="network" + tmp_path, required_capability="network", provider="sqlite", ) def completed_turn(*args, **kwargs): @@ -173,7 +173,7 @@ def completed_turn(*args, **kwargs): def test_agent_alias_uses_existing_identity_normalization(tmp_path): - _, runtime, registry = _write_fixture(tmp_path, required_capability="network") + _, runtime, registry = _write_fixture(tmp_path, required_capability="network", provider="sqlite") rc, first = _run_cli(registry, runtime, "quota", "should-run", "--goal-id", GOAL_ID, "--agent-id", AGENT_ID.upper(), "--available-capability", "network") assert rc == 0, first diff --git a/tests/control_plane/test_agent_context.py b/tests/control_plane/test_agent_context.py index 178207875f..27336e3933 100644 --- a/tests/control_plane/test_agent_context.py +++ b/tests/control_plane/test_agent_context.py @@ -159,12 +159,10 @@ def test_shared_host_delivers_signed_context_and_original_legacy_detail(legacy_r prompt = render_prompt(authority) assert json.dumps(envelope["agent_context"], ensure_ascii=False, sort_keys=True, separators=(",", ":")) in prompt - if legacy_reference: - assert authority["detail_ref"] == envelope["detail_ref"] - assert authority["action_signature"] == envelope["action_signature"] - else: - assert "detail_ref" not in authority - assert "action_signature" not in authority + # The captured source accompanies current inline work context as well as + # historical context references; both retain the original signed carrier. + assert authority["detail_ref"] == envelope["detail_ref"] + assert authority["action_signature"] == envelope["action_signature"] invalid = deepcopy(envelope) invalid["agent_context"]["scope"]["agent_id"] = "different-agent" with pytest.raises(ValueError, match="action signature"): diff --git a/tests/control_plane/test_agent_lifecycle_state.py b/tests/control_plane/test_agent_lifecycle_state.py index 9dabf15e5c..ac9d167893 100644 --- a/tests/control_plane/test_agent_lifecycle_state.py +++ b/tests/control_plane/test_agent_lifecycle_state.py @@ -3,11 +3,12 @@ `executing` and `unknown` come from execution facts (Turn lane liveness, delegation worker locks, task leases), never from a Todo timestamp. The facts fixtures here are real: a lane held in-process, a holder record edited the -way a crash or another machine leaves it, a lease file, a delegation row lock. +way a crash or another machine leaves it, a canonical lease, a delegation row lock. """ from contextlib import contextmanager from datetime import datetime, timedelta, timezone import json +from hashlib import sha256 import os from pathlib import Path import subprocess @@ -23,6 +24,9 @@ from loopx.control_plane.quota.task_orchestration import apply_task_orchestration_contract from loopx.control_plane.turn_driver.lane_fence import turn_lane_singleflight, turn_lane_target from loopx.file_lock import exclusive_file_lock, lock_holder_path +from canonical_authority_fixture import initialize_canonical_authority, advance_canonical_fixture_authority, read_fixture_todos +from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection +from loopx.control_plane.coordination.legacy_writer_fence import legacy_coordination_writer_fence_path NOW = datetime(2026, 9, 18, 12, tzinfo=timezone.utc) GOAL = "test-goal" @@ -45,6 +49,7 @@ def build_projection(monkeypatch, *, age=None, binding=False, status="open", "host_surface": "codex-app"}] if binding else [], }}]}, "todo_index": {"items": todos}} if runtime_root is not None: + _ensure_canonical_execution_fixture(runtime_root) facts = collect_agent_execution_facts(runtime_root=runtime_root, status_payload=payload) execution_facts = {"peer": facts} if facts is not None and runtime_root is None else facts return projection.build_agent_management_projection(payload, execution_facts=execution_facts), todo @@ -154,17 +159,39 @@ def _dead_pid() -> int: continue +def _ensure_canonical_execution_fixture(runtime_root: Path) -> None: + if legacy_coordination_writer_fence_path(runtime_root=runtime_root, goal_id=GOAL).exists(): + return + state = runtime_root.parent / "execution-state.md" + state.write_text("# Execution fixture\n\n## Agent Todo\n\n" + "- [ ] Inspect the public contract.\n" + " \n") + initialize_canonical_authority(runtime_root, GOAL, + build_todo_runtime_shadow_projection(goal_id=GOAL, todos=read_fixture_todos(state), handoff_mode="soft_claim"), + state_path=state) + + +def _canonical_document(runtime_root: Path) -> Path: + _ensure_canonical_execution_fixture(runtime_root) + document, = (runtime_root / "authority/file-v0").rglob("authority-store-*.json") + return document + + def _write_lease(runtime_root: Path, *, expires_in_hours: float, status: str = "active", owner: str = "peer") -> None: now = datetime.now(timezone.utc) - path = runtime_root / "goals" / GOAL / "task-leases" / "todo_peer.json" - path.parent.mkdir(parents=True, exist_ok=True) - path.write_text(json.dumps({ + _ensure_canonical_execution_fixture(runtime_root) + lease = { "schema_version": "task_lease_v0", "goal_id": GOAL, "todo_id": "todo_peer", "owner": owner, "idempotency_key": "execution-peer", "version": 1, "lease_epoch": 1, "status": status, "write_scopes": ["src/**"], "acquire_ttl_seconds": 600, "acquired_at": now.isoformat(), "updated_at": now.isoformat(), "expires_at": (now + timedelta(hours=expires_in_hours)).isoformat(), - }), encoding="utf-8") + } + canonical = execution_facts.read_canonical_todos_if_promoted(runtime_root=runtime_root, goal_id=GOAL) + projection = build_todo_runtime_shadow_projection(goal_id=GOAL, todos=canonical["todos"], + handoff_mode="soft_claim", leases=[lease]) + advance_canonical_fixture_authority(runtime_root, GOAL, projection, + operation_id="fixture-lease:" + sha256(json.dumps(lease, sort_keys=True).encode()).hexdigest()) def test_a_lane_held_in_process_makes_a_stale_todo_executing(monkeypatch, tmp_path): @@ -232,9 +259,7 @@ def test_an_expired_active_lease_with_nothing_live_is_unknown(monkeypatch, tmp_p def test_unreadable_claimed_lease_fails_closed_without_hiding_healthy_peer(monkeypatch, tmp_path, lease_bytes): runtime_root = tmp_path / "runtime" - lease_path = runtime_root / "goals" / GOAL / "task-leases" / "todo_peer.json" - lease_path.parent.mkdir(parents=True) - lease_path.write_bytes(lease_bytes) + _canonical_document(runtime_root).write_bytes(lease_bytes) other_todo = {"todo_id": "todo_healthy", "goal_id": GOAL, "role": "agent", "claimed_by": "healthy", "status": "open", "task_class": "advancement_task"} with turn_lane_singleflight(runtime_root=runtime_root, goal_id=GOAL, @@ -261,11 +286,10 @@ def test_unreadable_claimed_lease_fails_closed_without_hiding_healthy_peer(monke assert contract["blocked_peer_lanes"][0]["reason_codes"] == ["peer_runtime_not_active"] -def test_unreadable_second_todo_overrides_fresh_lease_but_not_live_lane(monkeypatch, tmp_path): +def test_canonical_snapshot_failure_overrides_fresh_lease_but_not_live_lane(monkeypatch, tmp_path): runtime_root = tmp_path / "runtime" _write_lease(runtime_root, expires_in_hours=1) - lease_path = runtime_root / "goals" / GOAL / "task-leases" / "todo_second.json" - lease_path.write_text("{broken json", encoding="utf-8") + _canonical_document(runtime_root).write_text("{broken json", encoding="utf-8") second = {"todo_id": "todo_second", "goal_id": GOAL, "role": "agent", "claimed_by": "peer", "status": "open", "task_class": "advancement_task"} kwargs = {"age": 0, "runtime_root": runtime_root, "extra_todos": [second]} @@ -293,14 +317,14 @@ def unavailable(**_kwargs): assert packet["agents"][0]["execution"]["lease"] == {"status": "unavailable"} -def test_local_lease_oserror_is_unavailable(monkeypatch, tmp_path): +def test_canonical_lease_oserror_is_unavailable(monkeypatch, tmp_path): runtime_root = tmp_path / "runtime" runtime_root.mkdir() - def unreadable(_path): - raise OSError("synthetic lease read failure") + def unreadable(**_kwargs): + raise OSError("synthetic canonical lease read failure") - monkeypatch.setattr(execution_facts, "read_lease", unreadable) + monkeypatch.setattr(execution_facts, "read_canonical_todos_if_promoted", unreadable) packet, _ = build_projection(monkeypatch, age=0, runtime_root=runtime_root) assert packet["agents"][0]["state"] == "unknown" assert packet["agents"][0]["execution"]["lease"] == {"status": "unavailable"} @@ -309,6 +333,7 @@ def unreadable(_path): def _delegation_row(runtime_root: Path, *, goal: str, requester: str, agent: str = "peer", operation: str = "op-1", status: str = "running") -> tuple[Path, Path]: """A journal row where the delegation service keeps it, plus its execution slot.""" + _ensure_canonical_execution_fixture(runtime_root) store = manager_context_root(runtime_root) row_path = (store / "executions" / manager_context_hash([goal, requester]) / (manager_context_hash(operation) + ".json")) @@ -376,7 +401,7 @@ def test_an_operation_lock_without_its_execution_slot_is_not_a_worker(monkeypatc def _payload(registered): - todo = {"todo_id": "todo_peer", "goal_id": GOAL, "role": "agent", "claimed_by": "peer", "status": "open"} + todo = {"todo_id": "todo_peer", "role": "agent", "claimed_by": "peer", "status": "open"} return {"goal_filter": GOAL, "run_history": {"goals": [{ "id": GOAL, "coordination": {"registered_agents": list(registered)}}]}, "todo_index": {"items": [todo]}} diff --git a/tests/control_plane/test_agent_preferences.py b/tests/control_plane/test_agent_preferences.py index 43d8eab75e..5ea0d06875 100644 --- a/tests/control_plane/test_agent_preferences.py +++ b/tests/control_plane/test_agent_preferences.py @@ -1,6 +1,8 @@ """Production CLI -> durable store -> next-turn disclosure; no provider mocks.""" import json +import re from copy import deepcopy +from pathlib import Path import pytest from loopx.control_plane.turn_driver.host_candidate import extract_turn_authority, render_prompt @@ -10,25 +12,24 @@ ) -@pytest.mark.parametrize("provider", ["legacy", "file", "sqlite"]) +@pytest.mark.parametrize("provider", ["file", "sqlite"]) def test_inactive_preferences_are_silent_and_participating_hook_delivers_current_view(tmp_path, monkeypatch, provider): nested = tmp_path / ("workspace dir" * 12) nested.mkdir() project, runtime, registry = _write_fixture(nested, required_capability="network") - if provider != "legacy": - from canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime - from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection - from loopx.control_plane.todos.active_state_todo_parser import parse_active_state_todos + from canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime + from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection + from loopx.control_plane.todos.active_state_todo_parser import parse_active_state_todos - if provider == "sqlite": - isolate_sqlite_runtime(tmp_path, monkeypatch) - goal = json.loads(registry.read_text())["goals"][0] - state = project / goal["state_file"] - # Initialize from the complete source owner, never a bounded quota display. - fields = parse_active_state_todos(state.read_text(), goal=goal, item_limit=None) - projection = build_todo_runtime_shadow_projection(goal_id=GOAL_ID, - todos=fields["agent_todos"]["items"] + fields.get("user_todos", {}).get("items", []), handoff_mode="soft_claim") - initialize_canonical_authority(runtime, GOAL_ID, projection, state_path=state, provider=provider) + if provider == "sqlite": + isolate_sqlite_runtime(tmp_path, monkeypatch) + goal = json.loads(registry.read_text())["goals"][0] + state = project / goal["state_file"] + # Initialize from the complete source owner, never a bounded quota display. + fields = parse_active_state_todos(state.read_text(), goal=goal, item_limit=None) + projection = build_todo_runtime_shadow_projection(goal_id=GOAL_ID, + todos=fields["agent_todos"]["items"] + fields.get("user_todos", {}).get("items", []), handoff_mode="soft_claim") + initialize_canonical_authority(runtime, GOAL_ID, projection, state_path=state, provider=provider) before = registry.read_bytes() scope = ("--goal-id", GOAL_ID, "--agent-id", AGENT_ID) def call(action, *args): @@ -125,7 +126,7 @@ def call(action, *args): def test_invalid_requests_and_corrupt_storage_do_not_become_empty_memory(tmp_path): - _, runtime, registry = _write_fixture(tmp_path, required_capability="network") + _, runtime, registry = _write_fixture(tmp_path, required_capability="network", provider="sqlite") scope = ("--goal-id", GOAL_ID, "--agent-id", AGENT_ID) def call(action, *args): return _run_cli(registry, runtime, "semantic-preference", "agent", action, *scope, *args) @@ -178,7 +179,7 @@ def call(action, *args): @pytest.mark.parametrize("other_scope", ["goal", "agent"]) def test_shared_runtime_preserves_unconfigured_scope_projection(tmp_path, other_scope): - _, runtime, registry = _write_fixture(tmp_path / "first", required_capability="network") + _, runtime, registry = _write_fixture(tmp_path / "first", required_capability="network", provider="sqlite") document = json.loads(registry.read_text()) peer = "peer-without-preferences" if other_scope == "goal": @@ -186,6 +187,9 @@ def test_shared_runtime_preserves_unconfigured_scope_projection(tmp_path, other_ other_goal = json.loads(other_registry.read_text())["goals"][0] other_goal["id"] = "other-goal" document["goals"].append(other_goal) + from canonical_authority_fixture import initialize_canonical_source_fixture + initialize_canonical_source_fixture(runtime, "other-goal", + state_path=Path(other_goal["repo"]) / other_goal["state_file"], provider="sqlite") scope = ("--goal-id", "other-goal", "--agent-id", AGENT_ID) else: document["goals"][0]["coordination"]["registered_agents"].append(peer) @@ -210,7 +214,13 @@ def projections(): after = projections() for old, new in zip(before, after): assert new.get("turn_start_capability_hook_dispatch") == old.get("turn_start_capability_hook_dispatch") - assert extract_turn_authority(after[1]) == extract_turn_authority(before[1]) + old_authority, new_authority = map(extract_turn_authority, (before[1], after[1])) + for authority in (old_authority, new_authority): + # The complete observation includes a fresh clock. Keep every signed + # action dimension; only the raw observation's digest may differ. + observed_digest = authority["action_signature"].pop("source_decision_hash") + assert re.fullmatch(r"sha256:[0-9a-f]{64}", observed_digest) + assert new_authority == old_authority rc, empty = _run_cli(registry, runtime, "semantic-preference", "agent", "read", *scope) assert rc == 0 and empty["current"]["items"] == [] assert len(list((runtime / "agent-preferences").iterdir())) == 1 @@ -218,7 +228,7 @@ def projections(): def test_fresh_hook_uses_one_real_provider_snapshot_for_body(tmp_path, monkeypatch): from loopx.capabilities.semantic_preference import agent_preferences as adapter - _, runtime, registry = _write_fixture(tmp_path, required_capability="network") + _, runtime, registry = _write_fixture(tmp_path, required_capability="network", provider="sqlite") original = adapter.agent_preferences calls = [] def traced(**kwargs): @@ -250,7 +260,7 @@ def test_permission_denied_hook_is_not_empty_and_recovers(tmp_path, denied_targe import os if os.name == "nt" or os.geteuid() == 0: pytest.skip("mode 000 read denial needs a non-root POSIX host") - _, runtime, registry = _write_fixture(tmp_path, required_capability="network") + _, runtime, registry = _write_fixture(tmp_path, required_capability="network", provider="sqlite") scope = ("--goal-id", GOAL_ID, "--agent-id", AGENT_ID) rc, written = _run_cli(registry, runtime, "semantic-preference", "agent", "remember", *scope, "--key", "review", "--statement", "Use the designated reviewer.", diff --git a/tests/control_plane/test_agent_scope_frontier_contract.py b/tests/control_plane/test_agent_scope_frontier_contract.py index e50af0a168..35bdba024d 100644 --- a/tests/control_plane/test_agent_scope_frontier_contract.py +++ b/tests/control_plane/test_agent_scope_frontier_contract.py @@ -8,6 +8,8 @@ import sys import pytest +from canonical_authority_fixture import advance_canonical_fixture_authority, read_fixture_todos +from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection from examples.control_plane.quota_plan_fixtures import ( SCOPED_AGENT_ID, @@ -165,6 +167,10 @@ def test_cli_should_run_reads_disposable_state_and_emits_v1(tmp_path): " \n" ) + advance_canonical_fixture_authority(runtime, "half-speed", + build_todo_runtime_shadow_projection(goal_id="half-speed", todos=read_fixture_todos(state), + handoff_mode="soft_claim"), + operation_id="source-prerequisite-fixture") result = subprocess.run( [ sys.executable, diff --git a/tests/control_plane/test_capability_gate_cli.py b/tests/control_plane/test_capability_gate_cli.py index aada9f3d8b..569bee2fe6 100644 --- a/tests/control_plane/test_capability_gate_cli.py +++ b/tests/control_plane/test_capability_gate_cli.py @@ -4,7 +4,7 @@ import pytest from canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime from loopx.control_plane.coordination.runtime_shadow import build_runtime_shadow_source_snapshot -from loopx.control_plane.coordination.local_authority import read_canonical_todos_if_promoted +from loopx.control_plane.coordination.local_authority import CanonicalAuthorityRequired, read_canonical_todos_if_promoted from loopx.control_plane.testing.canary_harness import write_fixture_registry, run_json_cli_result @@ -33,11 +33,24 @@ def test_real_quota_capability_gate_does_not_enable_repair_outputs(tmp_path, mon state_path=state, registry_path=registry) initialize_canonical_authority(runtime, "goal-a", projection, state_path=state, provider=provider) before = state.read_bytes() - authority = read_canonical_todos_if_promoted(runtime_root=runtime, goal_id="goal-a") args = ["quota", "should-run", "--goal-id", "goal-a", "--agent-id", "agent-a", "--include-detail", "agent-todos", ] if network_available: args += ["--available-capability", "network"] + if provider == "markdown": + # Markdown is authored source, no longer an ordinary authority backend. + # Availability of a host capability cannot bypass that migration gate. + with pytest.raises(CanonicalAuthorityRequired): + read_canonical_todos_if_promoted(runtime_root=runtime, goal_id="goal-a") + code, rejected = run_json_cli_result(*args, registry_path=registry, runtime_root=runtime) + assert code == 1 and rejected["error_code"] == "canonical_authority_required", rejected + assert rejected["write_admitted"] is False + assert rejected["decision_read_from_provider"] is False + assert rejected["legacy_fallback_used"] is False + assert state.read_bytes() == before + assert not (runtime / "authority").exists() + return + authority = read_canonical_todos_if_promoted(runtime_root=runtime, goal_id="goal-a") code, packet = run_json_cli_result(*args, registry_path=registry, runtime_root=runtime) assert code == 0, packet gate = packet["capability_gate"] diff --git a/tests/control_plane/test_capability_monitor_repair_tool_behavior.py b/tests/control_plane/test_capability_monitor_repair_tool_behavior.py index b5bf2bd4c5..a03a6d031d 100644 --- a/tests/control_plane/test_capability_monitor_repair_tool_behavior.py +++ b/tests/control_plane/test_capability_monitor_repair_tool_behavior.py @@ -198,10 +198,15 @@ def transport(**_: Any) -> Mapping[str, Any]: def test_actor_allows_bounded_fixture_layout_preview_before_repair( - tmp_path: Path, + tmp_path: Path, monkeypatch, ) -> None: fixture_root = tmp_path / "actor" fixture = _build_capability_repair_fixture(fixture_root) + # Construct canonical authority once; the actor reuses this prepared fixture. + monkeypatch.setattr( + "loopx.control_plane.testing.capability_monitor_repair_tool_behavior._build_capability_repair_fixture", + lambda root: fixture, + ) state_path = ( fixture.project_root / ".codex/goals/portfolio-goal/ACTIVE_GOAL_STATE.md" diff --git a/tests/control_plane/test_causal_blocked_closeout_cli.py b/tests/control_plane/test_causal_blocked_closeout_cli.py index a5d7f7e4c4..8646087c72 100644 --- a/tests/control_plane/test_causal_blocked_closeout_cli.py +++ b/tests/control_plane/test_causal_blocked_closeout_cli.py @@ -7,7 +7,7 @@ import pytest -from canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime +from canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime, read_fixture_todos from loopx.cli import main as cli_main from test_quota_settlement_cli import ( AGENT_ID, ALTERNATIVE_TODO_ID, GOAL_ID, TODO_ID, @@ -73,12 +73,11 @@ def cli(*args: str, cwd: Path = project) -> tuple[int, dict]: "\n- [ ] [P2] Complete the transitive prerequisite.\n" f" \n" )) - rc, listed = cli("todo", "list", "--goal-id", GOAL_ID) - assert rc == 0, listed + authored_todos = read_fixture_todos(state) if defer: - next(todo for todo in listed["todos"] if todo["todo_id"] == TODO_ID)["claimed_by"] = AGENT_ID + next(todo for todo in authored_todos if todo["todo_id"] == TODO_ID)["claimed_by"] = AGENT_ID initialize_canonical_authority(runtime, GOAL_ID, build_todo_runtime_shadow_projection( - goal_id=GOAL_ID, todos=listed["todos"], handoff_mode="hard_lease" if defer else "soft_claim", leases=[], + goal_id=GOAL_ID, todos=authored_todos, handoff_mode="hard_lease" if defer else "soft_claim", leases=[], ), state_path=state, provider=provider) binding = ("--agent-id", AGENT_ID, "--todo-id", TODO_ID, "--turn-instance-id", f"causal-blocked-{kind}-{provider}") diff --git a/tests/control_plane/test_contract_health.py b/tests/control_plane/test_contract_health.py index 3c64d5b061..eb75be6502 100644 --- a/tests/control_plane/test_contract_health.py +++ b/tests/control_plane/test_contract_health.py @@ -5,6 +5,7 @@ import pytest +from canonical_authority_fixture import initialize_canonical_source_fixture from loopx.contract import check_contract from loopx.control_plane.goals.contract_health import ( contract_error_diagnostic, @@ -115,6 +116,9 @@ def _registry(tmp_path: Path) -> tuple[Path, Path, Path]: "authority_sources": [], } ) + for goal in goals: + initialize_canonical_source_fixture(runtime_root, goal["id"], + state_path=tmp_path / goal["state_file"], provider="sqlite") registry_path = tmp_path / ".loopx" / "registry.json" registry_path.parent.mkdir(parents=True) registry_path.write_text( diff --git a/tests/control_plane/test_dated_vision_wait_cli.py b/tests/control_plane/test_dated_vision_wait_cli.py index a2b49ec6a8..b839bb150c 100644 --- a/tests/control_plane/test_dated_vision_wait_cli.py +++ b/tests/control_plane/test_dated_vision_wait_cli.py @@ -9,6 +9,7 @@ import tempfile import pytest +from canonical_authority_fixture import read_fixture_todos, initialize_canonical_source_fixture from loopx.control_plane.effect_runtime import restart_effect_runtime @@ -81,19 +82,16 @@ def cli(registry: Path, runtime: Path, *args: str) -> tuple[int, dict]: return result.returncode, json.loads(result.stdout) -@pytest.mark.parametrize("provider", ["legacy", "file", "sqlite"]) +@pytest.mark.parametrize("provider", ["file", "sqlite"]) def test_future_vision_wait_is_quiet_without_binding_or_poll(tmp_path, isolated_effect, monkeypatch, provider): from canonical_authority_fixture import initialize_canonical_authority from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection registry, runtime, state = fixture(tmp_path) - if provider != "legacy": - if provider == "sqlite": - monkeypatch.setenv("NODE_OPTIONS", os.environ.get("NODE_OPTIONS", "") + " --experimental-sqlite") - rc, listed = cli(registry, runtime, "todo", "list", "--goal-id", GOAL) - assert rc == 0, listed - projection = build_todo_runtime_shadow_projection(goal_id=GOAL, handoff_mode="soft_claim", todos=listed["todos"]) - initialize_canonical_authority(runtime, GOAL, projection, state_path=state, provider=provider) + if provider == "sqlite": + monkeypatch.setenv("NODE_OPTIONS", os.environ.get("NODE_OPTIONS", "") + " --experimental-sqlite") + projection = build_todo_runtime_shadow_projection(goal_id=GOAL, handoff_mode="soft_claim", todos=read_fixture_todos(state)) + initialize_canonical_authority(runtime, GOAL, projection, state_path=state, provider=provider) original_state = state.read_bytes() args = ("quota", "should-run", "--codex-app", "--goal-id", GOAL, "--agent-id", AGENT, "--turn-instance-id", "scheduled-before-due") @@ -140,6 +138,7 @@ def test_future_vision_wait_is_quiet_without_binding_or_poll(tmp_path, isolated_ def test_due_vision_wait_restores_explicit_resume(tmp_path, isolated_effect): registry, runtime, state = fixture(tmp_path, "resume_at:2026-01-01T00:00:00Z") + initialize_canonical_source_fixture(runtime, GOAL, state_path=state, provider="sqlite") original = state.read_bytes() rc, guard = cli(registry, runtime, "quota", "should-run", "--codex-app", "--goal-id", GOAL, "--agent-id", AGENT, "--turn-instance-id", "scheduled-due") @@ -170,10 +169,8 @@ def test_real_cli_worker_keeps_automatic_wait_alive_and_honors_explicit_stop( registry, runtime, state = fixture(tmp_path) if provider == "sqlite": monkeypatch.setenv("NODE_OPTIONS", os.environ.get("NODE_OPTIONS", "") + " --experimental-sqlite") - rc, listed = cli(registry, runtime, "todo", "list", "--goal-id", GOAL) - assert rc == 0, listed initialize_canonical_authority(runtime, GOAL, build_todo_runtime_shadow_projection( - goal_id=GOAL, handoff_mode="soft_claim", todos=listed["todos"], + goal_id=GOAL, handoff_mode="soft_claim", todos=read_fixture_todos(state), ), state_path=state, provider=provider) original_state = state.read_bytes() packets, sleeps = [], [] @@ -225,6 +222,7 @@ def test_independent_work_is_not_silenced_by_scheduled_wait(tmp_path, isolated_e output.write("\n- [ ] [P2] Validate independent public evidence.\n" f" \n") + initialize_canonical_source_fixture(runtime, GOAL, state_path=state, provider="sqlite") rc, guard = cli(registry, runtime, "quota", "should-run", "--codex-app", "--goal-id", GOAL, "--agent-id", AGENT, "--turn-instance-id", "independent-work") assert rc == 0, guard @@ -243,11 +241,9 @@ def test_bound_monitor_keeps_exact_turn_and_agent_fences(tmp_path, isolated_effe f" \n") - rc, listed = cli(registry, runtime, "todo", "list", "--goal-id", GOAL) - assert rc == 0, listed initialize_canonical_authority( runtime, GOAL, - build_todo_runtime_shadow_projection(goal_id=GOAL, todos=listed["todos"], handoff_mode="soft_claim"), + build_todo_runtime_shadow_projection(goal_id=GOAL, todos=read_fixture_todos(state), handoff_mode="soft_claim"), state_path=state, ) rc, guard = cli(registry, runtime, "quota", "should-run", "--codex-app", diff --git a/tests/control_plane/test_declared_terminal_settlement_cli.py b/tests/control_plane/test_declared_terminal_settlement_cli.py index a7147d62ae..4447582c27 100644 --- a/tests/control_plane/test_declared_terminal_settlement_cli.py +++ b/tests/control_plane/test_declared_terminal_settlement_cli.py @@ -11,7 +11,7 @@ import pytest import test_quota_settlement_cli as cli -from canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime +from canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime, read_fixture_todos from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection from loopx.control_plane.coordination.local_authority_shadow_projection import canonical_bytes @@ -60,10 +60,8 @@ def _fixture(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, provider: str, *, def run(*args: str) -> tuple[int, dict[str, Any]]: return cli._run_cli(registry_path, runtime, *args, cwd=workspace) - code, listed = run("todo", "list", "--goal-id", cli.GOAL_ID) - assert code == 0, listed projection = build_todo_runtime_shadow_projection( - goal_id=cli.GOAL_ID, todos=listed["todos"], handoff_mode="hard_lease", leases=[], + goal_id=cli.GOAL_ID, todos=read_fixture_todos(state), handoff_mode="hard_lease", leases=[], ) # Actor audit is canonical-only; legacy Markdown import cannot author it. historical = [row for row in projection["todos"] if row["todo_id"].startswith("todo_historical_")] diff --git a/tests/control_plane/test_deferred_original_turn_binding.py b/tests/control_plane/test_deferred_original_turn_binding.py index bc9499abfe..14ccf86ee5 100644 --- a/tests/control_plane/test_deferred_original_turn_binding.py +++ b/tests/control_plane/test_deferred_original_turn_binding.py @@ -17,7 +17,7 @@ @pytest.mark.parametrize("hidden_count", [0, 60]) @pytest.mark.parametrize("checked", [False, True]) -@pytest.mark.parametrize("provider", ["legacy", "file", "sqlite"]) +@pytest.mark.parametrize("provider", ["file", "sqlite"]) def test_deferred_replay_keeps_original_binding_during_replan( tmp_path: Path, hidden_count: int, checked: bool, provider: str, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture, @@ -26,7 +26,7 @@ def test_deferred_replay_keeps_original_binding_during_replan( from canonical_authority_fixture import isolate_sqlite_runtime isolate_sqlite_runtime(tmp_path, monkeypatch) - project, runtime, registry = _write_fixture(tmp_path) + project, runtime, registry = _write_fixture(tmp_path, provider=provider) turn = "turn-original-deferred" guard = ("quota", "should-run", "--runtime-profile", "codex_app_ssh_goal", "--goal-id", GOAL_ID, "--agent-id", AGENT_ID, @@ -53,16 +53,15 @@ def test_deferred_replay_keeps_original_binding_during_replan( for i in range(hidden_count)) text = text.replace("## Agent Todo\n\n", "## Agent Todo\n\n" + rows) state.write_text(text) - if provider != "legacy": - from canonical_authority_fixture import initialize_canonical_authority - from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection - from loopx.control_plane.todos.active_state_todo_parser import parse_active_state_todos + from canonical_authority_fixture import advance_canonical_fixture_authority + from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection + from loopx.control_plane.todos.active_state_todo_parser import parse_active_state_todos - todos = parse_active_state_todos(text, item_limit=None)["agent_todos"]["items"] - initialize_canonical_authority(runtime, GOAL_ID, - build_todo_runtime_shadow_projection(goal_id=GOAL_ID, - todos=todos, handoff_mode="soft_claim", leases=[]), - state_path=state, provider=provider) + todos = parse_active_state_todos(text, item_limit=None)["agent_todos"]["items"] + advance_canonical_fixture_authority(runtime, GOAL_ID, + build_todo_runtime_shadow_projection(goal_id=GOAL_ID, + todos=todos, handoff_mode="soft_claim", leases=[]), + operation_id="deferred-candidate-frontier") from loopx.cli import main from loopx.cli_commands import quota diff --git a/tests/control_plane/test_delivery_semantics_cli.py b/tests/control_plane/test_delivery_semantics_cli.py index d883a4d202..efba5e1301 100644 --- a/tests/control_plane/test_delivery_semantics_cli.py +++ b/tests/control_plane/test_delivery_semantics_cli.py @@ -6,6 +6,8 @@ import subprocess import sys +from canonical_authority_fixture import initialize_canonical_source_fixture + from loopx.control_plane.work_items.work_lane_context import outcome_followthrough_hint from loopx.status import compact_post_handoff_run @@ -37,6 +39,7 @@ def test_refresh_history_preserves_labels_without_inventing_delivery(tmp_path: P "authority_sources": [], }], }), encoding="utf-8") + initialize_canonical_source_fixture(runtime, "delivery-semantics", state_path=state, provider="sqlite") def cli(*args: str) -> dict: completed = subprocess.run( diff --git a/tests/control_plane/test_host_result_semantics.py b/tests/control_plane/test_host_result_semantics.py index 44f148bc9d..8e90b195c7 100644 --- a/tests/control_plane/test_host_result_semantics.py +++ b/tests/control_plane/test_host_result_semantics.py @@ -7,7 +7,7 @@ import pytest from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection -from tests.control_plane.canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime +from tests.control_plane.canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime, read_fixture_todos from tests.control_plane.test_turn_host_decision_capture import _fixture_module @@ -35,12 +35,12 @@ def test_native_turn_rejects_invalid_candidate_before_writeback( f = _fixture_module() project, runtime, workspace, registry = f._write_fixture(tmp_path, turn_count=2) prefix = ["--registry", str(registry), "--runtime-root", str(runtime), "--format", "json"] - code, listed = f._run_cli([*prefix, "todo", "list", "--goal-id", f.GOAL_ID]) - assert code == 0, listed state = project / ".codex/goals" / f.GOAL_ID / "ACTIVE_GOAL_STATE.md" initialize_canonical_authority(runtime, f.GOAL_ID, - build_todo_runtime_shadow_projection(goal_id=f.GOAL_ID, handoff_mode="soft_claim", todos=listed["todos"]), + build_todo_runtime_shadow_projection(goal_id=f.GOAL_ID, handoff_mode="soft_claim", todos=read_fixture_todos(state)), state_path=state, provider=provider) + code, listed = f._run_cli([*prefix, "todo", "list", "--goal-id", f.GOAL_ID]) + assert code == 0, listed candidate = { "result_kind": "validated_progress", "classification": "artifact_checked", "summary": "The isolated artifact was checked.", diff --git a/tests/control_plane/test_in_flight_turn_replay_cli.py b/tests/control_plane/test_in_flight_turn_replay_cli.py index 7a0a9eed79..369766e66a 100644 --- a/tests/control_plane/test_in_flight_turn_replay_cli.py +++ b/tests/control_plane/test_in_flight_turn_replay_cli.py @@ -9,6 +9,7 @@ from canonical_authority_fixture import ( initialize_canonical_authority, isolate_sqlite_runtime, + read_fixture_todos, ) from loopx.control_plane.coordination.runtime_shadow import ( build_todo_runtime_shadow_projection, @@ -51,9 +52,8 @@ def todo() -> dict[str, Any]: listed = cli("todo", "list", "--goal-id", GOAL_ID) return next(item for item in listed["todos"] if item["todo_id"] == TODO_ID) - listed = cli("todo", "list", "--goal-id", GOAL_ID) projection = build_todo_runtime_shadow_projection( - goal_id=GOAL_ID, todos=listed["todos"], handoff_mode="soft_claim", + goal_id=GOAL_ID, todos=read_fixture_todos(state), handoff_mode="soft_claim", ) initialize_canonical_authority( runtime, GOAL_ID, projection, state_path=state, provider=provider, diff --git a/tests/control_plane/test_interrupted_turn_continuation.py b/tests/control_plane/test_interrupted_turn_continuation.py index c8cba0869e..4f91ab57ac 100644 --- a/tests/control_plane/test_interrupted_turn_continuation.py +++ b/tests/control_plane/test_interrupted_turn_continuation.py @@ -6,12 +6,14 @@ from test_quota_settlement_cli import ( AGENT_ID, GOAL_ID, TODO_ID, _run_cli, _run_generated_cli, _spend_run_count, _write_fixture, _configure_read_only_todo, + _initialize_fixture_authority, ) def test_original_turn_can_resume_and_settle_without_mutating_todo(tmp_path: Path) -> None: project, runtime, registry = _write_fixture(tmp_path) _configure_read_only_todo(project) + _initialize_fixture_authority(project, runtime, provider="sqlite") prior_turn = "interrupted-original" recovery_turn = "interrupted-recovery" @@ -69,6 +71,7 @@ def guard(turn: str): def test_later_turn_refusal_names_the_earlier_turn_that_owes_closeout(tmp_path: Path) -> None: project, runtime, registry = _write_fixture(tmp_path) _configure_read_only_todo(project) + _initialize_fixture_authority(project, runtime, provider="sqlite") prior_turn = "owed-original" current_turn = "owed-current" diff --git a/tests/control_plane/test_quota_settlement_cli.py b/tests/control_plane/test_quota_settlement_cli.py index bfe0fbba30..174c2fd831 100644 --- a/tests/control_plane/test_quota_settlement_cli.py +++ b/tests/control_plane/test_quota_settlement_cli.py @@ -125,11 +125,13 @@ def _write_fixture( *, required_capability: str | None = None, claimed: bool = True, + provider: str | None = None, ) -> tuple[Path, Path, Path]: """Write unmigrated source input; canonical consumers must initialize authority. Use ``_initialize_fixture_authority`` after source edits for ordinary reads - or writes. Claim-authoring scenarios must also opt out of the initial claim. + or writes, or select ``provider`` when initial authoring is already complete. + Claim-authoring scenarios must also opt out of the initial claim. """ project = root / "project" runtime = root / "runtime" @@ -198,6 +200,8 @@ def _write_fixture( + "\n", encoding="utf-8", ) + if provider is not None: + _initialize_fixture_authority(project, runtime, provider=provider) return project, runtime, registry_path diff --git a/tests/control_plane/test_replan_semantic_action_behavior.py b/tests/control_plane/test_replan_semantic_action_behavior.py index 7597ffa0c0..d47997db3f 100644 --- a/tests/control_plane/test_replan_semantic_action_behavior.py +++ b/tests/control_plane/test_replan_semantic_action_behavior.py @@ -130,7 +130,7 @@ def _explore_context_action( context = channel["work_context"] assert context["complete"] is True read, = [source for source in context["sources"] - if source["kind"] == "explore_turn_context"] + if source.get("kind") == "explore_turn_context"] assert read["ordering"] == "before_work" assert read["content"]["goal_id"] == "replan-semantic-action-fixture" assert read["content"]["agent_id"] == "codex-replan-semantic-action" diff --git a/tests/control_plane_ts/protocol_action_packet_compat.test.ts b/tests/control_plane_ts/protocol_action_packet_compat.test.ts index c703b1a2f6..6ec2b2c73a 100644 --- a/tests/control_plane_ts/protocol_action_packet_compat.test.ts +++ b/tests/control_plane_ts/protocol_action_packet_compat.test.ts @@ -1,4 +1,5 @@ import assert from "node:assert/strict"; +import {createHash} from "node:crypto"; import {readFileSync} from "node:fs"; import test from "node:test"; import {interpretQuotaShouldRunPacket} from "../../loopx/control_plane/effect_program.ts"; @@ -7,12 +8,33 @@ import {buildTurnEnvelope, evaluateTurnEnvelope, quotaActionSignatureDocument, // Immutable release output; no current writer reconstructs the expected inputs. const {cases} = JSON.parse(readFileSync(new URL("../fixtures/protocol_packet_history_v0.json", import.meta.url), "utf8")); + +// Captured receipts are verified from their captured bytes, not re-created by +// the current writer, which now preserves complete execution contracts. +function capturedDigest(value: unknown): string { + function ordered(value: unknown): unknown { + if (Array.isArray(value)) return value.map(ordered); + if (value !== null && typeof value === "object") { + return Object.fromEntries(Object.entries(value).sort(([a], [b]) => a < b ? -1 : a > b ? 1 : 0) + .map(([key, item]) => [key, ordered(item)])); + } + return value; + } + return `sha256:${createHash("sha256").update(JSON.stringify(ordered(value))).digest("hex")}`; +} + for (const fixture of cases) { test(`release ${fixture.id}: supported absence or explicit retired-format rejection`, () => { const {decision, envelope} = structuredClone(fixture); const before = JSON.stringify({decision, envelope}); if (fixture.id === "missing_packet") { - assert.deepEqual(quotaActionSignatureDocument(decision), turnEnvelopeActionSignatureDocument(envelope)); + assert.equal(capturedDigest(turnEnvelopeActionSignatureDocument(envelope)), + envelope.action_signature.envelope_hash); + assert.equal(capturedDigest(decision), envelope.action_signature.source_decision_hash); + const current = buildTurnEnvelope({payload: decision, scheduler_execution_args: ""}); + assert.deepEqual((current.contract_capsule as Record).execution_obligation, decision.execution_obligation); + assert.deepEqual(quotaActionSignatureDocument(decision), turnEnvelopeActionSignatureDocument(current)); + assert.equal((current.action_signature as Record).matches, true); assert.equal(interpretQuotaShouldRunPacket(decision).observation.protocol_summary, null); } else { const error = /unsupported format: protocol_action_packet/; From 03e0549e7a2974543df7efc50131bcde26dcc26f Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Tue, 27 Oct 2026 15:14:27 +0800 Subject: [PATCH 2/9] test: prepare real authority for amendment and monitor consumers Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- tests/control_plane/test_goal_amendment_proposal.py | 9 +++++++++ tests/control_plane/test_quota_retry_monitor_journey.py | 8 ++++---- tests/control_plane/test_research_context_adoption.py | 2 +- 3 files changed, 14 insertions(+), 5 deletions(-) diff --git a/tests/control_plane/test_goal_amendment_proposal.py b/tests/control_plane/test_goal_amendment_proposal.py index 3395e41927..9d005ec004 100644 --- a/tests/control_plane/test_goal_amendment_proposal.py +++ b/tests/control_plane/test_goal_amendment_proposal.py @@ -26,6 +26,8 @@ import pytest +from canonical_authority_fixture import initialize_canonical_source_fixture + from loopx.control_plane.effect_runtime import ( EffectRuntimeRejected, effect_runtime_result, @@ -386,6 +388,13 @@ def _goal_state(goal_id: str, specs: list[dict[str, str]]) -> tuple[Path, str]: } if other_state_file is not None: paths["other_state_file"] = other_state_file + initialize_canonical_source_fixture( + runtime, GOAL_ID, state_path=state_file, provider="sqlite", + ) + if other_state_file is not None: + initialize_canonical_source_fixture( + runtime, OTHER_GOAL_ID, state_path=other_state_file, provider="sqlite", + ) return paths diff --git a/tests/control_plane/test_quota_retry_monitor_journey.py b/tests/control_plane/test_quota_retry_monitor_journey.py index 7a4c4e4422..c57c7abf67 100644 --- a/tests/control_plane/test_quota_retry_monitor_journey.py +++ b/tests/control_plane/test_quota_retry_monitor_journey.py @@ -12,6 +12,7 @@ from canonical_authority_fixture import ( initialize_canonical_authority, isolate_sqlite_runtime, + read_fixture_todos, ) from loopx.control_plane.coordination.runtime_shadow import ( @@ -62,13 +63,12 @@ def test_retry_monitor_guard_and_replay_keep_receipt_and_work_authority( def run(*args: str): return cli._run_cli(registry, runtime, *args, cwd=project) - code, listed = run("todo", "list", "--goal-id", cli.GOAL_ID) - assert code == 0, listed - for todo in listed["todos"]: + todos = read_fixture_todos(state) + for todo in todos: todo["claimed_by"] = cli.AGENT_ID initialize_canonical_authority( runtime, cli.GOAL_ID, build_todo_runtime_shadow_projection( - goal_id=cli.GOAL_ID, todos=listed["todos"], handoff_mode="soft_claim", leases=[], + goal_id=cli.GOAL_ID, todos=todos, handoff_mode="soft_claim", leases=[], ), state_path=state, provider=provider, ) code, mapped = run("read-only-map", "--goal-id", cli.GOAL_ID) diff --git a/tests/control_plane/test_research_context_adoption.py b/tests/control_plane/test_research_context_adoption.py index a440dd3031..8fcc072230 100644 --- a/tests/control_plane/test_research_context_adoption.py +++ b/tests/control_plane/test_research_context_adoption.py @@ -96,7 +96,7 @@ def run(*args): session_scope='agent',session_id='branch-session',session_profile_digest='synthetic-profile') # This owner-focused fixture substitutes only native history verification. # The real Host + CLI + canonical context journey lives in test_native_checkpoint. - monkeypatch.setattr('loopx.hosts.research_checkpoint._context_session', lambda *_: 'branch-session') + monkeypatch.setattr('loopx.hosts.research_checkpoint._context_session', lambda *_, **__: 'branch-session') before=run('todo','list','--goal-id','branch','--todo-id',destination_todo) args=dict(source_todo_id=source_todo,todo_id=destination_todo,operation_id='context-adoption', expected_revision=before['authority_read']['provider_revision'],lease_key='branch-lease',lease_version=1) From fb641ee18495c56ea0b29fe58108a4580203782f Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Tue, 27 Oct 2026 15:23:18 +0800 Subject: [PATCH 3/9] test: restore canonical input and peer-write setup for lifecycle consumers Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../test_fallback_wait_admission.py | 30 ++++++++++---- .../test_goal_amendment_proposal.py | 19 ++++----- .../test_goal_amendment_proposal_cli.py | 22 ++++++---- .../test_goal_amendment_proposal_lifecycle.py | 6 ++- .../test_monitor_unavailable_closeout_cli.py | 19 ++++++--- .../test_permanent_blocked_closeout_cli.py | 7 ++-- .../test_post_writeback_wait_cli.py | 6 +-- .../test_receipt_next_action_cli.py | 11 +++-- .../test_research_checkpoint_capture.py | 40 +++++++++++++++---- 9 files changed, 108 insertions(+), 52 deletions(-) diff --git a/tests/control_plane/test_fallback_wait_admission.py b/tests/control_plane/test_fallback_wait_admission.py index 2b7bf86ead..ebe8285fa8 100644 --- a/tests/control_plane/test_fallback_wait_admission.py +++ b/tests/control_plane/test_fallback_wait_admission.py @@ -10,7 +10,10 @@ from loopx.cli import main as cli_main from loopx.control_plane.goals.goal_vision import normalize_goal_vision_packet from loopx.control_plane.goals.goal_frontier import fallback_source -from canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime +from canonical_authority_fixture import ( + initialize_canonical_authority, initialize_canonical_source_fixture, + isolate_sqlite_runtime, read_fixture_todos, +) from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection from loopx.control_plane.coordination.local_authority import read_canonical_todos_if_promoted from loopx.control_plane.todos.todo_summary import TODO_ITEM_SCHEMA_VERSION @@ -90,6 +93,7 @@ def _write_fixture( unrelated_deferred_count: int, fallback_present: bool = True, prerequisite_status: str = "open", + provider: str | None = "sqlite", ) -> list[str]: state_file = tmp_path / "ACTIVE_GOAL_STATE.md" state_file.write_text( @@ -147,6 +151,8 @@ def _write_fixture( ) + "\n" ) + if provider is not None: + initialize_canonical_source_fixture(runtime, GOAL_ID, state_path=state_file, provider=provider) return [ "--format", "json", @@ -265,7 +271,7 @@ def test_real_cli_reads_only_direct_fallback_dependencies( monkeypatch: pytest.MonkeyPatch, depth: int, ) -> None: - args = _write_fixture(tmp_path, unrelated_deferred_count=20) + args = _write_fixture(tmp_path, unrelated_deferred_count=20, provider=None) state_file = tmp_path / "ACTIVE_GOAL_STATE.md" state = state_file.read_text() if depth: @@ -281,6 +287,9 @@ def test_real_cli_reads_only_direct_fallback_dependencies( f"resume_when=todo_done:todo_chain_{index + 1} -->\n" ) state_file.write_text(state) + initialize_canonical_source_fixture( + tmp_path / "runtime", GOAL_ID, state_path=state_file, provider="sqlite", + ) reads: list[str] = [] original_read = fallback_source.read_fallback_source_snapshot @@ -371,7 +380,7 @@ def test_real_cli_fallback_monitor_wait_requires_generation_condition( generation: int | None, expected_gap: str | None, ) -> None: - args = _write_fixture(tmp_path, unrelated_deferred_count=0) + args = _write_fixture(tmp_path, unrelated_deferred_count=0, provider=None) state_file = tmp_path / "ACTIVE_GOAL_STATE.md" generation_metadata = ( f" resume_monitor_generation={generation}" if generation is not None else "" @@ -391,6 +400,9 @@ def test_real_cli_fallback_monitor_wait_requires_generation_condition( "material_change_generation=3 -->\n" ) state_file.write_text(state) + initialize_canonical_source_fixture( + tmp_path / "runtime", GOAL_ID, state_path=state_file, provider="sqlite", + ) assert cli_main(args) == 0 result = json.loads(capsys.readouterr().out) @@ -408,7 +420,7 @@ def test_real_cli_ambiguous_canonical_fallback_read_is_uncertain( capsys: pytest.CaptureFixture[str], duplicate_id: str, ) -> None: - args = _write_fixture(tmp_path, unrelated_deferred_count=0) + args = _write_fixture(tmp_path, unrelated_deferred_count=0, provider=None) state_file = tmp_path / "ACTIVE_GOAL_STATE.md" state_file.write_text( state_file.read_text() @@ -416,6 +428,10 @@ def test_real_cli_ambiguous_canonical_fallback_read_is_uncertain( + f" \n" ) + initialize_canonical_source_fixture( + tmp_path / "runtime", GOAL_ID, state_path=state_file, provider="sqlite", + ) + # The canonical succession gate rejects duplicate identities before a # frontier can be projected. Do not fabricate fallback advice from it. assert cli_main(args) == 1 @@ -437,12 +453,10 @@ def test_real_provider_cli_uses_one_snapshot_including_archived_dependency( if provider == "sqlite": isolate_sqlite_runtime(tmp_path, monkeypatch) args = _write_fixture(tmp_path, unrelated_deferred_count=20, - prerequisite_status=dependency_status) + prerequisite_status=dependency_status, provider=None) state_file = tmp_path / "ACTIVE_GOAL_STATE.md" runtime_root = tmp_path / "runtime" - rows = fallback_source.read_fallback_source_snapshot( - registry_path=tmp_path / "registry.json", runtime_root=runtime_root, goal_id=GOAL_ID, - ) + rows = read_fixture_todos(state_file) for row in rows: row["schema_version"] = TODO_ITEM_SCHEMA_VERSION if dependency_status == "done": diff --git a/tests/control_plane/test_goal_amendment_proposal.py b/tests/control_plane/test_goal_amendment_proposal.py index 9d005ec004..eb10d1aef7 100644 --- a/tests/control_plane/test_goal_amendment_proposal.py +++ b/tests/control_plane/test_goal_amendment_proposal.py @@ -291,6 +291,7 @@ def _write_fixture( with_other_goal: bool = False, runs: Any = _DEFAULT_RUNS, other_goal_runs: Any = _DEFAULT_RUNS, + provider: str = "sqlite", ) -> dict[str, Path]: project = root / "project" runtime = root / "runtime" @@ -389,11 +390,11 @@ def _goal_state(goal_id: str, specs: list[dict[str, str]]) -> tuple[Path, str]: if other_state_file is not None: paths["other_state_file"] = other_state_file initialize_canonical_source_fixture( - runtime, GOAL_ID, state_path=state_file, provider="sqlite", + runtime, GOAL_ID, state_path=state_file, provider=provider, ) if other_state_file is not None: initialize_canonical_source_fixture( - runtime, OTHER_GOAL_ID, state_path=other_state_file, provider="sqlite", + runtime, OTHER_GOAL_ID, state_path=other_state_file, provider=provider, ) return paths @@ -1031,16 +1032,14 @@ def test_markdown_basis_zero_from_real_projection_admits(tmp_path: Path) -> None assert record["base_state_event_basis_sequence"] == 0 -def test_markdown_basis_with_fabricated_positive_sequence_fails_closed( +def test_canonical_basis_with_fabricated_positive_sequence_fails_closed( tmp_path: Path, ) -> None: - # The inverse counterexample: with sequence fabricated to 5 the old - # decoder happily admitted the proposal as unverifiable. 0 is the only - # markdown basis the Stage 1 producer can emit, so any other value is - # not a producible base and must fail closed instead of being retained. + # A canonical snapshot has no event append sequence. Fabricating a + # positive sequence must fail closed instead of retaining the proposal. paths = _write_fixture(tmp_path) - with pytest.raises(ValueError, match="markdown_active_state"): + with pytest.raises(ValueError, match="canonical_todo_snapshot"): _admit(paths, _proposal(paths, {"base_state_event_basis_sequence": 5})) assert _journal_rows(paths) == [] @@ -1206,7 +1205,9 @@ def test_corrupt_journal_line_fails_closed_and_retains_nothing( def test_retention_does_not_advance_the_derived_head(tmp_path: Path) -> None: - paths = _write_fixture(tmp_path) + # File provides the byte-stable oracle; SQLite read connections may create + # WAL/SHM sidecars without changing the committed authority. + paths = _write_fixture(tmp_path, provider="file") before_tree = _canonical_tree_snapshot(paths) first = _admit(paths, _proposal(paths)) diff --git a/tests/control_plane/test_goal_amendment_proposal_cli.py b/tests/control_plane/test_goal_amendment_proposal_cli.py index f58b64c5f7..d78db864dc 100644 --- a/tests/control_plane/test_goal_amendment_proposal_cli.py +++ b/tests/control_plane/test_goal_amendment_proposal_cli.py @@ -138,6 +138,10 @@ def _registry(common_runtime_root: Path) -> str: "registry_b": registry_b, } _append_runs(paths, _stall_runs()) + from canonical_authority_fixture import initialize_canonical_source_fixture + initialize_canonical_source_fixture( + runtime_a, GOAL_ID, state_path=state_file, provider="sqlite", + ) return paths @@ -200,7 +204,7 @@ def test_cli_submits_proposal_and_lists_journal_json( assert list_payload["rows"][0]["journal_append_sequence"] == 1 -def test_cli_does_not_use_current_directory_when_registered_state_is_missing( +def test_cli_does_not_use_current_directory_when_registered_authority_is_missing( tmp_path: Path, capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch, @@ -212,6 +216,7 @@ def test_cli_does_not_use_current_directory_when_registered_state_is_missing( registry = json.loads(paths["registry"].read_text(encoding="utf-8")) goal = next(item for item in registry["goals"] if item["id"] == GOAL_ID) goal["repo"] = str(tmp_path / "missing-registered-project") + registry["common_runtime_root"] = str(tmp_path / "missing-registered-runtime") paths["registry"].write_text(json.dumps(registry), encoding="utf-8") # A different current directory happens to contain a same-named Goal @@ -244,7 +249,7 @@ def test_cli_does_not_use_current_directory_when_registered_state_is_missing( ) assert exit_code == 1 assert payload["ok"] is False - assert "goal state file is missing" in payload["error"] + assert "Canonical authority is required" in payload["error"] assert not journal.exists() @@ -344,7 +349,7 @@ def test_cli_single_registry_submit_and_list_positive_control( assert list_payload["rows"][0]["proposal_id"] == "gap_stage2_001" -def test_cli_legacy_work_change_requires_rebase_and_preserves_original_receipt( +def test_cli_canonical_work_change_requires_rebase_and_preserves_original_receipt( tmp_path: Path, capsys: pytest.CaptureFixture[str], ) -> None: paths = _write_fixture(tmp_path) @@ -362,11 +367,12 @@ def submit(value): first = submit(proposal) assert first["admission"] == "admitted" - state = paths["state_file"].read_text(encoding="utf-8") - needle = "todo_id=todo_stage2_b " - assert needle in state - paths["state_file"].write_text(state.replace(needle, needle + "claimed_by=agent-b "), - encoding="utf-8") + code, claimed, _ = _run_amendment_cli( + capsys, paths["registry"], "todo", "claim", "--goal-id", GOAL_ID, + "--todo-id", "todo_stage2_b", "--agent-id", "agent-b", + "--claimed-by", "agent-b", "--execute", "--format", "json", + ) + assert code == 0 and claimed["claimed"] is True, claimed journal = paths["runtime"] / "goals" / GOAL_ID / "amendment-proposals" / "journal.jsonl" original_bytes = journal.read_bytes() assert submit(proposal) == first diff --git a/tests/control_plane/test_goal_amendment_proposal_lifecycle.py b/tests/control_plane/test_goal_amendment_proposal_lifecycle.py index e67e745293..49db736af2 100644 --- a/tests/control_plane/test_goal_amendment_proposal_lifecycle.py +++ b/tests/control_plane/test_goal_amendment_proposal_lifecycle.py @@ -45,7 +45,7 @@ def _write_fixture(root: Path) -> tuple[Path, Path, Path]: """One isolated, healthy synthetic Goal: active, one registered agent, - one open Todo, an empty runtime (no event log, no run rows).""" + one open canonical Todo and no prior run rows.""" project = root / "project" runtime = root / "runtime" @@ -109,6 +109,10 @@ def _write_fixture(root: Path) -> tuple[Path, Path, Path]: + "\n", encoding="utf-8", ) + from canonical_authority_fixture import initialize_canonical_source_fixture + initialize_canonical_source_fixture( + runtime, GOAL_ID, state_path=state_path, provider="sqlite", + ) return project, runtime, registry_path diff --git a/tests/control_plane/test_monitor_unavailable_closeout_cli.py b/tests/control_plane/test_monitor_unavailable_closeout_cli.py index e738ec3c5e..32a8ae923e 100644 --- a/tests/control_plane/test_monitor_unavailable_closeout_cli.py +++ b/tests/control_plane/test_monitor_unavailable_closeout_cli.py @@ -8,7 +8,7 @@ import pytest -from canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime +from canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime, read_fixture_todos from loopx.cli import main as cli_main from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection from test_quota_settlement_cli import ( @@ -57,11 +57,20 @@ def cli(*args: str) -> tuple[int, dict]: "--format", "json", *args]) return rc, json.loads(capsys.readouterr().out) - rc, listed = cli("todo", "list", "--goal-id", GOAL_ID) - assert rc == 0, listed + if provider == "markdown": + before = state.read_bytes() + rc, refused = cli("todo", "list", "--goal-id", GOAL_ID) + assert rc == 1 and refused["error_code"] == "canonical_authority_required", refused + assert refused["write_admitted"] is False + assert refused["legacy_fallback_used"] is False + assert state.read_bytes() == before + assert _classification_count(runtime, "monitor_observation_unavailable") == 0 + assert _spend_run_count(runtime) == 0 + return + source_todos = read_fixture_todos(state) if provider != "markdown": initialize_canonical_authority(runtime, GOAL_ID, build_todo_runtime_shadow_projection( - goal_id=GOAL_ID, todos=listed["todos"], handoff_mode="hard_lease", leases=[], + goal_id=GOAL_ID, todos=source_todos, handoff_mode="hard_lease", leases=[], ), state_path=state, provider=provider) original_turn = f"unavailable-original-{provider}" binding = ("--agent-id", AGENT_ID, "--todo-id", MONITOR_ID, @@ -223,7 +232,7 @@ def actor_metadata_request(request): monitor = next(todo for todo in after_late["todos"] if todo["todo_id"] == MONITOR_ID) assert {k: monitor.get(k) for k in OBSERVATION_FIELDS} == fields assert {todo["todo_id"] for todo in after_late["todos"]} == { - todo["todo_id"] for todo in listed["todos"] + todo["todo_id"] for todo in source_todos } assert _classification_count(runtime, "monitor_observation_unavailable") == 1 assert _classification_count(runtime, "quota_monitor_poll") == 0 diff --git a/tests/control_plane/test_permanent_blocked_closeout_cli.py b/tests/control_plane/test_permanent_blocked_closeout_cli.py index ac25b92119..b5e9e65d0f 100644 --- a/tests/control_plane/test_permanent_blocked_closeout_cli.py +++ b/tests/control_plane/test_permanent_blocked_closeout_cli.py @@ -6,7 +6,7 @@ import pytest -from canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime +from canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime, read_fixture_todos from loopx.cli import main as cli_main from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection from test_quota_settlement_cli import ( @@ -34,10 +34,9 @@ def cli(*args: str) -> tuple[int, dict]: "--format", "json", *args]) return rc, json.loads(capsys.readouterr().out) - rc, listed = cli("todo", "list", "--goal-id", GOAL_ID) - assert rc == 0, listed + source_todos = read_fixture_todos(state) initialize_canonical_authority(runtime, GOAL_ID, build_todo_runtime_shadow_projection( - goal_id=GOAL_ID, todos=listed["todos"], handoff_mode="hard_lease", leases=[], + goal_id=GOAL_ID, todos=source_todos, handoff_mode="hard_lease", leases=[], ), state_path=state, provider=provider) binding = ("--agent-id", AGENT_ID, "--todo-id", TODO_ID, "--turn-instance-id", f"blocked-lifecycle-{provider}") diff --git a/tests/control_plane/test_post_writeback_wait_cli.py b/tests/control_plane/test_post_writeback_wait_cli.py index ee84d2ff6a..881f2a1ce3 100644 --- a/tests/control_plane/test_post_writeback_wait_cli.py +++ b/tests/control_plane/test_post_writeback_wait_cli.py @@ -12,6 +12,7 @@ from canonical_authority_fixture import ( initialize_canonical_authority, isolate_sqlite_runtime, + read_fixture_todos, ) from loopx.cli import main as cli_main from loopx.control_plane.coordination.runtime_shadow import ( @@ -119,14 +120,13 @@ def run(*args: str) -> tuple[int, dict]: ) return rc, json.loads(capsys.readouterr().out) - rc, listed = run("todo", "list", "--goal-id", GOAL_ID) - assert rc == 0, listed + source_todos = read_fixture_todos(state) initialize_canonical_authority( runtime, GOAL_ID, build_todo_runtime_shadow_projection( goal_id=GOAL_ID, - todos=listed["todos"], + todos=source_todos, handoff_mode="soft_claim", leases=[], ), diff --git a/tests/control_plane/test_receipt_next_action_cli.py b/tests/control_plane/test_receipt_next_action_cli.py index cedef2cc65..e513fe2b5b 100644 --- a/tests/control_plane/test_receipt_next_action_cli.py +++ b/tests/control_plane/test_receipt_next_action_cli.py @@ -9,6 +9,7 @@ import test_quota_settlement_cli as fixture from canonical_authority_fixture import ( initialize_canonical_authority, isolate_sqlite_runtime, + read_fixture_todos, without_resume_observation_time as stable, ) @@ -44,10 +45,9 @@ def run(*args: str) -> tuple[int, dict]: "--format", "json", *args]) return code, json.loads(capsys.readouterr().out) - code, initial = run("todo", "list", "--goal-id", fixture.GOAL_ID) - assert code == 0, initial + source_todos = read_fixture_todos(state) canonical = initialize_canonical_authority(runtime, fixture.GOAL_ID, - build_todo_runtime_shadow_projection(goal_id=fixture.GOAL_ID, todos=initial["todos"], + build_todo_runtime_shadow_projection(goal_id=fixture.GOAL_ID, todos=source_todos, handoff_mode="soft_claim", leases=[]), state_path=state, provider=provider) binding = ("--agent-id", fixture.AGENT_ID, "--todo-id", fixture.TODO_ID, "--turn-instance-id", "original-next-action-turn") @@ -137,11 +137,10 @@ def test_todoless_replan_refuses_step_and_recovers_original_settlement( def run(*args: str) -> tuple[int, dict]: return fixture._run_cli(registry, runtime, *args, cwd=project) - code, listed = run("todo", "list", "--goal-id", fixture.GOAL_ID) - assert code == 0, listed + source_todos = read_fixture_todos(state) initialize_canonical_authority(runtime, fixture.GOAL_ID, build_todo_runtime_shadow_projection(goal_id=fixture.GOAL_ID, - todos=listed["todos"], handoff_mode="soft_claim", leases=[]), + todos=source_todos, handoff_mode="soft_claim", leases=[]), state_path=state, provider=provider) turn = "replan-recommendation-recovery" guard_args = ("quota", "should-run", "--codex-app", "--goal-id", fixture.GOAL_ID, diff --git a/tests/control_plane/test_research_checkpoint_capture.py b/tests/control_plane/test_research_checkpoint_capture.py index bfaf87bd51..12282ddf80 100644 --- a/tests/control_plane/test_research_checkpoint_capture.py +++ b/tests/control_plane/test_research_checkpoint_capture.py @@ -7,7 +7,10 @@ import time import test_quota_settlement_cli as cli -from canonical_authority_fixture import isolate_sqlite_runtime +from canonical_authority_fixture import ( + advance_canonical_fixture_authority, isolate_sqlite_runtime, read_fixture_todos, +) +from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection from benchmark.runtime.checkpoint import capture_after_wake, capture_before_replan from loopx.control_plane.effect_runtime import effect_runtime_result @@ -57,6 +60,7 @@ def test_worker_keeps_work_running_when_staged_capture_declaration_fails( project, runtime, registry = cli._write_fixture(tmp_path) if trigger: cli._configure_autonomous_replan_fixture(project, runtime, registry) + cli._initialize_fixture_authority(project, runtime, provider="sqlite") skills = tmp_path / "skills" skills.mkdir() binary = tmp_path / "codex" @@ -134,7 +138,7 @@ def denied_read(path): def test_real_cli_settlement_capture_verification_and_inert_materialization( tmp_path, monkeypatch, opportunity, corrupt, collisions): isolate_sqlite_runtime(tmp_path, monkeypatch) - project, runtime, registry = cli._write_fixture(tmp_path) + project, runtime, registry = cli._write_fixture(tmp_path, provider="sqlite") binding = ("--agent-id", cli.AGENT_ID, "--todo-id", cli.TODO_ID, "--turn-instance-id", cli.TURN_ID) def run(*args): code, result = cli._run_cli(registry, runtime, *args) @@ -250,7 +254,7 @@ def test_real_native_cursor_must_match_sealed_store_basis(tmp_path, monkeypatch, from loopx.control_plane.turn_driver.codex_sessions import _store_codex_cli_session isolate_sqlite_runtime(tmp_path, monkeypatch) - project, runtime, registry = cli._write_fixture(tmp_path) + project, runtime, registry = cli._write_fixture(tmp_path, provider="sqlite") binding = ("--agent-id", cli.AGENT_ID, "--todo-id", cli.TODO_ID, "--turn-instance-id", cli.TURN_ID) def run(*args): code, result = cli._run_cli(registry, runtime, *args, cwd=project) @@ -401,7 +405,7 @@ def test_real_pre_replan_capture_is_before_decision_and_same_node_is_not_recaptu import sys isolate_sqlite_runtime(tmp_path, monkeypatch) - project, runtime, registry = cli._write_fixture(tmp_path) + project, runtime, registry = cli._write_fixture(tmp_path, provider="sqlite") binding = ("--agent-id", cli.AGENT_ID, "--todo-id", cli.TODO_ID, "--turn-instance-id", cli.TURN_ID) def run(*args): @@ -432,6 +436,13 @@ def run(*args): f" \n")) # This helper supplies synthetic repeated observations; retain the real # source settlement history instead of replacing its authoritative readback. + advance_canonical_fixture_authority( + runtime, cli.GOAL_ID, build_todo_runtime_shadow_projection( + goal_id=cli.GOAL_ID, todos=read_fixture_todos( + project / f".codex/goals/{cli.GOAL_ID}/ACTIVE_GOAL_STATE.md"), + handoff_mode="soft_claim", leases=[], + ), operation_id="fixture-replan-source", + ) index.write_bytes(original_settlement + index.read_bytes()) logs = tmp_path / "logs" wakes = logs / "wakes" @@ -524,10 +535,9 @@ def authority_head(root, advance=False): project, runtime, registry = cli._write_fixture(tmp_path) source_ref = capture_first_party_host_goal_ref(registry_path=registry, goal_id=cli.GOAL_ID) state = project / f'.codex/goals/{cli.GOAL_ID}/ACTIVE_GOAL_STATE.md' - code, listed = cli._run_cli(registry, runtime, 'todo', 'list', '--goal-id', cli.GOAL_ID) - assert code == 0, listed + source_todos = read_fixture_todos(state) initialize_canonical_authority(runtime, cli.GOAL_ID, - build_todo_runtime_shadow_projection(goal_id=cli.GOAL_ID, todos=listed['todos'], + build_todo_runtime_shadow_projection(goal_id=cli.GOAL_ID, todos=source_todos, handoff_mode='hard_lease', leases=[]), state_path=state, provider=provider) def run(*args): @@ -707,7 +717,7 @@ def test_failed_replan_index_cannot_relabel_later_work_after_permissions_recover if os.geteuid() == 0: pytest.skip("Real POSIX permission denial requires an unprivileged test user") isolate_sqlite_runtime(tmp_path, monkeypatch) - project, runtime, registry = cli._write_fixture(tmp_path) + project, runtime, registry = cli._write_fixture(tmp_path, provider="sqlite") skills = tmp_path / "skills" skills.mkdir() binary = tmp_path / "codex" @@ -747,6 +757,13 @@ def run(*args): index = runtime / "goals" / cli.GOAL_ID / "runs/index.jsonl" original = index.read_bytes() cli._configure_autonomous_replan_fixture(project, runtime, registry) + advance_canonical_fixture_authority( + runtime, cli.GOAL_ID, build_todo_runtime_shadow_projection( + goal_id=cli.GOAL_ID, todos=read_fixture_todos( + project / f".codex/goals/{cli.GOAL_ID}/ACTIVE_GOAL_STATE.md"), + handoff_mode="soft_claim", leases=[], + ), operation_id="fixture-later-replan-0", + ) index.write_bytes(original + index.read_bytes()) solution = project / "solution.py" solution.write_text("retained_candidate = 7\n") @@ -807,6 +824,13 @@ def run(*args): from datetime import datetime, timedelta, timezone source_history = index.read_bytes() cli._configure_autonomous_replan_fixture(project, runtime, registry) + advance_canonical_fixture_authority( + runtime, cli.GOAL_ID, build_todo_runtime_shadow_projection( + goal_id=cli.GOAL_ID, todos=read_fixture_todos( + project / f".codex/goals/{cli.GOAL_ID}/ACTIVE_GOAL_STATE.md"), + handoff_mode="soft_claim", leases=[], + ), operation_id="fixture-later-replan-1", + ) observations = [json.loads(line) for line in index.read_text().splitlines()] observed_at = datetime.now(timezone.utc) for number, row in enumerate(observations): From aa53f1761f14a1eb588971a6696022647e31e798 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Tue, 27 Oct 2026 15:35:48 +0800 Subject: [PATCH 4/9] test: bind fixture routes and Host wake evidence to canonical owners Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../test_fallback_wait_admission.py | 19 +++++++++++-------- .../test_goal_amendment_proposal.py | 2 ++ .../test_goal_amendment_proposal_cli.py | 2 +- .../test_goal_amendment_proposal_lifecycle.py | 6 +++--- .../test_monitor_unavailable_closeout_cli.py | 2 +- .../test_research_checkpoint_capture.py | 2 +- 6 files changed, 19 insertions(+), 14 deletions(-) diff --git a/tests/control_plane/test_fallback_wait_admission.py b/tests/control_plane/test_fallback_wait_admission.py index ebe8285fa8..bbd07556b9 100644 --- a/tests/control_plane/test_fallback_wait_admission.py +++ b/tests/control_plane/test_fallback_wait_admission.py @@ -8,6 +8,7 @@ import pytest from loopx.cli import main as cli_main +from loopx.control_plane.effect_runtime import EffectRuntimeRejected from loopx.control_plane.goals.goal_vision import normalize_goal_vision_packet from loopx.control_plane.goals.goal_frontier import fallback_source from canonical_authority_fixture import ( @@ -415,7 +416,7 @@ def test_real_cli_fallback_monitor_wait_requires_generation_condition( @pytest.mark.parametrize("duplicate_id", [FALLBACK_TODO_ID, PREREQUISITE_TODO_ID]) -def test_real_cli_ambiguous_canonical_fallback_read_is_uncertain( +def test_duplicate_authored_fallback_is_rejected_before_canonical_creation( tmp_path: Path, capsys: pytest.CaptureFixture[str], duplicate_id: str, @@ -428,17 +429,19 @@ def test_real_cli_ambiguous_canonical_fallback_read_is_uncertain( + f" \n" ) - initialize_canonical_source_fixture( - tmp_path / "runtime", GOAL_ID, state_path=state_file, provider="sqlite", - ) - - # The canonical succession gate rejects duplicate identities before a - # frontier can be projected. Do not fabricate fallback advice from it. + before = state_file.read_bytes() + with pytest.raises(EffectRuntimeRejected, match="duplicate succession identity"): + initialize_canonical_source_fixture( + tmp_path / "runtime", GOAL_ID, state_path=state_file, provider="sqlite", + ) + assert state_file.read_bytes() == before + # Invalid authored input never becomes authority. The actual CLI cannot + # fabricate fallback advice from the uninitialized source afterward. assert cli_main(args) == 1 result = json.loads(capsys.readouterr().out) assert result["should_run"] is False assert result["state"] == "blocked_health" - assert "duplicate succession identity" in result["verbose_debug"]["error"] + assert "canonical_authority_required" in json.dumps(result) assert "goal_frontier_projection" not in result diff --git a/tests/control_plane/test_goal_amendment_proposal.py b/tests/control_plane/test_goal_amendment_proposal.py index eb10d1aef7..98c82ecbad 100644 --- a/tests/control_plane/test_goal_amendment_proposal.py +++ b/tests/control_plane/test_goal_amendment_proposal.py @@ -425,6 +425,8 @@ def _derived_source_basis(paths: dict[str, Path]) -> dict[str, object]: goal_id=GOAL_ID, agent_id="agent-a", project=paths["project"], + registry_path=paths["registry"], + runtime_root=paths["runtime"], ) basis = alignment["source_basis"] assert isinstance(basis, dict) diff --git a/tests/control_plane/test_goal_amendment_proposal_cli.py b/tests/control_plane/test_goal_amendment_proposal_cli.py index d78db864dc..d81a9e01c2 100644 --- a/tests/control_plane/test_goal_amendment_proposal_cli.py +++ b/tests/control_plane/test_goal_amendment_proposal_cli.py @@ -370,7 +370,7 @@ def submit(value): code, claimed, _ = _run_amendment_cli( capsys, paths["registry"], "todo", "claim", "--goal-id", GOAL_ID, "--todo-id", "todo_stage2_b", "--agent-id", "agent-b", - "--claimed-by", "agent-b", "--execute", "--format", "json", + "--claimed-by", "agent-b", "--format", "json", ) assert code == 0 and claimed["claimed"] is True, claimed journal = paths["runtime"] / "goals" / GOAL_ID / "amendment-proposals" / "journal.jsonl" diff --git a/tests/control_plane/test_goal_amendment_proposal_lifecycle.py b/tests/control_plane/test_goal_amendment_proposal_lifecycle.py index 49db736af2..ee85ee9b25 100644 --- a/tests/control_plane/test_goal_amendment_proposal_lifecycle.py +++ b/tests/control_plane/test_goal_amendment_proposal_lifecycle.py @@ -245,8 +245,8 @@ def test_production_quota_obligation_survives_the_full_amendment_lifecycle( assert settlement_identity["turn_instance_id"] == TURN_ID assert settlement_identity["todo_id"] == TODO_ID - # 3. The proposal binds the real Stage 1 basis (markdown goal, so the - # real sequence is 0) and the quota-returned obligation, verbatim. + # 3. The proposal binds the real canonical snapshot (no event append + # sequence) and the quota-returned obligation, verbatim. align_rc, alignment = _run_cli( registry_path, runtime, @@ -260,7 +260,7 @@ def test_production_quota_obligation_survives_the_full_amendment_lifecycle( ) assert align_rc == 0, alignment basis = alignment["source_basis"] - assert basis["revision_basis"] == "markdown_active_state" + assert basis["revision_basis"] == "canonical_todo_snapshot" assert basis["state_event_basis_sequence"] == 0 proposal = { diff --git a/tests/control_plane/test_monitor_unavailable_closeout_cli.py b/tests/control_plane/test_monitor_unavailable_closeout_cli.py index 32a8ae923e..e2ca77178b 100644 --- a/tests/control_plane/test_monitor_unavailable_closeout_cli.py +++ b/tests/control_plane/test_monitor_unavailable_closeout_cli.py @@ -319,7 +319,7 @@ def actor_metadata_request(request): def test_unavailable_closeout_does_not_accept_missing_evidence_or_wrong_identity( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], invalid: str, ) -> None: - project, runtime, registry = _write_fixture(tmp_path) + project, runtime, registry = _write_fixture(tmp_path, provider="sqlite") # This guard binds an advancement. A caller cannot close another Monitor # or omit the typed evidence required by the existing settlement owner. def cli(*args: str) -> tuple[int, dict]: diff --git a/tests/control_plane/test_research_checkpoint_capture.py b/tests/control_plane/test_research_checkpoint_capture.py index 12282ddf80..8452050d37 100644 --- a/tests/control_plane/test_research_checkpoint_capture.py +++ b/tests/control_plane/test_research_checkpoint_capture.py @@ -446,7 +446,7 @@ def run(*args): index.write_bytes(original_settlement + index.read_bytes()) logs = tmp_path / "logs" wakes = logs / "wakes" - prior = wakes / "wake-100-prior" + prior = wakes / cli.TURN_ID prior.mkdir(parents=True) prior_receipt = {"turn_id": cli.TURN_ID, "mode": "heartbeat", "stage": "execute", "ok": True} (prior / "receipt.json").write_text(json.dumps(prior_receipt)) From 960edd5bebf63aa65445674589b9bd5604d2c6e1 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Tue, 27 Oct 2026 15:38:56 +0800 Subject: [PATCH 5/9] test: model original Host wake identity in checkpoint fixtures Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- tests/control_plane/test_goal_amendment_proposal_cli.py | 2 +- tests/control_plane/test_research_checkpoint_capture.py | 7 ++++--- 2 files changed, 5 insertions(+), 4 deletions(-) diff --git a/tests/control_plane/test_goal_amendment_proposal_cli.py b/tests/control_plane/test_goal_amendment_proposal_cli.py index d81a9e01c2..4628310a75 100644 --- a/tests/control_plane/test_goal_amendment_proposal_cli.py +++ b/tests/control_plane/test_goal_amendment_proposal_cli.py @@ -372,7 +372,7 @@ def submit(value): "--todo-id", "todo_stage2_b", "--agent-id", "agent-b", "--claimed-by", "agent-b", "--format", "json", ) - assert code == 0 and claimed["claimed"] is True, claimed + assert code == 0 and claimed["ok"] is True, claimed journal = paths["runtime"] / "goals" / GOAL_ID / "amendment-proposals" / "journal.jsonl" original_bytes = journal.read_bytes() assert submit(proposal) == first diff --git a/tests/control_plane/test_research_checkpoint_capture.py b/tests/control_plane/test_research_checkpoint_capture.py index 8452050d37..0435610fe7 100644 --- a/tests/control_plane/test_research_checkpoint_capture.py +++ b/tests/control_plane/test_research_checkpoint_capture.py @@ -406,7 +406,8 @@ def test_real_pre_replan_capture_is_before_decision_and_same_node_is_not_recaptu isolate_sqlite_runtime(tmp_path, monkeypatch) project, runtime, registry = cli._write_fixture(tmp_path, provider="sqlite") - binding = ("--agent-id", cli.AGENT_ID, "--todo-id", cli.TODO_ID, "--turn-instance-id", cli.TURN_ID) + source_turn_id = "wake-100-prior" + binding = ("--agent-id", cli.AGENT_ID, "--todo-id", cli.TODO_ID, "--turn-instance-id", source_turn_id) def run(*args): code, result = cli._run_cli(registry, runtime, *args) @@ -446,9 +447,9 @@ def run(*args): index.write_bytes(original_settlement + index.read_bytes()) logs = tmp_path / "logs" wakes = logs / "wakes" - prior = wakes / cli.TURN_ID + prior = wakes / source_turn_id prior.mkdir(parents=True) - prior_receipt = {"turn_id": cli.TURN_ID, "mode": "heartbeat", "stage": "execute", "ok": True} + prior_receipt = {"turn_id": source_turn_id, "mode": "heartbeat", "stage": "execute", "ok": True} (prior / "receipt.json").write_text(json.dumps(prior_receipt)) solution = project / "solution.py" solution.write_text("retained_candidate = 7\n") From dc1f09c339f8847e43f0c50537957cf70616af49 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Tue, 27 Oct 2026 15:53:04 +0800 Subject: [PATCH 6/9] test: initialize checkpoint and planning consumers on real authority Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../test_checkpoint_provider_fence.py | 6 ++--- .../test_checkpoint_read_context.py | 27 ++++++++++--------- .../test_refresh_checkpoint_isolation.py | 8 +++++- .../test_refresh_checkpoint_recovery.py | 18 +++++-------- tests/control_plane/test_task_planning.py | 22 ++++++++++++++- 5 files changed, 52 insertions(+), 29 deletions(-) diff --git a/tests/control_plane/test_checkpoint_provider_fence.py b/tests/control_plane/test_checkpoint_provider_fence.py index 0cbdbdbbab..bc29c39b72 100644 --- a/tests/control_plane/test_checkpoint_provider_fence.py +++ b/tests/control_plane/test_checkpoint_provider_fence.py @@ -9,7 +9,7 @@ import time import pytest -from canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime +from canonical_authority_fixture import advance_canonical_fixture_authority, isolate_sqlite_runtime from loopx.control_plane.coordination.local_authority import read_canonical_todos_if_promoted from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection from loopx.control_plane.effect_runtime import EffectRuntimeConflict @@ -94,14 +94,14 @@ def _replace_source_goal(registry: Path, instance_id: str) -> None: def fixture(tmp_path, monkeypatch, provider): isolate_sqlite_runtime(tmp_path, monkeypatch) - project, runtime, registry, binding, delivery, original = _missing(tmp_path) + project, runtime, registry, binding, delivery, original = _missing(tmp_path, provider=provider) state = project / f".codex/goals/{GOAL_ID}/ACTIVE_GOAL_STATE.md" todos = [{"schema_version": "todo_item_v0", "todo_id": name, "index": index, "role": "agent", "status": "open", "done": False, "text": f"Synthetic page work {name}", "task_class": "advancement_task", "archive_state": "active", "source_section": "Agent Todo", "claimed_by": AGENT_ID} for index, name in enumerate((TODO_ID, "todo_dependency", "todo_unrelated"), 1)] projection = build_todo_runtime_shadow_projection(goal_id=GOAL_ID, todos=todos, handoff_mode="hard_lease", leases=[]) - initialize_canonical_authority(runtime, GOAL_ID, projection, state_path=state, provider=provider) + advance_canonical_fixture_authority(runtime, GOAL_ID, projection, operation_id="fixture-checkpoint-fences") for todo in todos: rc, lease = _run_cli(registry, runtime, "task-lease", "acquire", "--goal-id", GOAL_ID, "--todo-id", todo["todo_id"], "--owner", AGENT_ID, diff --git a/tests/control_plane/test_checkpoint_read_context.py b/tests/control_plane/test_checkpoint_read_context.py index db854ea4cc..3e21a89eec 100644 --- a/tests/control_plane/test_checkpoint_read_context.py +++ b/tests/control_plane/test_checkpoint_read_context.py @@ -14,8 +14,8 @@ ) -def _missing(root: Path): - project, runtime, registry = _write_fixture(root) +def _missing(root: Path, *, provider: str = "sqlite"): + project, runtime, registry = _write_fixture(root, provider=provider) binding = ("--goal-id", GOAL_ID, "--agent-id", AGENT_ID, "--todo-id", TODO_ID, "--turn-instance-id", TURN_ID) for args in ( ("refresh-state", "--goal-id", GOAL_ID, "--agent-id", AGENT_ID, @@ -34,7 +34,7 @@ def _missing(root: Path): def test_checkpoint_recovery_is_not_a_fresh_turn_preflight(tmp_path): - project, runtime, registry = _write_fixture(tmp_path) + project, runtime, registry = _write_fixture(tmp_path, provider="sqlite") binding = ("--goal-id", GOAL_ID, "--agent-id", AGENT_ID, "--todo-id", TODO_ID, "--turn-instance-id", TURN_ID) rc, admitted = _run_cli(registry, runtime, "quota", "should-run", @@ -65,8 +65,10 @@ def test_missing_replaced_stale_context_requires_reread_and_preserves_delivery(t # A synthetic source edit represents a peer's completed update. The separate # concurrency test proves the cooperating writer cannot cross the commit guard. - state.write_text(state.read_text(encoding="utf-8").replace( - "Validate and settle the selected delivery.", "Validate the updated delivery scope."), encoding="utf-8") + rc, updated = _run_cli(registry, runtime, "todo", "update", "--goal-id", GOAL_ID, + "--todo-id", TODO_ID, "--agent-id", AGENT_ID, "--text", "Validate the updated delivery scope.", + "--execute", cwd=project) + assert rc == 0 and updated["ok"], updated rc, stale = _run_cli(registry, runtime, *delivery, *vision, "--checkpoint-read-context", b["read_context_id"], cwd=project) assert rc == 1 and stale["error_code"] == "checkpoint_read_context_stale", stale @@ -90,17 +92,17 @@ def test_missing_replaced_stale_context_requires_reread_and_preserves_delivery(t assert _spend_run_count(runtime) == 0 -@pytest.mark.parametrize("provider", ["legacy", "file", "sqlite"]) +@pytest.mark.parametrize("provider", ["file", "sqlite"]) def test_large_goal_context_keeps_complete_basis_and_checkpoint_only_recovery( tmp_path, monkeypatch, provider, ): - from canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime + from canonical_authority_fixture import advance_canonical_fixture_authority, isolate_sqlite_runtime from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection from loopx.control_plane.effect_runtime import MAX_REQUEST_BYTES if provider == "sqlite": isolate_sqlite_runtime(tmp_path, monkeypatch) - project, runtime, registry, binding, delivery, original = _missing(tmp_path) + project, runtime, registry, binding, delivery, original = _missing(tmp_path, provider=provider) state = project / f".codex/goals/{GOAL_ID}/ACTIVE_GOAL_STATE.md" if provider != "legacy": todo = {"schema_version": "todo_item_v0", "todo_id": TODO_ID, "index": 1, @@ -109,7 +111,7 @@ def test_large_goal_context_keeps_complete_basis_and_checkpoint_only_recovery( "task_class": "advancement_task", "archive_state": "active", "source_section": "Agent Todo"} projection = build_todo_runtime_shadow_projection(goal_id=GOAL_ID, todos=[todo]) - initialize_canonical_authority(runtime, GOAL_ID, projection, state_path=state, provider=provider) + advance_canonical_fixture_authority(runtime, GOAL_ID, projection, operation_id="fixture-checkpoint-context") # The original failure came from complete source prose, not from the # small prior Turn receipt. Keep the source intact and cross both 2 MiB # request and response boundaries through the real local CLI/runtime. @@ -142,7 +144,7 @@ def test_large_goal_context_keeps_complete_basis_and_checkpoint_only_recovery( @pytest.mark.parametrize("provider", ["file", "sqlite"]) def test_context_reads_real_canonical_todo_and_owner_acceptance(tmp_path, monkeypatch, provider): - from canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime + from canonical_authority_fixture import advance_canonical_fixture_authority, isolate_sqlite_runtime from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection from loopx.control_plane.coordination.local_authority_shadow_projection import canonical_bytes from loopx.control_plane.goals.checkpoint_context_io import read_checkpoint_context @@ -152,8 +154,7 @@ def test_context_reads_real_canonical_todo_and_owner_acceptance(tmp_path, monkey isolate_sqlite_runtime(tmp_path, monkeypatch) import tempfile monkeypatch.setattr(tempfile, "tempdir", str(tmp_path)) - project, runtime, registry, _, _, _ = _missing(tmp_path) - state = project / f".codex/goals/{GOAL_ID}/ACTIVE_GOAL_STATE.md" + project, runtime, registry, _, _, _ = _missing(tmp_path, provider=provider) todo = {"schema_version": "todo_item_v0", "todo_id": TODO_ID, "index": 1, "role": "agent", "status": "done", "done": True, "text": "Canonical delivered result", "task_class": "advancement_task", "archive_state": "active", "source_section": "Agent Todo"} @@ -164,7 +165,7 @@ def test_context_reads_real_canonical_todo_and_owner_acceptance(tmp_path, monkey projection["goal_acceptance"] = {"schema_version": "loopx_goal_acceptance_v0", "enabled": True, "revision": 1, "digest": hashlib.sha256(canonical_bytes(document)).hexdigest(), "document": document, "bindings": [], "verification": None} - initialize_canonical_authority(runtime, GOAL_ID, projection, state_path=state, provider=provider) + advance_canonical_fixture_authority(runtime, GOAL_ID, projection, operation_id="fixture-checkpoint-context") context = read_checkpoint_context(registry_path=registry, runtime_root_override=str(runtime), goal_id=GOAL_ID, agent_id=AGENT_ID, todo_id=TODO_ID, turn_instance_id=TURN_ID) assert context["basis"]["todo"]["text"] == "Canonical delivered result" diff --git a/tests/control_plane/test_refresh_checkpoint_isolation.py b/tests/control_plane/test_refresh_checkpoint_isolation.py index 342d787601..ba3faaade0 100644 --- a/tests/control_plane/test_refresh_checkpoint_isolation.py +++ b/tests/control_plane/test_refresh_checkpoint_isolation.py @@ -110,7 +110,7 @@ def unexpected_notification_failure(error): pytest.fail(f"Unexpected notification failure: {type(error).__name__}") monkeypatch.setattr(goal_channel_lifecycle, "_exception_reason", unexpected_notification_failure) - project, runtime, registry = _write_fixture(tmp_path) + project, runtime, registry = _write_fixture(tmp_path, provider="sqlite") shared = tmp_path / "shared-runtime" monkeypatch.setenv("LOOPX_RUNTIME_ROOT", str(shared)) monkeypatch.setenv("PYTHONPATH", str(REPO_ROOT)) @@ -136,6 +136,12 @@ def run(argv, output="json", expected_rc=0): ]) assert guard["decision"] == "run", guard _append_blocking_user_gate(project) + from canonical_authority_fixture import advance_canonical_fixture_authority, read_fixture_todos + from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection + advance_canonical_fixture_authority(runtime, GOAL_ID, + build_todo_runtime_shadow_projection(goal_id=GOAL_ID, + todos=read_fixture_todos(project / f".codex/goals/{GOAL_ID}/ACTIVE_GOAL_STATE.md"), + handoff_mode="soft_claim", leases=[]), operation_id="fixture-delivery-gate") _configure_sinks(registry, runtime) sync = sync_project_registry_to_global( registry_path=registry, runtime_root_override=str(shared), goal_id=GOAL_ID, diff --git a/tests/control_plane/test_refresh_checkpoint_recovery.py b/tests/control_plane/test_refresh_checkpoint_recovery.py index fa6e91d2fa..2418b8e340 100644 --- a/tests/control_plane/test_refresh_checkpoint_recovery.py +++ b/tests/control_plane/test_refresh_checkpoint_recovery.py @@ -130,7 +130,7 @@ def test_recovery_markdown_preserves_routing_and_error_precedence(decision): @pytest.mark.parametrize("decision", ["unchanged", "patch"]) def test_same_turn_checkpoint_supplement_with_read_context_is_idempotent(tmp_path: Path, decision: str): - project, runtime, registry = _write_fixture(tmp_path) + project, runtime, registry = _write_fixture(tmp_path, provider="sqlite") rc, initial = _run_cli( registry, runtime, @@ -299,7 +299,7 @@ def test_same_turn_checkpoint_supplement_with_read_context_is_idempotent(tmp_pat def test_checkpoint_only_recovery_bypasses_open_todo_completion_validation( tmp_path: Path, ) -> None: - project, runtime, registry = _write_fixture(tmp_path) + project, runtime, registry = _write_fixture(tmp_path, provider="sqlite") _initialize_git_checkout(project) state_path = project / f".codex/goals/{GOAL_ID}/ACTIVE_GOAL_STATE.md" state_text = state_path.read_text(encoding="utf-8") @@ -378,14 +378,10 @@ def test_checkpoint_only_recovery_bypasses_open_todo_completion_validation( # A controller may install completion validation before checkpoint repair. # The supplement repairs the existing receipt while the Todo remains open. - state_path.write_text( - state_path.read_text(encoding="utf-8").replace( - selected_marker, - selected_marker.replace(" -->", " validation_command=pytest -->"), - 1, - ), - encoding="utf-8", - ) + rc, updated = _run_cli(registry, runtime, "todo", "update", "--goal-id", GOAL_ID, + "--todo-id", TODO_ID, "--agent-id", AGENT_ID, "--validation-command", "pytest", + "--execute", cwd=project) + assert rc == 0 and updated["ok"], updated vision = ( "--vision-summary", @@ -438,7 +434,7 @@ def test_checkpoint_only_recovery_bypasses_open_todo_completion_validation( def test_invalid_supplement_leaves_original_writeback_intact(tmp_path: Path): - project, runtime, registry = _write_fixture(tmp_path) + project, runtime, registry = _write_fixture(tmp_path, provider="sqlite") binding = ( "--agent-id", AGENT_ID, diff --git a/tests/control_plane/test_task_planning.py b/tests/control_plane/test_task_planning.py index cdd3ac06e3..635047ea85 100644 --- a/tests/control_plane/test_task_planning.py +++ b/tests/control_plane/test_task_planning.py @@ -7,6 +7,11 @@ import pytest +from canonical_authority_fixture import ( + advance_canonical_fixture_authority, initialize_canonical_source_fixture, read_fixture_todos, +) +from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection + from loopx.cli import main from loopx.control_plane.goals.start_contract import ( build_goal_start_prompt, @@ -24,6 +29,7 @@ def bound_goal(tmp_path): json.dumps( { "schema_version": "0.1", + "common_runtime_root": str(tmp_path / "runtime"), "goals": [ { "id": "planning-goal", @@ -37,6 +43,9 @@ def bound_goal(tmp_path): } ) ) + initialize_canonical_source_fixture( + tmp_path / "runtime", "planning-goal", state_path=state, provider="sqlite", + ) return dict( registry_path=registry, goal_id="planning-goal", @@ -47,6 +56,15 @@ def bound_goal(tmp_path): ) +def _replace_frontier(bound_goal, operation_id): + advance_canonical_fixture_authority( + Path(bound_goal["runtime_root_arg"]), bound_goal["goal_id"], + build_todo_runtime_shadow_projection(goal_id=bound_goal["goal_id"], + todos=read_fixture_todos(bound_goal["project"] / "state.md"), + handoff_mode="soft_claim", leases=[]), operation_id=operation_id, + ) + + def test_empty_frontier_plans_before_writes_without_starting_a_loop(bound_goal): registry = bound_goal["registry_path"] state = bound_goal["project"] / "state.md" @@ -82,6 +100,7 @@ def test_existing_plan_is_an_incremental_frontier_not_a_new_goal(bound_goal): ## User Todos """) + _replace_frontier(bound_goal, "fixture-existing-plan") packet = build_task_planning_packet(**bound_goal) assert packet["runnable_todo_ids"] == ["todo_existing"] assert [step["id"] for step in packet["ordered_steps"]] == [ @@ -136,9 +155,10 @@ def test_full_frontier_and_live_blockers_ignore_display_order_and_terminal_items f" " for status in ("open", "blocked", "done", "deferred") ] - for ordered in (work, list(reversed(work))): + for position, ordered in enumerate((work, list(reversed(work)))): state.write_text("# Active Goal State\n\n## Agent Todos\n" + "\n".join(ordered) + "\n\n## User Todos\n" + "\n".join(gates) + "\n") + _replace_frontier(bound_goal, f"fixture-frontier-order-{position}") packet = build_task_planning_packet(**bound_goal) assert set(packet["runnable_todo_ids"]) == {f"todo_work_{i}" for i in range(40)} assert set(packet["blocking_todo_ids"]) == {"todo_gate_open", "todo_gate_blocked"} From 91d754df6584581845ab36583b5a5915dab61303 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Tue, 27 Oct 2026 16:07:12 +0800 Subject: [PATCH 7/9] test: retain canonical validation revisions and a single user section Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- tests/control_plane/test_quota_settlement_cli.py | 13 +++++++------ .../test_refresh_checkpoint_recovery.py | 5 +++++ 2 files changed, 12 insertions(+), 6 deletions(-) diff --git a/tests/control_plane/test_quota_settlement_cli.py b/tests/control_plane/test_quota_settlement_cli.py index 174c2fd831..1186f4c185 100644 --- a/tests/control_plane/test_quota_settlement_cli.py +++ b/tests/control_plane/test_quota_settlement_cli.py @@ -528,16 +528,17 @@ def _append_due_monitors( def _append_blocking_user_gate(project: Path) -> None: state_path = project / f".codex/goals/{GOAL_ID}/ACTIVE_GOAL_STATE.md" state_text = state_path.read_text(encoding="utf-8") - state_path.write_text( - state_text.rstrip() - + "\n\n## User Todo\n\n" - + "- [ ] [P0-user] Approve the newly introduced delivery gate.\n" + gate = ( + "- [ ] [P0-user] Approve the newly introduced delivery gate.\n" + " \n", - encoding="utf-8", + + "decision_scope=direction:action:fixture priority=P0-USER -->\n" ) + marker = "\n## User Todo\n" + updated = (state_text.replace(marker, marker + "\n" + gate, 1) + if marker in state_text else state_text.rstrip() + "\n\n## User Todo\n\n" + gate) + state_path.write_text(updated, encoding="utf-8") def _heartbeat_receipt_events( diff --git a/tests/control_plane/test_refresh_checkpoint_recovery.py b/tests/control_plane/test_refresh_checkpoint_recovery.py index 2418b8e340..99233b927d 100644 --- a/tests/control_plane/test_refresh_checkpoint_recovery.py +++ b/tests/control_plane/test_refresh_checkpoint_recovery.py @@ -378,8 +378,13 @@ def test_checkpoint_only_recovery_bypasses_open_todo_completion_validation( # A controller may install completion validation before checkpoint repair. # The supplement repairs the existing receipt while the Todo remains open. + rc, current = _run_cli(registry, runtime, "todo", "list", "--goal-id", GOAL_ID, + "--todo-id", TODO_ID, cwd=project) + assert rc == 0, current rc, updated = _run_cli(registry, runtime, "todo", "update", "--goal-id", GOAL_ID, "--todo-id", TODO_ID, "--agent-id", AGENT_ID, "--validation-command", "pytest", + "--update-operation-id", "fixture-checkpoint-validation", + "--update-expected-provider-revision", str(current["authority_read"]["provider_revision"]), "--execute", cwd=project) assert rc == 0 and updated["ok"], updated From af1222d4655d290a5fc587aade4040b903518c2a Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Tue, 27 Oct 2026 16:14:46 +0800 Subject: [PATCH 8/9] test: reuse the canonical section owner for fixture gates Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../control_plane/test_quota_settlement_cli.py | 17 +++++++++++++---- 1 file changed, 13 insertions(+), 4 deletions(-) diff --git a/tests/control_plane/test_quota_settlement_cli.py b/tests/control_plane/test_quota_settlement_cli.py index 1186f4c185..7ff68d9ce3 100644 --- a/tests/control_plane/test_quota_settlement_cli.py +++ b/tests/control_plane/test_quota_settlement_cli.py @@ -526,6 +526,9 @@ def _append_due_monitors( def _append_blocking_user_gate(project: Path) -> None: + from loopx.control_plane.todos.active_state_todo_parser import parse_active_state_todos + from loopx.control_plane.todos.machine_section_projection import render_canonical_todo_sections + state_path = project / f".codex/goals/{GOAL_ID}/ACTIVE_GOAL_STATE.md" state_text = state_path.read_text(encoding="utf-8") gate = ( @@ -535,10 +538,16 @@ def _append_blocking_user_gate(project: Path) -> None: + f"blocks_agent={AGENT_ID} " + "decision_scope=direction:action:fixture priority=P0-USER -->\n" ) - marker = "\n## User Todo\n" - updated = (state_text.replace(marker, marker + "\n" + gate, 1) - if marker in state_text else state_text.rstrip() + "\n\n## User Todo\n\n" + gate) - state_path.write_text(updated, encoding="utf-8") + additions = parse_active_state_todos( + "## User Todo\n\n" + gate, state_path=state_path, item_limit=None, + )["user_todos"]["items"] + # Reuse the section owner for both authored and already materialized input. + # This edits fixture source only; callers still initialize or advance CAS. + rendered = render_canonical_todo_sections( + state_text, [*read_fixture_todos(state_path), *additions], + provider_revision="fixture-user-gate", + ) + state_path.write_text(rendered.markdown, encoding="utf-8") def _heartbeat_receipt_events( From bce90c268c1b638a2b494cef01ed3a603cebafeb Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Tue, 27 Oct 2026 16:25:16 +0800 Subject: [PATCH 9/9] test: normalize fixture rows before machine-section rendering Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- tests/control_plane/test_quota_settlement_cli.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/tests/control_plane/test_quota_settlement_cli.py b/tests/control_plane/test_quota_settlement_cli.py index 7ff68d9ce3..4ba5881a10 100644 --- a/tests/control_plane/test_quota_settlement_cli.py +++ b/tests/control_plane/test_quota_settlement_cli.py @@ -528,6 +528,7 @@ def _append_due_monitors( def _append_blocking_user_gate(project: Path) -> None: from loopx.control_plane.todos.active_state_todo_parser import parse_active_state_todos from loopx.control_plane.todos.machine_section_projection import render_canonical_todo_sections + from loopx.control_plane.todos.todo_summary import canonical_todo_read_record state_path = project / f".codex/goals/{GOAL_ID}/ACTIVE_GOAL_STATE.md" state_text = state_path.read_text(encoding="utf-8") @@ -544,7 +545,8 @@ def _append_blocking_user_gate(project: Path) -> None: # Reuse the section owner for both authored and already materialized input. # This edits fixture source only; callers still initialize or advance CAS. rendered = render_canonical_todo_sections( - state_text, [*read_fixture_todos(state_path), *additions], + state_text, [canonical_todo_read_record(row) + for row in [*read_fixture_todos(state_path), *additions]], provider_revision="fixture-user-gate", ) state_path.write_text(rendered.markdown, encoding="utf-8")