test: repair authority fixtures and deterministic CI collection

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 -------------------------------------
 
 

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")

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),

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"
 

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,13 +50,25 @@ 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()
     writes = list(provider.writes)
     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"]
 
 

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

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