From 9a2cebb13ab5a210daf3f928769341b6d79e0ced Mon Sep 17 00:00:00 2001 From: hyk <4408344+hhyykk@users.noreply.github.com> Date: Wed, 7 Oct 2026 18:51:09 +0800 Subject: [PATCH 1/2] test(control-plane): adopt canonical Todo read contracts Signed-off-by: hyk <4408344+hhyykk@users.noreply.github.com> --- .../test_canonical_claim_execution_proof.py | 2 +- .../test_canonical_todo_writer_isolation.py | 4 +- .../test_causal_blocked_closeout_cli.py | 4 +- .../test_heartbeat_registry_route.py | 7 ++- .../test_shadow_registry_source.py | 2 +- .../test_succession_provider_readback.py | 53 +++++++++++++++---- tests/test_quota_bound_wait_recovery.py | 4 +- 7 files changed, 56 insertions(+), 20 deletions(-) diff --git a/tests/control_plane/test_canonical_claim_execution_proof.py b/tests/control_plane/test_canonical_claim_execution_proof.py index 83151b7285..5f9bac937a 100644 --- a/tests/control_plane/test_canonical_claim_execution_proof.py +++ b/tests/control_plane/test_canonical_claim_execution_proof.py @@ -72,7 +72,7 @@ def cli(command, *args, expected_exit=0): assert not state.exists(), "rejected recovery cannot revive a legacy writer" readback = cli(["todo", "list"]) assert first["source_authority"] == provider + "_v0" - assert readback["todos"][0]["claimed_by"] == ("agent-b" if retirement == "transfer" else "agent-a") + assert readback["todo"]["claimed_by"] == ("agent-b" if retirement == "transfer" else "agent-a") assert not (runtime / "goals" / goal / "task-leases" / f"{target}.json").exists() finally: subprocess.run([sys.executable, "-c", "from loopx.control_plane.effect_runtime import effect_runtime_result; effect_runtime_result('runtime.shutdown',{},retry_safe=False)"], diff --git a/tests/control_plane/test_canonical_todo_writer_isolation.py b/tests/control_plane/test_canonical_todo_writer_isolation.py index 4a481f0dfc..3ae1760835 100644 --- a/tests/control_plane/test_canonical_todo_writer_isolation.py +++ b/tests/control_plane/test_canonical_todo_writer_isolation.py @@ -72,8 +72,8 @@ def add(role, text, operation): "--note", "Retain source identity", "--update-operation-id", "update", "--update-expected-provider-revision", listing["authority_read"]["provider_revision"]) assert updated["todo_id"] == original - updated_records = cli("list", "--goal-id", "goal-a", "--todo-id", original)["todos"] - assert updated_records[0]["note"] == "Retain source identity" + updated_record = cli("list", "--goal-id", "goal-a", "--todo-id", original)["todo"] + assert updated_record["note"] == "Retain source identity" completed = cli("complete", "--goal-id", "goal-a", "--todo-id", user, "--no-follow-up", "--note", "Independent user action accepted") assert completed["completed"] and completed["todo_id"] == user diff --git a/tests/control_plane/test_causal_blocked_closeout_cli.py b/tests/control_plane/test_causal_blocked_closeout_cli.py index 47476571da..6751a8e633 100644 --- a/tests/control_plane/test_causal_blocked_closeout_cli.py +++ b/tests/control_plane/test_causal_blocked_closeout_cli.py @@ -79,7 +79,7 @@ def cli(*args: str, cwd: Path = project) -> tuple[int, dict]: assert rc == 0 and guard["should_run"] is True, guard rc, original = cli("todo", "list", "--goal-id", GOAL_ID, "--todo-id", TODO_ID) assert rc == 0, original - digest = original["todos"][0]["completion_validation_sha256"] + digest = original["todo"]["completion_validation_sha256"] rc, wait = cli("todo", "update", "--goal-id", GOAL_ID, "--todo-id", TODO_ID, "--agent-id", AGENT_ID, "--resume-when", f"{kind}:{MONITOR_ID}", @@ -121,7 +121,7 @@ def cli(*args: str, cwd: Path = project) -> tuple[int, dict]: assert _spend_run_count(runtime) == 0 rc, after = cli("todo", "list", "--goal-id", GOAL_ID, "--todo-id", TODO_ID) assert rc == 0, after - todo = after["todos"][0] + todo = after["todo"] assert todo["status"] == ("deferred" if defer else "open") and todo["resume_ready"] is False assert todo["resume_when"] == f"{kind}:{MONITOR_ID}" assert todo["completion_validation_sha256"] == digest diff --git a/tests/control_plane/test_heartbeat_registry_route.py b/tests/control_plane/test_heartbeat_registry_route.py index 24b14a030f..a8333a105d 100644 --- a/tests/control_plane/test_heartbeat_registry_route.py +++ b/tests/control_plane/test_heartbeat_registry_route.py @@ -120,7 +120,12 @@ def test_cli_generated_guard_selection_and_settlement_ignore_conflicting_cwd_reg code, guard = _execute(shlex.split(selection.replace("{todo_id}", cli.TODO_ID))[1:], project) assert code == 0, guard assert guard["selected_todo"]["todo_id"] == cli.TODO_ID - assert any(read["kind"] == "agent_preferences" for read in guard["required_reads"]) + work_context = guard["interaction_contract"]["agent_channel"]["work_context"] + preference = next(source for source in work_context["sources"] + if source.get("kind") == "agent_preferences") + assert preference["content"]["current"]["items"][0]["statement"] == "Use the designated reviewer." + assert not any(read.get("kind") == "agent_preferences" + for read in guard["interaction_contract"]["agent_channel"]["required_reads"]) identity = guard["heartbeat_receipt"]["settlement_identity"] # A later no-argument reentry must retain the original choice and authority. code, guard = _execute(guard_argv, project) diff --git a/tests/control_plane/test_shadow_registry_source.py b/tests/control_plane/test_shadow_registry_source.py index f5b2ed5b4c..78bd4040bc 100644 --- a/tests/control_plane/test_shadow_registry_source.py +++ b/tests/control_plane/test_shadow_registry_source.py @@ -171,7 +171,7 @@ def test_terminal_lease_survives_public_cutover_and_original_recovery(tmp_path: assert recovered["promotion"]["status"] == "replayed", recovered assert ws.cli("task-lease", "inspect", "--todo-id", todo_id)["lease"] == expected retained = ws.cli("todo", "list", "--todo-id", later["todo_id"]) - assert retained["todos"][0]["text"] == "Later canonical work must survive recovery" + assert retained["todo"]["text"] == "Later canonical work must survive recovery" assert lease.read_bytes() == original assert not ws.state.exists() diff --git a/tests/control_plane/test_succession_provider_readback.py b/tests/control_plane/test_succession_provider_readback.py index ab6adfcb6c..84c89e9064 100644 --- a/tests/control_plane/test_succession_provider_readback.py +++ b/tests/control_plane/test_succession_provider_readback.py @@ -13,7 +13,9 @@ from loopx.control_plane.coordination.local_authority import read_canonical_todos_if_promoted from loopx.control_plane.todos.machine_section_projection import render_canonical_todo_sections from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection +from loopx.control_plane.todos.goal_todo_projection import filtered_todo_summary from loopx.todos import list_goal_todos +from loopx.control_plane.todos.todo_summary import compact_todo_group def fixture_projection(): @@ -53,13 +55,17 @@ def test_real_cli_historical_route_advisory_preserves_old_source_and_provider(tm state.unlink() source = state.read_bytes() if state.exists() else None before = read_canonical_todos_if_promoted(runtime_root=runtime, goal_id='goal-a') + inventory = cli(registry) + handoff_gates = inventory['agent_todos']['handoff_gates'] for todo_id, expected in [('todo_legacy', True), ('todo_ordinary', False), ('todo_closed', False)]: result = cli(registry, '--todo-id', todo_id) - summary = result['agent_todos'] - handoff = summary['handoff_gates'][0] + assert result['matched'] is True + assert result['todo']['todo_id'] == todo_id + assert result['relations']['unblocks_todo_id'] == 'todo_work' + handoff = next(gate for gate in handoff_gates if gate['todo_id'] == todo_id) assert (handoff.get('route_continuation_replan_required') is True) is expected assert handoff['gate_state'] == ('cleared_no_followup' if todo_id == 'todo_closed' else 'blocking') - assert 'terminal_closure_proof' not in summary or todo_id == 'todo_closed' + assert 'terminal_closure_proof' not in inventory['agent_todos'] assert result['todo']['excluded_agents'] == ['agent-b'] assert read_canonical_todos_if_promoted(runtime_root=runtime, goal_id='goal-a') == before assert (state.read_bytes() if state.exists() else None) == source @@ -86,13 +92,35 @@ def test_real_cli_graph_selection_history_and_read_only_manager(tmp_path, monkey state.unlink() # Promoted readback must not rely on or regenerate Markdown. state_before = state.read_bytes() if state.exists() else None before = read_canonical_todos_if_promoted(runtime_root=runtime, goal_id='goal-a') + inventory = cli(registry) + summary = inventory['agent_todos'] + full_source = list_goal_todos(registry_path=registry, goal_id='goal-a') + source_rows = full_source['todos'] + agent_source_rows = [row for row in source_rows if row.get('role') == 'agent'] + source_summary = compact_todo_group(agent_source_rows, source_section='Agent Todo', role='agent', + resume_source_items=source_rows, item_limit=None) + assert summary['total_count'] == len(agent_source_rows) + assert summary['completed_without_successor_count'] == source_summary['completed_without_successor_count'] + assert 'terminal_closure_proof' not in summary for name, gap in [('inferred_source', 0), ('missing_source', 1), ('self_source', 1), ('handoff_source', 0)]: result = cli(registry, '--todo-id', cases[name], '--limit', '1') - summary = result['agent_todos'] - assert summary.get('completed_without_successor_count', 0) == gap - assert 'terminal_closure_proof' not in summary + assert result['matched'] is True + assert result['todo']['todo_id'] == cases[name] + assert 'agent_todos' not in result and 'todos' not in result and 'user_todos' not in result + selected = filtered_todo_summary(source_summary, role='agent', todo_id=cases[name]) + assert selected.get('completed_without_successor_count', 0) == gap + assert 'terminal_closure_proof' not in selected + if name == 'inferred_source': + assert result['relations']['completion_continuation'] == 'successor' + elif name == 'missing_source': + assert result['relations']['successor_todo_ids'] == ['todo_missing_continuation'] + elif name == 'self_source': + assert result['relations']['successor_todo_ids'] == [cases[name]] + elif name == 'handoff_source': + assert result['relations']['successor_todo_ids'] == [cases['explicit_target']] if name == 'handoff_source': - assert summary['handoff_gates'][0]['gate_state'] == 'cleared_with_successor' + handoff = next(gate for gate in summary['handoff_gates'] if gate['todo_id'] == cases[name]) + assert handoff['gate_state'] == 'cleared_with_successor' details = read_manager_goal_details(registry, runtime, 'goal-a', owner_scope=True, limit=3) assert details['status'] == 'read' assert details['coverage']['active'] > details['coverage']['included'] @@ -127,15 +155,18 @@ def test_large_goal_reuses_complete_succession_without_exceeding_rpc_budget(tmp_ }]})) initialize_canonical_authority(runtime, 'goal-a', projection, state_path=state, provider=provider) before = read_canonical_todos_if_promoted(runtime_root=runtime, goal_id='goal-a') - result = cli(registry) - summary = result['agent_todos'] + inventory = cli(registry) + summary = inventory['agent_todos'] assert summary['total_count'] == 5000 assert summary['done_count'] == 4999 assert summary['open_count'] == 1 assert summary.get('completed_without_successor_count', 0) == 0 assert 'terminal_closure_proof' not in summary - selected = cli(registry, '--todo-id', 'todo_capacity_00000', '--limit', '1')['agent_todos'] - assert selected.get('completed_without_successor_count', 0) == 0 + selected = cli(registry, '--todo-id', 'todo_capacity_00000', '--limit', '1') + assert selected['matched'] is True + assert selected['todo']['todo_id'] == 'todo_capacity_00000' + assert selected['todo']['status'] == 'done' + assert 'agent_todos' not in selected and 'todos' not in selected and 'user_todos' not in selected assert 'terminal_closure_proof' not in selected after = read_canonical_todos_if_promoted(runtime_root=runtime, goal_id='goal-a') assert after == before diff --git a/tests/test_quota_bound_wait_recovery.py b/tests/test_quota_bound_wait_recovery.py index af95dcbf5f..cffa2bc893 100644 --- a/tests/test_quota_bound_wait_recovery.py +++ b/tests/test_quota_bound_wait_recovery.py @@ -152,7 +152,7 @@ def cli(*args: str, cwd: Path = project) -> tuple[int, dict]: assert rc == 0 and guard["should_run"] is True, guard rc, original = cli("todo", "list", "--goal-id", GOAL_ID, "--todo-id", TODO_ID) assert rc == 0, original - digest = original["todos"][0]["completion_validation_sha256"] + digest = original["todo"]["completion_validation_sha256"] rc, wait = cli( "todo", "update", @@ -273,7 +273,7 @@ def cli(*args: str, cwd: Path = project) -> tuple[int, dict]: assert _spend_run_count(runtime) == 0 rc, after = cli("todo", "list", "--goal-id", GOAL_ID, "--todo-id", TODO_ID) assert rc == 0, after - todo = after["todos"][0] + todo = after["todo"] assert todo["status"] == "open" and todo["resume_ready"] is False assert todo["resume_when"] == f"{kind}:{MONITOR_ID}" assert todo["completion_validation_sha256"] == digest From af083642d87d9c4cd64386b091fb184fa769c1be Mon Sep 17 00:00:00 2001 From: hyk <4408344+hhyykk@users.noreply.github.com> Date: Wed, 7 Oct 2026 19:13:10 +0800 Subject: [PATCH 2/2] test(control-plane): defer shared Todo assertions to current owner Signed-off-by: hyk <4408344+hhyykk@users.noreply.github.com> --- .../test_canonical_claim_execution_proof.py | 2 +- .../test_canonical_todo_writer_isolation.py | 4 +- .../test_causal_blocked_closeout_cli.py | 4 +- .../test_shadow_registry_source.py | 2 +- .../test_succession_provider_readback.py | 53 ++++--------------- 5 files changed, 17 insertions(+), 48 deletions(-) diff --git a/tests/control_plane/test_canonical_claim_execution_proof.py b/tests/control_plane/test_canonical_claim_execution_proof.py index 5f9bac937a..83151b7285 100644 --- a/tests/control_plane/test_canonical_claim_execution_proof.py +++ b/tests/control_plane/test_canonical_claim_execution_proof.py @@ -72,7 +72,7 @@ def cli(command, *args, expected_exit=0): assert not state.exists(), "rejected recovery cannot revive a legacy writer" readback = cli(["todo", "list"]) assert first["source_authority"] == provider + "_v0" - assert readback["todo"]["claimed_by"] == ("agent-b" if retirement == "transfer" else "agent-a") + assert readback["todos"][0]["claimed_by"] == ("agent-b" if retirement == "transfer" else "agent-a") assert not (runtime / "goals" / goal / "task-leases" / f"{target}.json").exists() finally: subprocess.run([sys.executable, "-c", "from loopx.control_plane.effect_runtime import effect_runtime_result; effect_runtime_result('runtime.shutdown',{},retry_safe=False)"], diff --git a/tests/control_plane/test_canonical_todo_writer_isolation.py b/tests/control_plane/test_canonical_todo_writer_isolation.py index 3ae1760835..4a481f0dfc 100644 --- a/tests/control_plane/test_canonical_todo_writer_isolation.py +++ b/tests/control_plane/test_canonical_todo_writer_isolation.py @@ -72,8 +72,8 @@ def add(role, text, operation): "--note", "Retain source identity", "--update-operation-id", "update", "--update-expected-provider-revision", listing["authority_read"]["provider_revision"]) assert updated["todo_id"] == original - updated_record = cli("list", "--goal-id", "goal-a", "--todo-id", original)["todo"] - assert updated_record["note"] == "Retain source identity" + updated_records = cli("list", "--goal-id", "goal-a", "--todo-id", original)["todos"] + assert updated_records[0]["note"] == "Retain source identity" completed = cli("complete", "--goal-id", "goal-a", "--todo-id", user, "--no-follow-up", "--note", "Independent user action accepted") assert completed["completed"] and completed["todo_id"] == user diff --git a/tests/control_plane/test_causal_blocked_closeout_cli.py b/tests/control_plane/test_causal_blocked_closeout_cli.py index 6751a8e633..47476571da 100644 --- a/tests/control_plane/test_causal_blocked_closeout_cli.py +++ b/tests/control_plane/test_causal_blocked_closeout_cli.py @@ -79,7 +79,7 @@ def cli(*args: str, cwd: Path = project) -> tuple[int, dict]: assert rc == 0 and guard["should_run"] is True, guard rc, original = cli("todo", "list", "--goal-id", GOAL_ID, "--todo-id", TODO_ID) assert rc == 0, original - digest = original["todo"]["completion_validation_sha256"] + digest = original["todos"][0]["completion_validation_sha256"] rc, wait = cli("todo", "update", "--goal-id", GOAL_ID, "--todo-id", TODO_ID, "--agent-id", AGENT_ID, "--resume-when", f"{kind}:{MONITOR_ID}", @@ -121,7 +121,7 @@ def cli(*args: str, cwd: Path = project) -> tuple[int, dict]: assert _spend_run_count(runtime) == 0 rc, after = cli("todo", "list", "--goal-id", GOAL_ID, "--todo-id", TODO_ID) assert rc == 0, after - todo = after["todo"] + todo = after["todos"][0] assert todo["status"] == ("deferred" if defer else "open") and todo["resume_ready"] is False assert todo["resume_when"] == f"{kind}:{MONITOR_ID}" assert todo["completion_validation_sha256"] == digest diff --git a/tests/control_plane/test_shadow_registry_source.py b/tests/control_plane/test_shadow_registry_source.py index 78bd4040bc..f5b2ed5b4c 100644 --- a/tests/control_plane/test_shadow_registry_source.py +++ b/tests/control_plane/test_shadow_registry_source.py @@ -171,7 +171,7 @@ def test_terminal_lease_survives_public_cutover_and_original_recovery(tmp_path: assert recovered["promotion"]["status"] == "replayed", recovered assert ws.cli("task-lease", "inspect", "--todo-id", todo_id)["lease"] == expected retained = ws.cli("todo", "list", "--todo-id", later["todo_id"]) - assert retained["todo"]["text"] == "Later canonical work must survive recovery" + assert retained["todos"][0]["text"] == "Later canonical work must survive recovery" assert lease.read_bytes() == original assert not ws.state.exists() diff --git a/tests/control_plane/test_succession_provider_readback.py b/tests/control_plane/test_succession_provider_readback.py index 84c89e9064..ab6adfcb6c 100644 --- a/tests/control_plane/test_succession_provider_readback.py +++ b/tests/control_plane/test_succession_provider_readback.py @@ -13,9 +13,7 @@ from loopx.control_plane.coordination.local_authority import read_canonical_todos_if_promoted from loopx.control_plane.todos.machine_section_projection import render_canonical_todo_sections from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection -from loopx.control_plane.todos.goal_todo_projection import filtered_todo_summary from loopx.todos import list_goal_todos -from loopx.control_plane.todos.todo_summary import compact_todo_group def fixture_projection(): @@ -55,17 +53,13 @@ def test_real_cli_historical_route_advisory_preserves_old_source_and_provider(tm state.unlink() source = state.read_bytes() if state.exists() else None before = read_canonical_todos_if_promoted(runtime_root=runtime, goal_id='goal-a') - inventory = cli(registry) - handoff_gates = inventory['agent_todos']['handoff_gates'] for todo_id, expected in [('todo_legacy', True), ('todo_ordinary', False), ('todo_closed', False)]: result = cli(registry, '--todo-id', todo_id) - assert result['matched'] is True - assert result['todo']['todo_id'] == todo_id - assert result['relations']['unblocks_todo_id'] == 'todo_work' - handoff = next(gate for gate in handoff_gates if gate['todo_id'] == todo_id) + summary = result['agent_todos'] + handoff = summary['handoff_gates'][0] assert (handoff.get('route_continuation_replan_required') is True) is expected assert handoff['gate_state'] == ('cleared_no_followup' if todo_id == 'todo_closed' else 'blocking') - assert 'terminal_closure_proof' not in inventory['agent_todos'] + assert 'terminal_closure_proof' not in summary or todo_id == 'todo_closed' assert result['todo']['excluded_agents'] == ['agent-b'] assert read_canonical_todos_if_promoted(runtime_root=runtime, goal_id='goal-a') == before assert (state.read_bytes() if state.exists() else None) == source @@ -92,35 +86,13 @@ def test_real_cli_graph_selection_history_and_read_only_manager(tmp_path, monkey state.unlink() # Promoted readback must not rely on or regenerate Markdown. state_before = state.read_bytes() if state.exists() else None before = read_canonical_todos_if_promoted(runtime_root=runtime, goal_id='goal-a') - inventory = cli(registry) - summary = inventory['agent_todos'] - full_source = list_goal_todos(registry_path=registry, goal_id='goal-a') - source_rows = full_source['todos'] - agent_source_rows = [row for row in source_rows if row.get('role') == 'agent'] - source_summary = compact_todo_group(agent_source_rows, source_section='Agent Todo', role='agent', - resume_source_items=source_rows, item_limit=None) - assert summary['total_count'] == len(agent_source_rows) - assert summary['completed_without_successor_count'] == source_summary['completed_without_successor_count'] - assert 'terminal_closure_proof' not in summary for name, gap in [('inferred_source', 0), ('missing_source', 1), ('self_source', 1), ('handoff_source', 0)]: result = cli(registry, '--todo-id', cases[name], '--limit', '1') - assert result['matched'] is True - assert result['todo']['todo_id'] == cases[name] - assert 'agent_todos' not in result and 'todos' not in result and 'user_todos' not in result - selected = filtered_todo_summary(source_summary, role='agent', todo_id=cases[name]) - assert selected.get('completed_without_successor_count', 0) == gap - assert 'terminal_closure_proof' not in selected - if name == 'inferred_source': - assert result['relations']['completion_continuation'] == 'successor' - elif name == 'missing_source': - assert result['relations']['successor_todo_ids'] == ['todo_missing_continuation'] - elif name == 'self_source': - assert result['relations']['successor_todo_ids'] == [cases[name]] - elif name == 'handoff_source': - assert result['relations']['successor_todo_ids'] == [cases['explicit_target']] + summary = result['agent_todos'] + assert summary.get('completed_without_successor_count', 0) == gap + assert 'terminal_closure_proof' not in summary if name == 'handoff_source': - handoff = next(gate for gate in summary['handoff_gates'] if gate['todo_id'] == cases[name]) - assert handoff['gate_state'] == 'cleared_with_successor' + assert summary['handoff_gates'][0]['gate_state'] == 'cleared_with_successor' details = read_manager_goal_details(registry, runtime, 'goal-a', owner_scope=True, limit=3) assert details['status'] == 'read' assert details['coverage']['active'] > details['coverage']['included'] @@ -155,18 +127,15 @@ def test_large_goal_reuses_complete_succession_without_exceeding_rpc_budget(tmp_ }]})) initialize_canonical_authority(runtime, 'goal-a', projection, state_path=state, provider=provider) before = read_canonical_todos_if_promoted(runtime_root=runtime, goal_id='goal-a') - inventory = cli(registry) - summary = inventory['agent_todos'] + result = cli(registry) + summary = result['agent_todos'] assert summary['total_count'] == 5000 assert summary['done_count'] == 4999 assert summary['open_count'] == 1 assert summary.get('completed_without_successor_count', 0) == 0 assert 'terminal_closure_proof' not in summary - selected = cli(registry, '--todo-id', 'todo_capacity_00000', '--limit', '1') - assert selected['matched'] is True - assert selected['todo']['todo_id'] == 'todo_capacity_00000' - assert selected['todo']['status'] == 'done' - assert 'agent_todos' not in selected and 'todos' not in selected and 'user_todos' not in selected + selected = cli(registry, '--todo-id', 'todo_capacity_00000', '--limit', '1')['agent_todos'] + assert selected.get('completed_without_successor_count', 0) == 0 assert 'terminal_closure_proof' not in selected after = read_canonical_todos_if_promoted(runtime_root=runtime, goal_id='goal-a') assert after == before