From a45ed0229990a8cc054120369ceedf1ee88cdf0b Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Fri, 16 Oct 2026 19:31:58 +0800 Subject: [PATCH 1/3] test: align coordination and chat fixtures with canonical authority Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../test_coordination_authority_core.py | 45 +++++++++++++++---- tests/test_chat_external_todo_confirmation.py | 23 ++++++++-- tests/test_chat_team_plan_action.py | 14 ++++-- 3 files changed, 66 insertions(+), 16 deletions(-) diff --git a/tests/control_plane/test_coordination_authority_core.py b/tests/control_plane/test_coordination_authority_core.py index 1882c6d331..7bc9ada942 100644 --- a/tests/control_plane/test_coordination_authority_core.py +++ b/tests/control_plane/test_coordination_authority_core.py @@ -54,7 +54,7 @@ def lease(**overrides: object) -> LeaseSnapshot: def snapshot(**overrides: object) -> CoordinationSnapshot: values: dict[str, object] = { - "handoff_mode": HandoffMode.LEGACY, + "handoff_mode": HandoffMode.SOFT_CLAIM, "registered_agents": (AGENT_A, AGENT_B, ORCHESTRATOR), "todo": todo(), } @@ -124,11 +124,18 @@ def test_missing_todo_is_a_typed_refusal_before_actor_admission(mode, action): "holder", [None, lease(active=False, status="expired"), lease(active=False, status="released"), lease()], ) -def test_snapshot_guard_keeps_claim_neutral_edits_and_holder_state(mode, holder): +def test_snapshot_guard_preserves_current_mode_edits_and_rejects_retired_mode(mode, holder): state = snapshot(handoff_mode=mode, lease=holder) result = decide( state, TodoMutationCommand(action=TodoAction.UPDATE, actor_agent_id=AGENT_A) ) + if mode is HandoffMode.LEGACY: + assert result.outcome is DecisionOutcome.REJECTED + assert result.code == "handoff_mode_retired" + assert result.next_snapshot is None + assert result.authority_mode is None + assert result.lease_fence is LeaseFence.NOT_REQUIRED + return assert result.outcome is DecisionOutcome.APPLY assert result.next_snapshot.todo == state.todo assert result.next_snapshot.lease == holder @@ -148,7 +155,13 @@ def test_update_authority_keeps_claim_neutral_edits_separate_from_ownership( ownership_mutation=ownership, ) result = decide(state, command) - if mode is HandoffMode.HARD_LEASE and ownership: + if mode is HandoffMode.LEGACY: + assert result.outcome is DecisionOutcome.REJECTED + assert result.code == "handoff_mode_retired" + assert result.next_snapshot is None + assert result.authority_mode is None + elif mode is HandoffMode.HARD_LEASE and ownership: + assert result.outcome is DecisionOutcome.REJECTED assert result.code == "handoff_mode_requires_lease" assert result.next_snapshot is None else: @@ -296,7 +309,8 @@ def test_exact_user_gate_override_requires_the_exact_linked_scope() -> None: @pytest.mark.parametrize( ("mode", "current_lease", "expected", "gate"), [ - (HandoffMode.LEGACY, None, DecisionOutcome.APPLY, OwnershipGate.NOT_REQUIRED), + (HandoffMode.LEGACY, None, DecisionOutcome.REJECTED, OwnershipGate.NOT_REQUIRED), + (HandoffMode.LEGACY, lease(), DecisionOutcome.REJECTED, OwnershipGate.NOT_REQUIRED), ( HandoffMode.SOFT_CLAIM, None, @@ -317,7 +331,7 @@ def test_exact_user_gate_override_requires_the_exact_linked_scope() -> None: ), ], ) -def test_ownership_handoff_preserves_the_three_modes( +def test_ownership_handoff_accepts_current_modes_and_rejects_retired_mode( mode: HandoffMode, current_lease: LeaseSnapshot | None, expected: DecisionOutcome, @@ -326,6 +340,10 @@ def test_ownership_handoff_preserves_the_three_modes( plan = decide(snapshot(handoff_mode=mode, lease=current_lease), claim()) assert plan.outcome is expected assert plan.ownership_gate is gate + if mode is HandoffMode.LEGACY: + assert plan.code == "handoff_mode_retired" + assert plan.next_snapshot is None + assert plan.authority_mode is None def test_delegated_hard_lease_ownership_change_uses_the_only_override() -> None: @@ -356,9 +374,20 @@ def test_delegated_hard_lease_ownership_change_uses_the_only_override() -> None: @pytest.mark.parametrize("action", [TodoAction.COMPLETE, TodoAction.SUPERSEDE]) def test_terminal_verbs_share_one_lease_fence(action: TodoAction) -> None: - legacy = decide(snapshot(), terminal(action)) - assert legacy.outcome is DecisionOutcome.APPLY - assert legacy.lease_fence is LeaseFence.NOT_REQUIRED + soft = decide(snapshot(), terminal(action)) + assert soft.outcome is DecisionOutcome.APPLY + assert soft.lease_fence is LeaseFence.NOT_REQUIRED + + for current_lease in (None, lease()): + retired = decide( + snapshot(handoff_mode=HandoffMode.LEGACY, lease=current_lease), + terminal(action), + ) + assert retired.outcome is DecisionOutcome.REJECTED + assert retired.code == "handoff_mode_retired" + assert retired.next_snapshot is None + assert retired.authority_mode is None + assert retired.lease_fence is LeaseFence.NOT_REQUIRED hard_missing = decide( snapshot(handoff_mode=HandoffMode.HARD_LEASE), diff --git a/tests/test_chat_external_todo_confirmation.py b/tests/test_chat_external_todo_confirmation.py index 0a97f6de6e..adf18d4273 100644 --- a/tests/test_chat_external_todo_confirmation.py +++ b/tests/test_chat_external_todo_confirmation.py @@ -4,16 +4,19 @@ import pytest +from loopx.control_plane.coordination.local_authority import ( + read_canonical_todos_if_promoted, +) from test_chat_team_plan_action import GOAL_ID, _fixture, _todos from test_native_steward_private import finish, steward # noqa: F401 def prepare(components, tmp_path, monkeypatch, *, goal_id=GOAL_ID): store, runtime, provider, transport, binding, capture, _ = components - project, registry, _ = _fixture(tmp_path / "canonical") - payload = json.loads(registry.read_text()) - payload["common_runtime_root"] = str(store.root.parent) - runtime.registry_path.write_text(json.dumps(payload)) + project, registry, _ = _fixture( + tmp_path / "canonical", runtime_root=store.root.parent + ) + runtime.registry_path.write_text(registry.read_text()) response = {"schema_version": "loopx_chat_agent_response_v0", "message": "待办预览；尚未写入。", "proposals": [{"kind": "todo", "goal_id": goal_id, "text": "Add reader comments later", "priority": "P2", "rationale": "Record the request without starting implementation."}], @@ -47,6 +50,15 @@ def test_original_owner_can_confirm_todo_once_without_creating_a_goal_or_executi assert applied["status"] == "applied", transport.core.pending() assert applied["receipt"]["resource_ids"]["goal_id"] == GOAL_ID assert _todos(project).count("Add reader comments later") == 1 + canonical = read_canonical_todos_if_promoted( + runtime_root=store.root.parent, goal_id=GOAL_ID + ) + assert canonical is not None + assert len(canonical["todos"]) == 1 + recorded = canonical["todos"][0] + assert recorded["todo_id"] == applied["receipt"]["resource_ids"]["todo_id"] + assert recorded["title"] == "Add reader comments later" + assert recorded["priority"] == "P2" assert len(store.list_sessions()) == sessions_before assert len(json.loads(runtime.registry_path.read_text())["goals"]) == 1 assert not capture.exists() or "thread/goal/set" not in capture.read_text() @@ -54,6 +66,9 @@ def test_original_owner_can_confirm_todo_once_without_creating_a_goal_or_executi transport.admit("steward-app", confirm) transport.reconcile() assert provider.writes == writes and _todos(project).count("Add reader comments later") == 1 + assert read_canonical_todos_if_promoted( + runtime_root=store.root.parent, goal_id=GOAL_ID + )["todos"] == canonical["todos"] assert transport.core.read_request(original["request_ref"])["session_id"] == original["session_id"] diff --git a/tests/test_chat_team_plan_action.py b/tests/test_chat_team_plan_action.py index 900192f335..2dee964c52 100644 --- a/tests/test_chat_team_plan_action.py +++ b/tests/test_chat_team_plan_action.py @@ -21,7 +21,13 @@ _PREVIEWS = itertools.count(1) -def _fixture(tmp_path: Path, *, agents: tuple[str, ...] = (AGENT_ID,)): +def _fixture( + tmp_path: Path, + *, + agents: tuple[str, ...] = (AGENT_ID,), + runtime_root: Path | None = None, +): + runtime_root = runtime_root if runtime_root is not None else tmp_path / "runtime" project = tmp_path / "project" state_file = f".codex/goals/{GOAL_ID}/ACTIVE_GOAL_STATE.md" state_path = project / state_file @@ -45,7 +51,7 @@ def _fixture(tmp_path: Path, *, agents: tuple[str, ...] = (AGENT_ID,)): json.dumps( { "schema_version": "0.1", - "common_runtime_root": str(tmp_path / "runtime"), + "common_runtime_root": str(runtime_root), "updated_at": "2026-01-01T00:00:00+00:00", "goals": [ { @@ -65,7 +71,7 @@ def _fixture(tmp_path: Path, *, agents: tuple[str, ...] = (AGENT_ID,)): encoding="utf-8", ) initialize_canonical_authority( - tmp_path / "runtime", + runtime_root, GOAL_ID, build_todo_runtime_shadow_projection( goal_id=GOAL_ID, @@ -75,7 +81,7 @@ def _fixture(tmp_path: Path, *, agents: tuple[str, ...] = (AGENT_ID,)): state_path=state_path, ) service = ChatActionService( - store=ChatActionStore(tmp_path / "runtime" / "chat" / "actions"), + store=ChatActionStore(runtime_root / "chat" / "actions"), registry_path=registry_path, ) return project, registry_path, service From 8572b4e1853a8b2c8cdead5676581442d304313d Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Fri, 16 Oct 2026 20:15:50 +0800 Subject: [PATCH 2/3] test: initialize replay and memory consumers on current authority Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../test_reward_memory_agent_bindings.py | 20 +++++++++++++++---- tests/control_plane/test_turn_resume_entry.py | 11 ++-------- 2 files changed, 18 insertions(+), 13 deletions(-) diff --git a/tests/capabilities/test_reward_memory_agent_bindings.py b/tests/capabilities/test_reward_memory_agent_bindings.py index bf1bd2745c..fa4dd27731 100644 --- a/tests/capabilities/test_reward_memory_agent_bindings.py +++ b/tests/capabilities/test_reward_memory_agent_bindings.py @@ -282,11 +282,16 @@ def _send_json(self, payload, *, status=200): assert handler.response["source_written"] is True and handler.response["readback_verified"] is False -def test_app_stage_resolves_the_registered_file_when_projection_omits_private_path(tmp_path, capsys): +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_app_stage_resolves_the_registered_file_when_projection_omits_private_path(tmp_path, capsys, monkeypatch, provider): import sys from loopx.cli import main from loopx.capabilities.reward_memory.codex_app_outcome import stage_codex_app_turn_outcome_candidate + 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 tests.control_plane.canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime from tests.capabilities.test_reward_memory_codex_app_outcome import _reflection + isolate_sqlite_runtime(tmp_path, monkeypatch) registry = fixture_registry(tmp_path) data = json.loads(registry.read_text()) goal = data["goals"][0] @@ -296,12 +301,19 @@ def test_app_stage_resolves_the_registered_file_when_projection_omits_private_pa registry.write_text(json.dumps(data)) state = Path(goal["repo"]) / "STATE.md" state.write_text("# Active Goal State\n\n## Agent Todo\n") + initialize_canonical_authority(runtime, "binding-goal", build_todo_runtime_shadow_projection( + goal_id="binding-goal", handoff_mode="soft_claim", todos=[]), state_path=state, provider=provider) validator = [sys.executable, "-c", "import sys,json;r=json.load(sys.stdin);print(json.dumps({'schema_version':'reward_memory_reflection_validation_v0','status':'validated','reflection_digest':r['reflection_digest'],'evidence_refs':r['reflection']['evidence_refs']}))"] - assert main(["--registry", str(registry), "--runtime-root", str(runtime), "--format", "json", "todo", "add", + code = main(["--registry", str(registry), "--runtime-root", str(runtime), "--format", "json", "todo", "add", "--goal-id", "binding-goal", "--role", "agent", "--text", "Verify the registered source", "--claimed-by", "writer", - "--validation-command-json", json.dumps(validator)]) == 0 + "--validation-command-json", json.dumps(validator)]) added = json.loads(capsys.readouterr().out) - assert added["ok"] + assert code == 0 and added["ok"], added + canonical = read_canonical_todos_if_promoted(runtime_root=runtime, goal_id="binding-goal") + assert canonical is not None and len(canonical["todos"]) == 1 + assert canonical["source_authority"] == ("sqlite_v0" if provider == "sqlite" else "file_v0") + assert canonical["todos"][0]["todo_id"] == added["todo_id"] + assert canonical["todos"][0]["completion_validation_required"] is True staged = stage_codex_app_turn_outcome_candidate(registry_path=registry, runtime_root=runtime, goal_id="binding-goal", agent_id="writer", todo_id=added["todo_id"], turn_instance_id="binding-source-test", effect_id="effect:binding-source", state_file=Path("."), validation_workspace=Path(goal["repo"]), reflection_json=_reflection(), observed_at="2026-10-16T00:00:00Z") diff --git a/tests/control_plane/test_turn_resume_entry.py b/tests/control_plane/test_turn_resume_entry.py index d516496484..fee53ad94c 100644 --- a/tests/control_plane/test_turn_resume_entry.py +++ b/tests/control_plane/test_turn_resume_entry.py @@ -15,8 +15,6 @@ from loopx.control_plane import effect_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.todos.active_state_todo_parser import parse_active_state_todos -from tests.control_plane.canonical_authority_fixture import initialize_canonical_authority from tests.test_loopx_turn_driver import ( _completion_host_and_validation_scripts, _turn_run_once_completion_argv, @@ -47,13 +45,7 @@ def _invoke(argv): def _committed_turn(root, provider): - project, runtime, registry = _write_live_fixture(root) - state = project / ".codex/goals" / GOAL / "ACTIVE_GOAL_STATE.md" - todos = parse_active_state_todos(state.read_text(), item_limit=None)["agent_todos"]["items"] - initialize_canonical_authority( - runtime, GOAL, build_todo_runtime_shadow_projection(goal_id=GOAL, todos=todos, handoff_mode="soft_claim"), - state_path=state, provider=provider, - ) + project, runtime, registry = _write_live_fixture(root, provider=provider) host, validator = _completion_host_and_validation_scripts() host = host.replace("request = json.load(sys.stdin)", 'request = json.load(sys.stdin)\npathlib.Path("host-calls").open("a").write("host\\n")') @@ -108,6 +100,7 @@ def test_committed_replay_survives_frontier_growth_beyond_fresh_request_limit(tm assert ledger.read_bytes() == original_ledger source = read_canonical_todos_if_promoted(runtime_root=runtime, goal_id=GOAL) assert source is not None + assert source["source_authority"] == ("sqlite_v0" if provider == "sqlite" else "file_v0") assert len(source["todos"]) == 2001 assert next(todo for todo in source["todos"] if todo["todo_id"] == "todo_fixture0001")["status"] == "done" From b6f1ae4ea64293be11fd07ca0c6c8a4ebdbbbb51 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sat, 17 Oct 2026 00:19:42 +0800 Subject: [PATCH 3/3] test: stabilize receipt collection and fence prose matching Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../test_content_digest_single_owner.py | 36 +++++++++++++++++- tests/test_projection_envelope.py | 37 +++++++++++++++++++ 2 files changed, 72 insertions(+), 1 deletion(-) diff --git a/tests/architecture/test_content_digest_single_owner.py b/tests/architecture/test_content_digest_single_owner.py index 039d36b0b7..e0a1f61faa 100644 --- a/tests/architecture/test_content_digest_single_owner.py +++ b/tests/architecture/test_content_digest_single_owner.py @@ -213,13 +213,18 @@ ) # A consumer that hands `re` a pattern this file cannot fold. Folding reads -# every construction except the declared marker reader, which is fenced below. +# every construction except the declared readers, each fenced below. UNREADABLE_CONSUMER_CONSTRUCTIONS: dict[str, str] = { "loopx.capabilities.benchmark_toolkit.integrity": ( "_marker_present matches an escaped caller-supplied filename/token inside " "command text with explicit basename boundaries; it does not validate a " "whole-value digest. Its sole dynamic regex is fenced to that reader." ), + "loopx.control_plane.testing.cli_output_semantics": ( + "authoring_input_observations matches an escaped, local literal instruction " + "inside a complete Markdown Todo line, not a whole-value digest. A focused " + "fence pins its sole unreadable call, surrounding grammar and literal input." + ), } # Reading `.pattern` off the owner and recompiling it states the decision a second time @@ -1229,6 +1234,35 @@ def test_every_declared_unreadable_consumer_site_is_still_there() -> None: assert unreadable, f"{dotted} has no unreadable construction left to declare" +def test_cli_output_dynamic_prose_pattern_is_fenced_to_its_literal_question() -> None: + dotted = "loopx.control_plane.testing.cli_output_semantics" + module = importlib.import_module(dotted) + tree = ast.parse(Path(module.__file__).read_text(encoding="utf-8")) + reader = next(node for node in tree.body if isinstance(node, ast.FunctionDef) + and node.name == "authoring_input_observations") + calls = [node for node in ast.walk(reader) if isinstance(node, ast.Call) + and isinstance(node.func, ast.Attribute) and isinstance(node.func.value, ast.Name) + and (node.func.value.id, node.func.attr) == ("re", "search")] + assert len(calls) == 1 + unreadable = [site for site in _module_of(dotted)["constructions"] if not site["readable"]] + assert [site["line"] for site in unreadable] == [calls[0].lineno] + # Fixed surrounding Markdown grammar makes this a full prose observation, + # never a whole-value digest validator. The escaped middle must stay local + # literal text, not owner.pattern, a parameter or a provider's regex. + pattern = calls[0].args[0] + assert isinstance(pattern, ast.BinOp) and isinstance(pattern.op, ast.Add) + assert isinstance(pattern.right, ast.Constant) and pattern.right.value == "$" + assert isinstance(pattern.left, ast.BinOp) and isinstance(pattern.left.op, ast.Add) + assert isinstance(pattern.left.left, ast.Constant) + assert pattern.left.left.value == r"^3\. Todos: [^\n]*" + escaped = pattern.left.right + assert ast.unparse(escaped) == "re.escape(first_delivery_guidance)" + literal = next(node.value for node in reader.body if isinstance(node, ast.Assign) + and any(isinstance(target, ast.Name) and target.id == "first_delivery_guidance" + for target in node.targets)) + assert isinstance(literal, ast.Constant) and isinstance(literal.value, str) and literal.value + + # --- layer 4: behaviour, per spelling and per surface ------------------------------------- diff --git a/tests/test_projection_envelope.py b/tests/test_projection_envelope.py index 86628f275b..62a309f987 100644 --- a/tests/test_projection_envelope.py +++ b/tests/test_projection_envelope.py @@ -503,6 +503,8 @@ def test_global_risks_empty_contract_container_is_a_read_source( } ), ], + ids=["malformed-json", "array", "wrong-schema", "invalid-timestamp", + "invalid-continuation", "wrong-goal"], ) def test_global_risks_discloses_unreadable_host_poll_receipts_and_recovers( tmp_path: Path, @@ -563,6 +565,41 @@ def test_global_risks_discloses_unreadable_host_poll_receipts_and_recovers( assert recovered["alert"] is False +def test_receipt_case_collection_does_not_depend_on_the_collecting_process_clock(tmp_path): + import os + import subprocess + import sys + + # Two serial collectors model the independent imports in xdist workers, + # without starting a second worker pool inside a running test shard. + (tmp_path / "receipt_collection_clock.py").write_text( + "import os\n" + "import loopx.control_plane.projection_envelope_facts as facts\n" + "def pytest_configure(config):\n" + " facts.now_utc_iso = lambda: os.environ['LOOPX_RECEIPT_COLLECTION_CLOCK']\n", + encoding="utf-8", + ) + target = ( + "tests/test_projection_envelope.py::" + "test_global_risks_discloses_unreadable_host_poll_receipts_and_recovers" + ) + collections = [] + for clock in ("2026-01-01T00:00:00Z", "2026-01-01T00:00:01Z"): + result = subprocess.run( + [sys.executable, "-m", "pytest", target, "--collect-only", "-q", + "-o", "addopts=", "-p", "receipt_collection_clock"], + cwd=Path(__file__).resolve().parents[1], + env={**os.environ, "LOOPX_RECEIPT_COLLECTION_CLOCK": clock, + "PYTHONPATH": os.pathsep.join((str(tmp_path), os.environ.get("PYTHONPATH", "")))}, + capture_output=True, text=True, timeout=60, + ) + assert result.returncode == 0, result.stdout + result.stderr + nodeids = [line for line in result.stdout.splitlines() if line.startswith(target + "[")] + assert len(nodeids) == len(set(nodeids)) == 6 + collections.append(nodeids) + assert collections[0] == collections[1], "independent worker clocks changed collection identity" + + def test_real_cli_global_membership_failure_and_recovery(tmp_path): """Global scope must never use local membership as its denominator.""" import subprocess