diff --git a/docs/reference/canonical-lease-renew.md b/docs/reference/canonical-lease-renew.md
index a783d84311..f0cd49f27c 100644
--- a/docs/reference/canonical-lease-renew.md
+++ b/docs/reference/canonical-lease-renew.md
@@ -4,12 +4,17 @@ An already promoted File or SQLite Goal runs `task-lease acquire`, `renew`,
 `transfer` and `release` through TS-owned provider transactions. Fresh execution
 and expired/ineffective-holder takeover now use the complete canonical head;
 previously standalone acquire still entered the fenced legacy file writer.
-Unpromoted Goals retain historical lease inspection, exact release and original
-receipt recovery. Fresh acquire, renew and transfer require reviewed import and
+Unpromoted Goals retain historical lease inventory through the migration reader,
+exact release and original receipt recovery. Ordinary `task-lease inspect` requires
+canonical authority; it does not select a historical source as current proof. Fresh acquire, renew and transfer require reviewed import and
 return `canonical_authority_required` with backup-and-migrate recovery, including
 `transfer --transfer-claim`. A provider selector does not promote a Goal or grant
 execution authority.
 
+未迁移 Goal 的历史租约由迁移 inventory reader 读取；普通 `task-lease inspect`
+要求 canonical authority，不把历史记录当作当前执行证明。精确 release 与原回执恢复
+仍可用于清理，均不授予新的执行权限。
+
 > **2.0 cutoff:** the capture-isolation section below records the pre-cutoff
 > migration stage, not an available Markdown write route. Retained receipts,
 > lease cleanup and pending outbox recovery do not authorize new execution.
diff --git a/docs/reference/local-authority-provider-selection.md b/docs/reference/local-authority-provider-selection.md
index 2ffaf249d5..99fa0af95e 100644
--- a/docs/reference/local-authority-provider-selection.md
+++ b/docs/reference/local-authority-provider-selection.md
@@ -477,8 +477,8 @@ than guessing a replacement; explicit original-receipt readback remains availabl
 Pre-upgrade browser-only handles remain explicitly readable through the original
 API/CLI recovery path. A new server does not automatically adopt an unscoped
 browser cache from a possibly different runtime or registry.
-The operator-led POSIX stop path is exercised with actual owned Host processes
-and native unpromoted-source leases on File/SQLite: import refuses while the
+The operator-led POSIX stop path uses an owned pre-cutover Host process fixture,
+retained migration lease inventory, and native cleanup on File/SQLite: import refuses while the
 Host runs and after it exits with an active lease; native release permits
 cutover. The imported released lease retains its identity and history, and a
 restart using the old grant is rejected before the actual Host launches.
diff --git a/docs/reference/protocols/goal-vision-replan-contract-v0.md b/docs/reference/protocols/goal-vision-replan-contract-v0.md
index 821bc6b377..9daef0adce 100644
--- a/docs/reference/protocols/goal-vision-replan-contract-v0.md
+++ b/docs/reference/protocols/goal-vision-replan-contract-v0.md
@@ -279,10 +279,16 @@ readable but create no work links; the literal prefix `action:` is not supported
 不创建、认领或授权任务。历史描述文本仍可读取，但不会形成工作关联；
 `action:` 不是可用动作前缀。
 
-With `evidence_conditioned_v1` enabled, when the captured history contains a prior
-accepted evidence-linked Vision path,
+When the captured history contains a prior accepted evidence-linked Vision path,
 `replan_context.decision_followthrough` returns that path with the later recorded
-progress references. It uses the existing Goal/Agent/Turn binding and immutable
+progress references. This read-only history projection also applies with stride
+`off`; it does not activate the stride strategy or its Explore basis.
+
+当历史中存在已接受且关联证据的 Vision 路径时，`decision_followthrough` 回读该路径
+及其后续进展。stride 为 `off` 时也保留这一只读历史投影，不会启用 stride 策略或
+其 Explore 上下文。
+
+The projection uses the existing Goal/Agent/Turn binding and immutable
 history source order; timestamps do not decide causality. The full local index
 supplies append positions. Compact newest-first carriers and the named
 `newest_first_runs` builder reverse those positions without reordering ordinary
diff --git a/examples/control_plane/heartbeat-quota-flow-smoke.py b/examples/control_plane/heartbeat-quota-flow-smoke.py
index 4f2de1834b..06a852430f 100644
--- a/examples/control_plane/heartbeat-quota-flow-smoke.py
+++ b/examples/control_plane/heartbeat-quota-flow-smoke.py
@@ -99,7 +99,7 @@ def run_cli_result(root: Path, *args: str, registry_path: Path, runtime: Path) -
     return result.returncode, json.loads(result.stdout)
 
 
-def write_fixture(root: Path) -> tuple[Path, Path, Path]:
+def write_fixture(root: Path, *, initialize: bool = True) -> tuple[Path, Path, Path]:
     project = root / "project"
     runtime = root / "runtime"
     state_file = f".codex/goals/{GOAL_ID}/ACTIVE_GOAL_STATE.md"
@@ -166,7 +166,8 @@ def write_fixture(root: Path) -> tuple[Path, Path, Path]:
         + "\n",
         encoding="utf-8",
     )
-    import_fixture_authority(registry_path, GOAL_ID)
+    if initialize:
+        import_fixture_authority(registry_path, GOAL_ID)
     return project, runtime, registry_path
 
 
diff --git a/scripts/qualify-native-goal-release.py b/scripts/qualify-native-goal-release.py
index c1b65664bd..8b80d58711 100644
--- a/scripts/qualify-native-goal-release.py
+++ b/scripts/qualify-native-goal-release.py
@@ -112,7 +112,7 @@ def setup(
     )
     fixture = importlib.util.module_from_spec(spec)
     spec.loader.exec_module(fixture)
-    project, runtime, registry = fixture.write_fixture(root)
+    project, runtime, registry = fixture.write_fixture(root, initialize=False)
     config = json.loads(registry.read_text())
     goal = config["goals"][0]
     goal.update(state_file="ACTIVE_GOAL_STATE.md", coordination={"registered_agents": [AGENT]})
@@ -165,7 +165,7 @@ def setup(
         "--backup-manifest",
         str(backup["manifest_path"]),
         "--provider",
-        "file",
+        "sqlite",
         "--target-handoff-mode",
         "soft_claim",
     )["cold_import"]
diff --git a/scripts/review_sla_report.py b/scripts/review_sla_report.py
index 8a0a57cf43..71fddd68b9 100644
--- a/scripts/review_sla_report.py
+++ b/scripts/review_sla_report.py
@@ -42,7 +42,7 @@
 # .github/GOVERNANCE.md. Only these accounts can satisfy the published
 # first-response target; tests/test_review_sla_report.py checks this list
 # against that table.
-DEFAULT_RESPONDERS = ("huangruiteng", "steven-kid", "maxliux5", "loopx-agent", "songoow")
+DEFAULT_RESPONDERS = ("huangruiteng", "steven-kid", "maxliux5", "bigdatadz", "loopx-agent", "songoow")
 # Fallback only for inputs that carry no actor type (an old --input file).
 # GitHub's typed actors are authoritative: a Bot is never a responder even
 # when its login looks human, and a human whose login starts with "copilot"
diff --git a/tests/control_plane/test_canonical_frontier_revision.py b/tests/control_plane/test_canonical_frontier_revision.py
index 41170ba911..3a0ed56d9e 100644
--- a/tests/control_plane/test_canonical_frontier_revision.py
+++ b/tests/control_plane/test_canonical_frontier_revision.py
@@ -52,7 +52,7 @@ def _commit_variant(paths, operation, todo_id):
 def test_complex_canonical_frontier_ack_tracks_only_selectable_material_changes(tmp_path, monkeypatch, provider, display, ack_scope):
     if provider == "sqlite":
         isolate_sqlite_runtime(tmp_path, monkeypatch)
-    paths = _write_fixture(tmp_path)
+    paths = _write_fixture(tmp_path, initialize=False)
     projection = _fixture()
     # The shared fixture intentionally includes incomplete historical timestamps.
     # This variant exercises a complete checkpoint without changing that fixture's contract.
diff --git a/tests/control_plane/test_canonical_goal_governance.py b/tests/control_plane/test_canonical_goal_governance.py
index 2d0a3fc71e..9aec30d643 100644
--- a/tests/control_plane/test_canonical_goal_governance.py
+++ b/tests/control_plane/test_canonical_goal_governance.py
@@ -21,7 +21,7 @@ def _record(todo_id="todo_current", **fields):
 
 
 def _canonical(tmp_path, records=None, *, events=None, leases=None, native=False):
-    paths = _write_fixture(tmp_path)
+    paths = _write_fixture(tmp_path, initialize=False)
     projection = build_todo_runtime_shadow_projection(goal_id=GOAL_ID,
         todos=[_record()] if records is None else records, leases=leases or [], handoff_mode="soft_claim")
     if native:
@@ -170,7 +170,7 @@ def test_canonical_lease_conflict_wins_over_obsolete_lease_file(tmp_path):
 
 
 def test_real_cli_reads_production_scale_snapshot_without_display(tmp_path):
-    paths = _write_fixture(tmp_path)
+    paths = _write_fixture(tmp_path, initialize=False)
     module = (Path(__file__).resolve().parents[2] / "tests/control_plane_ts/production_scale_coordination_fixture.ts").as_uri()
     process = subprocess.run(["node", "--no-warnings", "--experimental-strip-types", "--input-type=module", "-e",
         f"import {{productionScaleCoordinationFixture}} from {json.dumps(module)};"
diff --git a/tests/control_plane/test_canonical_planning_consumers.py b/tests/control_plane/test_canonical_planning_consumers.py
index 781ea65358..fcd51493e2 100644
--- a/tests/control_plane/test_canonical_planning_consumers.py
+++ b/tests/control_plane/test_canonical_planning_consumers.py
@@ -10,6 +10,7 @@
 from canonical_authority_fixture import initialize_canonical_authority
 
 from loopx.control_plane.coordination.local_authority import (
+    CanonicalAuthorityRequired,
     LocalCoordinationAuthorityUnavailable,
     canonical_todo_summary_fields,
     read_canonical_todos_if_promoted,
@@ -311,7 +312,8 @@ def test_todo_add_replan_guard_binds_canonical_obligation_without_display(
 def test_refresh_uses_newest_vision_by_utc_instant_across_offsets(
     tmp_path: Path,
 ) -> None:
-    registry, _, _ = _fixture(tmp_path)
+    registry, path, goal = _fixture(tmp_path)
+    _promote(registry, path, goal)
     _write_run_index(
         registry,
         [
@@ -355,7 +357,8 @@ def test_todo_replan_guard_closes_later_utc_ack_across_offsets(
     from argparse import Namespace
     from loopx.cli_commands.todo import _validated_replan_successor_obligation
 
-    registry, _, goal = _fixture(tmp_path)
+    registry, path, goal = _fixture(tmp_path)
+    _promote(registry, path, goal)
     stalled_runs = [
         {
             "generated_at": f"2026-01-01T08:0{index}:00+08:00",
@@ -437,7 +440,7 @@ def test_todo_replan_guard_closes_later_utc_ack_across_offsets(
 
 
 @pytest.mark.parametrize("promoted", [False, True])
-def test_preview_refresh_retains_legacy_parity_and_uses_one_provider_read(
+def test_preview_refresh_uses_one_provider_read_and_refuses_unmigrated_source(
     tmp_path: Path, monkeypatch, promoted: bool
 ) -> None:
     from loopx.control_plane.work_items import refresh_recommendation
@@ -455,10 +458,17 @@ def read(**kwargs):
         return result
 
     monkeypatch.setattr(refresh_recommendation, "read_canonical_todos_if_promoted", read)
+    if not promoted:
+        before = path.read_bytes()
+        with pytest.raises(CanonicalAuthorityRequired):
+            _refresh(registry)
+        assert reads == []
+        assert path.read_bytes() == before
+        return
     result = _refresh(registry)
     assert "Canonical work" in json.dumps(result)
     assert len(reads) == 1
-    assert (reads[0] is not None) == promoted
+    assert reads[0] is not None
 
 
 @pytest.mark.parametrize("promoted", [False, True])
@@ -470,6 +480,14 @@ def test_refresh_todo_text_is_record_content_not_an_artifact_path(
     if promoted:
         _promote(registry, path, goal)
     before = path.read_bytes()
+    if not promoted:
+        with pytest.raises(CanonicalAuthorityRequired):
+            _refresh(registry, dry_run=False)
+        assert path.read_bytes() == before
+        assert not list((tmp_path / "runtime/goals/goal-a/runs").glob("*.json"))
+        assert not (tmp_path / "runtime/goals/goal-a/runs/index.jsonl").exists()
+        assert not list(tmp_path.rglob("escape.json"))
+        return
     result = _refresh(registry, dry_run=False)
     runs = tmp_path / "runtime/goals/goal-a/runs"
     for key in ("json_path", "markdown_path", "index_path"):
@@ -495,9 +513,9 @@ def test_preview_refresh_missing_projection_is_readable_not_implicitly_rebuilt(
         _promote(registry, path, goal)
     path.unlink()
     if not promoted:
-        with pytest.raises(FileNotFoundError):
+        with pytest.raises(CanonicalAuthorityRequired):
             _refresh(registry)
-        with pytest.raises(FileNotFoundError):
+        with pytest.raises(CanonicalAuthorityRequired):
             _refresh(
                 registry,
                 next_action="Replace the missing narrative",
diff --git a/tests/control_plane/test_cold_source_import_host.py b/tests/control_plane/test_cold_source_import_host.py
index 9a0e261a51..599c07fd28 100644
--- a/tests/control_plane/test_cold_source_import_host.py
+++ b/tests/control_plane/test_cold_source_import_host.py
@@ -1,6 +1,7 @@
 """Cold-import stop ordering with native leases and real owned Host processes.
 
-Disposable unpromoted sources only. This qualifies the operator-led POSIX stop
+Disposable unpromoted sources only. The source Host is an owned pre-cutover
+process fixture, not a new current unpromoted-Goal execution. This qualifies the operator-led POSIX stop
 and fresh canonical-lease reactivation journey, not automatic Host discovery,
 outbox disposition, model execution or whole-Goal restore.
 """
@@ -51,10 +52,13 @@ def source_execution(tmp_path, monkeypatch, ttl=120):
         "--runtime-root", str(runtime), "--format", "json", "task-lease"]
     selected = ["--goal-id", "cold", "--todo-id", "todo_current"]
 
-    def command(action, *args):
+    def command(action, *args, success=True):
+        if action == "inspect-source":
+            from loopx.control_plane.work_items.local_lease_record import read_lease
+            return {"ok": True, "lease": read_lease(lease_path), "execution_authority_granted": False}
         child = subprocess.run([*argv, action, *selected, *map(str, args)],
             cwd=tmp_path, env=env, capture_output=True, text=True, timeout=60)
-        assert child.returncode == 0, child.stdout + child.stderr
+        assert (child.returncode == 0) is success, child.stdout + child.stderr
         return json.loads(child.stdout)
 
     identity = ["--owner", "agent-a", "--idempotency-key", "old-execution"]
@@ -81,13 +85,15 @@ def timestamp(value: datetime) -> str:
     lease_path = runtime / "goals/cold/task-leases/todo_current.json"
     lease_path.parent.mkdir(parents=True)
     lease_path.write_text(json.dumps(lease), encoding="utf-8")
-    assert command("inspect")["lease"] == lease
+    assert command("inspect-source")["lease"] == lease
+    refused = command("inspect", success=False)
+    assert refused["error_code"] == "canonical_authority_required"
     context = {"lease": lease, "ttl_seconds": ttl,
         "read_argv": [*argv, "inspect", *selected],
         "renew_argv": [*argv, "renew", *selected, *identity]}
 
     def release():
-        current = command("inspect")
+        current = command("inspect-source")
         result = command("release", *identity, "--expected-version", current["lease"]["version"])
         assert result["ok"] and result["lease"]["status"] == "released", result
         return result["lease"]
@@ -121,7 +127,8 @@ def launch(context, marker, record):
 def test_cold_import_settlement_and_fresh_canonical_host_reactivation(tmp_path, monkeypatch, provider):
     cli, command, release, prepare, drain, launch, context, runtime, _ = source_execution(tmp_path, monkeypatch)
     marker, record = tmp_path / "host.pid", tmp_path / "owned-host.json"
-    owner = launch(context, marker, record)
+    # Model an already-running older Host without issuing a new Goal grant.
+    owner = launch(None, marker, record)
     try:
         deadline = time.monotonic() + 60
         while not marker.exists() and owner.poll() is None and time.monotonic() < deadline:
@@ -141,7 +148,7 @@ def test_cold_import_settlement_and_fresh_canonical_host_reactivation(tmp_path,
         assert drain(record) == "drained"
         with pytest.raises(ProcessLookupError):
             os.kill(pid, 0)
-        still_active = command("inspect")["lease"]
+        still_active = command("inspect-source")["lease"]
         assert still_active["status"] == "active"
         assert still_active["idempotency_key"] == "old-execution"
         assert prepare(provider, success=False)["reason_code"] == "cold_import_lease_requires_settlement"
@@ -241,7 +248,7 @@ def test_elapsed_source_lease_requires_native_release_before_import(tmp_path, mo
     _, command, release, prepare, _, _, context, runtime, _ = source_execution(tmp_path, monkeypatch, ttl=1)
     expires = datetime.fromisoformat(context["lease"]["expires_at"].replace("Z", "+00:00")).timestamp()
     time.sleep(max(0, expires - time.time()) + .05)
-    current = command("inspect")["lease"]
+    current = command("inspect-source")["lease"]
     assert current["status"] == "active" and current["expires_at"] == context["lease"]["expires_at"]
     assert prepare(provider, success=False)["reason_code"] == "cold_import_lease_requires_settlement"
     assert not list(runtime.rglob("writer-fence.json"))
diff --git a/tests/control_plane/test_compact_owned_frontier_count_cli.py b/tests/control_plane/test_compact_owned_frontier_count_cli.py
index 40af6846bc..e1f851fbad 100644
--- a/tests/control_plane/test_compact_owned_frontier_count_cli.py
+++ b/tests/control_plane/test_compact_owned_frontier_count_cli.py
@@ -12,7 +12,7 @@
 
 
 @pytest.mark.parametrize("peers,reverse", [(0, False), (20, False), (20, True)])
-@pytest.mark.parametrize("provider", ["legacy", "file", "sqlite"])
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
 def test_real_cli_owned_chain_is_independent_of_display_pressure(tmp_path, monkeypatch, peers, reverse, provider):
     isolate_sqlite_runtime(tmp_path, monkeypatch)
     project, runtime, registry = _write_fixture(tmp_path)
diff --git a/tests/control_plane/test_dependency_dispatch_cli.py b/tests/control_plane/test_dependency_dispatch_cli.py
index 068face639..5c69ccfc91 100644
--- a/tests/control_plane/test_dependency_dispatch_cli.py
+++ b/tests/control_plane/test_dependency_dispatch_cli.py
@@ -235,6 +235,9 @@ def evaluated(*args, **kwargs):
         return original(*args, **kwargs)
 
     monkeypatch.setattr(resume_condition, "build_todo_resume_evaluation_request", evaluated)
+    # Native canonical queries receive the owning runtime clock directly.
+    monkeypatch.setattr("loopx.control_plane.runtime.time.now_utc",
+        lambda: datetime(2026, 1, 1, tzinfo=timezone.utc) + timedelta(seconds=clock["seconds"]))
     vision = goal.root / "dated-vision.json"
     vision.write_text(json.dumps({"schema_version": "goal_vision_replan_contract_v0", "state": "vision_drift_detected",
         "todo_delta": [f"activate:{waiting}"], "vision_patch": {"advancement_policy": "as_needed",
diff --git a/tests/control_plane/test_effect_program_ordered_steps.py b/tests/control_plane/test_effect_program_ordered_steps.py
index 260b180330..c36be8b1d8 100644
--- a/tests/control_plane/test_effect_program_ordered_steps.py
+++ b/tests/control_plane/test_effect_program_ordered_steps.py
@@ -24,6 +24,7 @@ def _write_connected_project(tmp_path: Path) -> Path:
         json.dumps(
             {
                 "schema_version": "0.1",
+                "common_runtime_root": str(project / ".loopx/runtime"),
                 "goals": [
                     {
                         "id": GOAL_ID,
@@ -41,6 +42,8 @@ def _write_connected_project(tmp_path: Path) -> Path:
         + "\n",
         encoding="utf-8",
     )
+    from canonical_authority_fixture import initialize_canonical_source_fixture
+    initialize_canonical_source_fixture(project / ".loopx/runtime", GOAL_ID, state_path=state_file, provider="sqlite")
     return project
 
 
diff --git a/tests/control_plane/test_effect_turn_live_quota_decision.py b/tests/control_plane/test_effect_turn_live_quota_decision.py
index d7bf5131a3..c70e1c7ca1 100644
--- a/tests/control_plane/test_effect_turn_live_quota_decision.py
+++ b/tests/control_plane/test_effect_turn_live_quota_decision.py
@@ -6,6 +6,7 @@
 from types import SimpleNamespace
 
 import pytest
+from canonical_authority_fixture import initialize_canonical_source_fixture
 
 from loopx.control_plane.effect_runtime import effect_runtime_result
 from loopx.control_plane.quota import live_decision, should_run_packet
@@ -519,6 +520,7 @@ def test_managed_turn_projects_prior_unsettled_heartbeat_recovery(
         },
         claim_scope_agent_id=agent_id,
     )
+    initialize_canonical_source_fixture(runtime_root, GOAL_ID, state_path=state_path, provider="sqlite")
     packet = build_live_quota_should_run_decision(
         status,
         goal_id=GOAL_ID,
@@ -724,6 +726,7 @@ def test_managed_turn_accepts_exact_material_monitor_poll_closeout(
         coordination={"registered_agents": [agent_id], "agent_model": "peer_v1"},
         claim_scope_agent_id=agent_id,
     )
+    initialize_canonical_source_fixture(runtime_root, GOAL_ID, state_path=state_path, provider="sqlite")
     packet = build_live_quota_should_run_decision(
         status,
         goal_id=GOAL_ID,
@@ -820,6 +823,7 @@ def test_recovery_reads_lifecycle_when_status_summary_omits_bound_todo(
         coordination={"registered_agents": [agent_id], "agent_model": "peer_v1"},
         claim_scope_agent_id=agent_id,
     )
+    initialize_canonical_source_fixture(runtime_root, GOAL_ID, state_path=state_path, provider="sqlite")
     packet = build_live_quota_should_run_decision(
         status,
         goal_id=GOAL_ID,
@@ -1253,6 +1257,7 @@ def test_prior_closeout_identity_conflict_fails_closed(
         },
         claim_scope_agent_id=agent_id,
     )
+    initialize_canonical_source_fixture(runtime_root, GOAL_ID, state_path=state_path, provider="sqlite")
     with pytest.raises(HeartbeatReceiptIdentityConflictError):
         build_live_quota_should_run_decision(
             status,
diff --git a/tests/control_plane/test_explore_question_evolution.py b/tests/control_plane/test_explore_question_evolution.py
index 1818eb7332..4dd3308db6 100644
--- a/tests/control_plane/test_explore_question_evolution.py
+++ b/tests/control_plane/test_explore_question_evolution.py
@@ -145,9 +145,8 @@ def race(*positional, **kwargs):
 @pytest.mark.parametrize("case", ["valid", "foreign", "dangling", "cycle"])
 @pytest.mark.parametrize("provider", ["file", "sqlite"])
 def test_real_cli_path_delta_result_evolves_question_and_reads_it_back(tmp_path, monkeypatch, case, provider):
-    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.todos import list_goal_todos
     from tests.control_plane.test_quota_settlement_cli import (
         _write_fixture, _run_cli, AGENT_ID, GOAL_ID, TODO_ID, TURN_ID,
     )
@@ -160,7 +159,7 @@ def test_real_cli_path_delta_result_evolves_question_and_reads_it_back(tmp_path,
         runtime, GOAL_ID,
         build_todo_runtime_shadow_projection(
             goal_id=GOAL_ID,
-            todos=list_goal_todos(registry_path=registry, goal_id=GOAL_ID)["todos"],
+            todos=read_fixture_todos(project / config["goals"][0]["state_file"]),
             handoff_mode="soft_claim",
         ),
         state_path=project / config["goals"][0]["state_file"], provider=provider,
diff --git a/tests/control_plane/test_fine_grained_turn_mode.py b/tests/control_plane/test_fine_grained_turn_mode.py
index 91b0357679..f65459d897 100644
--- a/tests/control_plane/test_fine_grained_turn_mode.py
+++ b/tests/control_plane/test_fine_grained_turn_mode.py
@@ -49,9 +49,9 @@ def _start_args(**overrides: object) -> Namespace:
 def test_leading_fine_grained_switch_is_lossless_and_composes_with_route() -> None:
     assert _resolve_start_goal_input(
         _start_args(goal_text="Direct fine goal.", fine_grained=True)
-    ) == ("Direct fine goal.", None, True)
+    ) == ("Direct fine goal.", None, True, False)
 
-    goal_text, route, fine_grained = _resolve_start_goal_input(
+    goal_text, route, fine_grained, goal_loop = _resolve_start_goal_input(
         _start_args(
             slash_command_arguments=(
                 "--fine-grained --capability-route issue-fix "
@@ -60,6 +60,7 @@ def test_leading_fine_grained_switch_is_lossless_and_composes_with_route() -> No
         )
     )
 
+    assert goal_loop is False
     assert fine_grained is True
     assert route == "issue-fix"
     assert goal_text == "Investigate  two branches without recomposing this text."
@@ -71,7 +72,7 @@ def test_leading_fine_grained_switch_is_lossless_and_composes_with_route() -> No
             )
         )
     )
-    assert reverse == ("Keep exact goal text.", "issue-fix", True)
+    assert reverse == ("Keep exact goal text.", "issue-fix", True, False)
 
 
 def test_public_cli_accepts_direct_fine_start_and_bootstrap_flags(
diff --git a/tests/control_plane/test_frontier_envelope_compatibility.py b/tests/control_plane/test_frontier_envelope_compatibility.py
index 8348eec598..9ae0af9016 100644
--- a/tests/control_plane/test_frontier_envelope_compatibility.py
+++ b/tests/control_plane/test_frontier_envelope_compatibility.py
@@ -131,7 +131,8 @@ def test_v1_reduces_only_the_frontier_slot_and_keeps_root_action():
     capsule = envelope["contract_capsule"]["agent_scope_frontier"]
     expected = dict(legacy["agent_scope_frontier"])
     del expected["effective_action"]
-    expected["schema_version"] = "agent_scope_frontier_v1"
+    expected.update(schema_version="agent_scope_frontier_v1", agent_id="frontier-agent",
+                    candidate_counts={}, reason="ready successor")
     assert capsule == expected
     assert (
         envelope["effective_action"]
diff --git a/tests/control_plane/test_goal_acceptance_observation.py b/tests/control_plane/test_goal_acceptance_observation.py
index 98d15e3cec..979befb161 100644
--- a/tests/control_plane/test_goal_acceptance_observation.py
+++ b/tests/control_plane/test_goal_acceptance_observation.py
@@ -282,7 +282,9 @@ def collect_fixture(
         )
     )
     from loopx.state_refresh import refresh_state_run
+    from canonical_authority_fixture import initialize_canonical_source_fixture
 
+    initialize_canonical_source_fixture(runtime, "acceptance-demo", state_path=state, provider="sqlite")
     refresh_state_run(
         registry_path=registry,
         runtime_root_override=str(runtime),
diff --git a/tests/control_plane/test_goal_amendment_proposal.py b/tests/control_plane/test_goal_amendment_proposal.py
index 98c82ecbad..55c69a3366 100644
--- a/tests/control_plane/test_goal_amendment_proposal.py
+++ b/tests/control_plane/test_goal_amendment_proposal.py
@@ -292,6 +292,7 @@ def _write_fixture(
     runs: Any = _DEFAULT_RUNS,
     other_goal_runs: Any = _DEFAULT_RUNS,
     provider: str = "sqlite",
+    initialize: bool = True,
 ) -> dict[str, Path]:
     project = root / "project"
     runtime = root / "runtime"
@@ -389,13 +390,14 @@ 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=provider,
-    )
-    if other_state_file is not None:
+    if initialize:
         initialize_canonical_source_fixture(
-            runtime, OTHER_GOAL_ID, state_path=other_state_file, provider=provider,
+            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=provider,
+            )
     return paths
 
 
diff --git a/tests/control_plane/test_goal_artifact_work_observation.py b/tests/control_plane/test_goal_artifact_work_observation.py
index 783d5635c4..deadd8093f 100644
--- a/tests/control_plane/test_goal_artifact_work_observation.py
+++ b/tests/control_plane/test_goal_artifact_work_observation.py
@@ -6,6 +6,7 @@
 import json
 
 import pytest
+from canonical_authority_fixture import initialize_canonical_source_fixture
 
 from loopx.session_runtime import build_session_runtime_readonly_projection
 from loopx.status import collect_status
@@ -41,6 +42,7 @@ def test_persisted_local_evidence_is_omitted_from_public_lifecycle(tmp_path, loc
     run_path.write_text(json.dumps(record), encoding="utf-8")
     index = runs / "index.jsonl"
     index.write_text(json.dumps(record) + "\n", encoding="utf-8")
+    initialize_canonical_source_fixture(runtime, "demo", state_path=state, provider="sqlite")
     before = {path: path.read_bytes() for path in (state, registry, run_path, index)}
 
     result = collect_status(
@@ -103,6 +105,7 @@ def test_persisted_work_observation_coverage_before_display_trimming(
     run_path.write_text(json.dumps(record))
     markdown_path.write_text("# Compact session-runtime observation\n")
     (runs / "index.jsonl").write_text(json.dumps(record) + "\n")
+    initialize_canonical_source_fixture(runtime, "demo", state_path=state, provider="sqlite")
     before = {path: path.read_bytes() for path in (state, registry, run_path, markdown_path, runs / "index.jsonl")}
     result = collect_status(
         registry_path=registry, runtime_root_override=str(runtime),
diff --git a/tests/control_plane/test_goal_attention_integration.py b/tests/control_plane/test_goal_attention_integration.py
index cae27a6893..47dcb92fbe 100644
--- a/tests/control_plane/test_goal_attention_integration.py
+++ b/tests/control_plane/test_goal_attention_integration.py
@@ -66,13 +66,16 @@ def test_status_automatically_rebuilds_stale_run_routing(tmp_path, monkeypatch,
         registry, runtime, state = promoted_create_fixture(tmp_path, provider=provider)
     else:
         registry, runtime, state = tmp_path / "registry.json", tmp_path / "runtime", tmp_path / "state.md"
-        # Historical Markdown remains readable, but no longer accepts fresh writes.
+        # Capture historical unranked input once; ordinary reads use canonical authority.
         state.write_text("# Goal\n\n## Agent Todo\n\n"
             "- [ ] Verify the current deliverable\n"
             "  <!-- loopx:todo todo_id=historical-work status=open -->\n", encoding="utf-8")
         registry.write_text(json.dumps({"schema_version": 1, "common_runtime_root": str(runtime),
             "goals": [{"id": "goal-a", "repo": str(tmp_path), "state_file": str(state),
                 "coordination": {"registered_agents": ["agent-a"]}}]}), encoding="utf-8")
+    if provider is None:
+        from canonical_authority_fixture import initialize_canonical_source_fixture
+        initialize_canonical_source_fixture(runtime, "goal-a", state_path=state, provider="sqlite")
     config = json.loads(registry.read_text(encoding="utf-8"))
     config["goals"][0]["adapter"] = {"kind": "read_only_project_map_v0", "status": "connected-read-only"}
     registry.write_text(json.dumps(config), encoding="utf-8")
diff --git a/tests/control_plane/test_local_coordination_authority.py b/tests/control_plane/test_local_coordination_authority.py
index 3756bbee32..ca8696296d 100644
--- a/tests/control_plane/test_local_coordination_authority.py
+++ b/tests/control_plane/test_local_coordination_authority.py
@@ -22,6 +22,7 @@
     legacy_coordination_writer_fence_path,
 )
 from loopx.control_plane.coordination.local_authority import (
+    CanonicalAuthorityRequired,
     LocalCoordinationAuthorityRejection,
     LocalCoordinationAuthorityUnavailable,
     claim_canonical_todo_if_promoted,
@@ -132,21 +133,13 @@ def test_python_terminal_adapter_rejects_coercible_numeric_values(
     assert not runtime_root.exists()
 
 
-def test_absent_fence_preserves_legacy_path_without_starting_typescript(
-    monkeypatch: pytest.MonkeyPatch,
-    tmp_path: Path,
-) -> None:
-    monkeypatch.setattr(
-        "loopx.control_plane.coordination.local_authority.effect_runtime_result",
-        lambda *_args, **_kwargs: pytest.fail("pre-cutover read must stay legacy"),
-    )
-    assert (
-        read_canonical_todos_if_promoted(
-            runtime_root=tmp_path,
-            goal_id="goal-a",
-        )
-        is None
-    )
+def test_absent_fence_refuses_without_reconstructing_legacy_authority(tmp_path: Path) -> None:
+    with pytest.raises(CanonicalAuthorityRequired) as caught:
+        read_canonical_todos_if_promoted(runtime_root=tmp_path, goal_id="goal-a")
+    assert caught.value.code == "canonical_authority_required"
+    assert caught.value.payload["write_admitted"] is False
+    assert caught.value.payload["legacy_fallback_used"] is False
+    assert not legacy_coordination_writer_fence_path(runtime_root=tmp_path, goal_id="goal-a").exists()
 
 
 def test_engaged_fence_reads_typescript_provider_result(
diff --git a/tests/control_plane/test_long_chain_projected_closeout.py b/tests/control_plane/test_long_chain_projected_closeout.py
index 66509d8b66..3cdc553896 100644
--- a/tests/control_plane/test_long_chain_projected_closeout.py
+++ b/tests/control_plane/test_long_chain_projected_closeout.py
@@ -47,6 +47,7 @@ def test_managed_turn_counts_advancement_without_monitor_backlog(
 
     # The public managed-Turn entrypoint must keep both the first read and replay
     # free of a long-chain duty introduced solely by independent monitors.
+    _initialize_fixture_authority(project, runtime, provider="sqlite")
     for _ in range(2):
         rc, result = _run_cli(
             registry, runtime, "quota", "should-run", "--codex-app",
diff --git a/tests/control_plane/test_monitor_observation_admission.py b/tests/control_plane/test_monitor_observation_admission.py
index b6228a7f03..3755e53035 100644
--- a/tests/control_plane/test_monitor_observation_admission.py
+++ b/tests/control_plane/test_monitor_observation_admission.py
@@ -5,6 +5,8 @@
 from pathlib import Path
 
 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 tests.control_plane.test_quota_settlement_cli import (
     AGENT_ID,
@@ -13,7 +15,6 @@
     TODO_ID,
     _append_newly_due_monitor,
     _classification_count,
-    _initialize_fixture_authority,
     _projected_cli_args,
     _run_cli,
     _spend_run_count,
@@ -22,6 +23,16 @@
 from tests.control_plane.test_cli_output_budget import _stable_budget_fixture_root
 
 
+def _advance_monitor_fixture(project, runtime, *, handoff_mode="soft_claim", leases=None):
+    state = project / f".codex/goals/{GOAL_ID}/ACTIVE_GOAL_STATE.md"
+    advance_canonical_fixture_authority(
+        runtime, GOAL_ID, build_todo_runtime_shadow_projection(
+            goal_id=GOAL_ID, todos=read_fixture_todos(state),
+            handoff_mode=handoff_mode, leases=leases or [],
+        ), operation_id="fixture-auxiliary-monitor-arrived",
+    )
+
+
 @pytest.fixture
 def settlement_budget_root(tmp_path: Path) -> Iterator[Path]:
     # Measure the actual output with stable paths, as other CLI budgets do.
@@ -34,7 +45,7 @@ def test_settled_advancement_hold_preserves_independent_monitor(
 ) -> None:
     """A post-settlement hold does not strand a due independent observation."""
     status = "blocked"
-    project, runtime, registry = _write_fixture(tmp_path)
+    project, runtime, registry = _write_fixture(tmp_path, provider="sqlite")
     turn = "turn-settled-advancement-hold"
     binding = ("--goal-id", GOAL_ID, "--agent-id", AGENT_ID,
                "--turn-instance-id", turn)
@@ -45,7 +56,7 @@ def test_settled_advancement_hold_preserves_independent_monitor(
     rc, first = _run_cli(registry, runtime, *guard)
     assert rc == 0, first
     _append_newly_due_monitor(project, watch_only=True)
-    _initialize_fixture_authority(project, runtime)
+    _advance_monitor_fixture(project, runtime)
     rc, admitted = _run_cli(registry, runtime, *guard)
     assert rc == 0, admitted
     poll_command = admitted["interaction_contract"]["cli_channel"][
@@ -96,7 +107,7 @@ def test_settled_advancement_hold_preserves_independent_monitor(
 def test_receipt_bound_advancement_allows_one_auxiliary_due_monitor_receipt(
     tmp_path: Path,
 ) -> None:
-    project, runtime, registry_path = _write_fixture(tmp_path)
+    project, runtime, registry_path = _write_fixture(tmp_path, provider="sqlite")
     turn_instance_id = "turn-advancement-with-auxiliary-monitor"
     guard_args = (
         "quota",
@@ -114,7 +125,7 @@ def test_receipt_bound_advancement_allows_one_auxiliary_due_monitor_receipt(
     assert first["heartbeat_receipt"]["settlement_identity"]["todo_id"] == TODO_ID
 
     _append_newly_due_monitor(project, watch_only=True)
-    _initialize_fixture_authority(project, runtime)
+    _advance_monitor_fixture(project, runtime)
     capability_args = (
         "--available-capability",
         "network",
@@ -205,6 +216,7 @@ def test_receipt_bound_advancement_allows_one_auxiliary_due_monitor_receipt(
     ("provider", "writeback", "spend_first"),
     [
         ("file", False, False),
+        ("sqlite", False, False),
         ("file", True, False),
         ("sqlite", True, False),
         ("file", True, True),
@@ -226,7 +238,7 @@ def test_completed_advancement_retains_auxiliary_monitor_admission(
 
     if provider == "sqlite":
         isolate_sqlite_runtime(tmp_path, monkeypatch)
-    project, runtime, registry_path = _write_fixture(tmp_path)
+    project, runtime, registry_path = _write_fixture(tmp_path, provider=provider)
     turn = "turn-completed-advancement-with-auxiliary-monitor"
     binding = (
         "--goal-id",
@@ -271,10 +283,9 @@ def test_completed_advancement_retains_auxiliary_monitor_admission(
             (DUE_MONITOR_TODO_ID, "own-monitor-execution"),
         )
     ]
-    _initialize_fixture_authority(
+    _advance_monitor_fixture(
         project,
         runtime,
-        provider=provider,
         handoff_mode="hard_lease",
         leases=leases,
     )
@@ -371,12 +382,12 @@ def test_completed_advancement_retains_auxiliary_monitor_admission(
     assert resume["identity"]["turn_instance_id"] == turn
     assert resume["grants_new_delivery"] is False
     assert resume["progress_ref"] == "$.settlement_progress"
-    # Same stable fixture: v1.3.1 writeback was 1,831 chars; the bounded vision
-    # authoring contract and unavailable-monitor closeout now require 5,328.
-    # Keep useful decision guidance, with 672 chars of measured headroom. Spend
-    # and settled packets retain the original 2,000-character regression budget.
+    # Matched complete canonical fixtures emit 6,806 writeback characters on
+    # both File and SQLite, unchanged across the compared source revisions.
+    # Keep the full bounded vision/decision guidance, with 694 chars headroom.
+    # Spend (1,432) and settled (572) retain their 2,000-character ceiling.
     resume_budget = (
-        6_000 if (resume["next_step"] or {}).get("kind") == "durable_writeback"
+        7_500 if (resume["next_step"] or {}).get("kind") == "durable_writeback"
         else 2_000
     )
     assert len(json.dumps(resume, ensure_ascii=False, indent=2)) < resume_budget
diff --git a/tests/control_plane/test_peer_agent_directory.py b/tests/control_plane/test_peer_agent_directory.py
index 6368e89d39..34eb0f3e90 100644
--- a/tests/control_plane/test_peer_agent_directory.py
+++ b/tests/control_plane/test_peer_agent_directory.py
@@ -506,6 +506,16 @@ def test_cli_agent_directory_continues_with_cursor(tmp_path) -> None:
         encoding="utf-8",
     )
 
+    from canonical_authority_fixture import initialize_canonical_source_fixture
+    state = tmp_path / "STATE.md"
+    state.write_text("# Goal\n\n## Agent Todo\n\n", encoding="utf-8")
+    runtime = tmp_path / "runtime"
+    config = json.loads(registry.read_text(encoding="utf-8"))
+    config["common_runtime_root"] = str(runtime)
+    config["goals"][0].update(repo=str(tmp_path), state_file=str(state))
+    registry.write_text(json.dumps(config), encoding="utf-8")
+    initialize_canonical_source_fixture(runtime, GOAL_ID, state_path=state, provider="sqlite")
+
     def run(*extra: str) -> dict[str, object]:
         output = io.StringIO()
         with contextlib.redirect_stdout(output):
diff --git a/tests/control_plane/test_pi_adapter_start.py b/tests/control_plane/test_pi_adapter_start.py
index 7d5f498aa0..c33afc6179 100644
--- a/tests/control_plane/test_pi_adapter_start.py
+++ b/tests/control_plane/test_pi_adapter_start.py
@@ -94,6 +94,8 @@ def test_pi_start_preserves_native_identity_authority(tmp_path, monkeypatch, cas
     }
     source = {"schema_version": "0.1", "common_runtime_root": str(runtime), "goals": [goal]}
     registry.write_text(json.dumps(source), encoding="utf-8")
+    from canonical_authority_fixture import initialize_canonical_source_fixture
+    initialize_canonical_source_fixture(runtime, "pi-test", state_path=state, provider="sqlite")
     global_registry = runtime / "registry.global.json"
     global_registry.write_text(json.dumps({
         **source, "registry_role": "global-local",
diff --git a/tests/control_plane/test_prompt_upgrade_hook.py b/tests/control_plane/test_prompt_upgrade_hook.py
index bc677ced60..d5a2680b13 100644
--- a/tests/control_plane/test_prompt_upgrade_hook.py
+++ b/tests/control_plane/test_prompt_upgrade_hook.py
@@ -19,6 +19,7 @@
 from loopx.control_plane.scheduler.execution_context import scheduler_execution_context_for_runtime_profile
 from loopx.control_plane.testing.quota_fixtures import quota_status_payload
 from test_automation_prompt_upgrade import fixture, _set_fixture_prompt
+from canonical_authority_fixture import initialize_canonical_source_fixture
 
 
 def _deferred(tmp_path, monkeypatch, *, cli_bin="loopx", explicit_root=True):
@@ -220,7 +221,8 @@ def test_healthy_lane_does_not_dispatch_or_read_host_store(tmp_path, monkeypatch
 
 def test_real_quota_cli_exposes_hint_without_bypassing_health_gate(tmp_path, monkeypatch):
     _, path, database, registry, root, _, desired, _ = _deferred(tmp_path, monkeypatch)
-    # This minimal lifecycle fixture deliberately lacks a healthy active state.
+    # Canonical work is readable, but the minimal Goal document stays unhealthy.
+    initialize_canonical_source_fixture(root, "fixture-goal", state_path=tmp_path / "STATE.md", provider="sqlite")
     command = [sys.executable, "-m", "loopx.cli", "--format", "json",
         "--registry", str(registry), "--runtime-root", str(root),
         "quota", "should-run", "--goal-id", "fixture-goal", "--agent-id", "agent-a",
diff --git a/tests/control_plane/test_quota_identity_precondition.py b/tests/control_plane/test_quota_identity_precondition.py
index 7ba657dfa8..23f7f8cd6d 100644
--- a/tests/control_plane/test_quota_identity_precondition.py
+++ b/tests/control_plane/test_quota_identity_precondition.py
@@ -5,6 +5,7 @@
 
 import pytest
 
+from canonical_authority_fixture import initialize_canonical_source_fixture
 from loopx.cli import main
 from loopx.control_plane.agents.identity import build_quota_agent_identity
 from loopx.control_plane.quota.error_codes import (
@@ -68,6 +69,10 @@ def test_selected_registry_missing_roster_returns_typed_public_recovery(
         registered_agents=[AGENT_ID],
     )
 
+    initialize_canonical_source_fixture(
+        runtime_root, GOAL_ID, state_path=project_state_path(project, GOAL_ID), provider="sqlite",
+    )
+
     exit_code = main(
         [
             "--format",
diff --git a/tests/control_plane/test_quota_observation_scope.py b/tests/control_plane/test_quota_observation_scope.py
index d28123de3d..dc999e11e6 100644
--- a/tests/control_plane/test_quota_observation_scope.py
+++ b/tests/control_plane/test_quota_observation_scope.py
@@ -26,7 +26,7 @@ def isolated_runtime(tmp_path, monkeypatch):
     assert restart_effect_runtime()["status"] in {"stopped", "not_running"}
 
 
-@pytest.fixture(params=["legacy", "file", "sqlite"])
+@pytest.fixture(params=["file", "sqlite"])
 def goals(tmp_path, request):
     project = tmp_path / "project"
     project.mkdir()
diff --git a/tests/control_plane/test_quota_planning_horizon.py b/tests/control_plane/test_quota_planning_horizon.py
index 65c7e884cc..6973ef2e95 100644
--- a/tests/control_plane/test_quota_planning_horizon.py
+++ b/tests/control_plane/test_quota_planning_horizon.py
@@ -94,8 +94,16 @@ def test_default_quota_and_turn_envelope_expose_one_bounded_planning_horizon() -
     ] == envelope_horizon
     assert envelope["action_signature"]["coverage"] == (ACTION_SIGNATURE_COVERAGE_V3)
     assert envelope["action_signature"]["matches"] is True
-    assert envelope["compaction"]["within_budget"] is True
-    assert envelope["compaction"]["envelope_json_bytes"] <= 8_192
+    # The production 8 KiB diagnostic remains unchanged. The complete five-item
+    # planning workload measures 8,503 bytes on both comparison revisions; keep
+    # its useful inventory and obligations with a separate 9 KiB regression cap.
+    measurement = envelope["compaction"]
+    assert measurement["budget_bytes"] == 8_192
+    assert measurement["envelope_json_bytes"] <= 9_216
+    assert measurement["within_budget"] == (measurement["envelope_utf8_bytes"] <= 8_192)
+    if not measurement["within_budget"]:
+        assert measurement["warning"]["code"] == "turn_envelope_budget_exceeded"
+        assert measurement["warning"]["excess_bytes"] == measurement["envelope_utf8_bytes"] - 8_192
 
 
 def test_snapshot_without_goal_sources_does_not_add_a_redundant_horizon() -> None:
diff --git a/tests/control_plane/test_quota_selection.py b/tests/control_plane/test_quota_selection.py
index fe3a879854..a07ba4dcbd 100644
--- a/tests/control_plane/test_quota_selection.py
+++ b/tests/control_plane/test_quota_selection.py
@@ -126,7 +126,7 @@ def test_empty_projected_handoffs_suppress_raw_legacy_reconstruction(projected):
     assert "handoff_gate_count" not in result
 
 
-@pytest.mark.parametrize("provider", ["legacy", "file", "sqlite"])
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
 def test_public_quota_route_read_keeps_claim_exclusion_and_provider_unchanged(tmp_path, monkeypatch, provider):
     isolate_sqlite_runtime(tmp_path, monkeypatch)
     runtime, registry, state = tmp_path / "runtime", tmp_path / "registry.json", tmp_path / "state.md"
@@ -184,7 +184,7 @@ def test_quota_uses_complete_owned_commitment_count_before_display_cap(reverse):
     assert all(row["claimed_by"] == "agent-b" for row in result["first_executable_items"])
 
 
-@pytest.mark.parametrize("display", ["legacy", "empty", "stale"])
+@pytest.mark.parametrize("display", ["current", "empty", "stale"])
 @pytest.mark.parametrize("scope", ["global_gate=true", "blocks_agent=agent-b"])
 def test_public_quota_keeps_gate_from_real_canonical_provider(tmp_path: Path, display: str, scope: str):
     runtime, registry, state = tmp_path / "runtime", tmp_path / "registry.json", tmp_path / "state.md"
@@ -199,16 +199,15 @@ def test_public_quota_keeps_gate_from_real_canonical_provider(tmp_path: Path, di
     write_fixture_registry(project=tmp_path, runtime_root=runtime, registry_path=registry,
         goal_id="goal-scope", domain="quota-scope", adapter_kind="generic_project_goal_v0",
         state_file=str(state), registered_agents=["agent-a", "agent-b"], quota_allowed_slots=None)
-    if display != "legacy":
-        goal = json.loads(registry.read_text())["goals"][0]
-        fields = parse_active_state_todos(state.read_text(), goal=goal, item_limit=None)
-        projection = build_todo_runtime_shadow_projection(goal_id="goal-scope",
-            todos=fields["agent_todos"]["items"] + fields["user_todos"]["items"], handoff_mode="soft_claim")
-        initialize_canonical_authority(runtime, "goal-scope", projection, state_path=state)
-        if display == "empty":
-            empty_todo_display(state)
-        else:
-            state.write_text("# Stale projection\n## User Todo\n- [x] Old approval.\n")
+    goal = json.loads(registry.read_text())["goals"][0]
+    fields = parse_active_state_todos(state.read_text(), goal=goal, item_limit=None)
+    projection = build_todo_runtime_shadow_projection(goal_id="goal-scope",
+        todos=fields["agent_todos"]["items"] + fields["user_todos"]["items"], handoff_mode="soft_claim")
+    initialize_canonical_authority(runtime, "goal-scope", projection, state_path=state)
+    if display == "empty":
+        empty_todo_display(state)
+    elif display == "stale":
+        state.write_text("# Stale projection\n## User Todo\n- [x] Old approval.\n")
     before = state.read_bytes() if state.exists() else None
     code, packet = run_json_cli_result("quota", "should-run", "--goal-id", "goal-scope",
         "--agent-id", "agent-b",  registry_path=registry, runtime_root=runtime)
diff --git a/tests/control_plane/test_quota_settlement_cli.py b/tests/control_plane/test_quota_settlement_cli.py
index 9a5eacf70c..62ab69d138 100644
--- a/tests/control_plane/test_quota_settlement_cli.py
+++ b/tests/control_plane/test_quota_settlement_cli.py
@@ -525,7 +525,7 @@ def _append_due_monitors(
     )
 
 
-def _append_blocking_user_gate(project: Path) -> None:
+def _append_blocking_user_gate(project: Path, *, global_gate: bool = False) -> 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
@@ -537,6 +537,7 @@ def _append_blocking_user_gate(project: Path) -> None:
         + "  <!-- loopx:todo todo_id=todo_fixture_user_gate status=open "
         + "task_class=user_gate action_kind=approve_delivery "
         + f"blocks_agent={AGENT_ID} "
+        + ("global_gate=true " if global_gate else "")
         + "decision_scope=direction:action:fixture priority=P0-USER -->\n"
     )
     additions = parse_active_state_todos(
diff --git a/tests/control_plane/test_recovery_due_successor_identity.py b/tests/control_plane/test_recovery_due_successor_identity.py
index 7a60ffd484..bb32cc76f7 100644
--- a/tests/control_plane/test_recovery_due_successor_identity.py
+++ b/tests/control_plane/test_recovery_due_successor_identity.py
@@ -8,7 +8,7 @@
 import pytest
 
 from canonical_authority_fixture import (
-    initialize_canonical_authority,
+    advance_canonical_fixture_authority, initialize_canonical_source_fixture, read_fixture_todos,
     isolate_sqlite_runtime,
 )
 from loopx.cli import main as cli_main
@@ -69,6 +69,7 @@ def cli(*args: str) -> tuple[int, dict]:
             )
             return rc, json.loads(capsys.readouterr().out)
 
+    initialize_canonical_source_fixture(runtime, GOAL_ID, state_path=state, provider=provider)
     rc, vision = cli(
         "refresh-state",
         "--goal-id",
@@ -101,9 +102,8 @@ def cli(*args: str) -> tuple[int, dict]:
             )
         )
 
-    rc, listed = cli("todo", "list", "--goal-id", GOAL_ID, "--limit", "200")
-    assert rc == 0, listed
-    initialize_canonical_authority(
+    listed = {"todos": read_fixture_todos(state)}
+    advance_canonical_fixture_authority(
         runtime,
         GOAL_ID,
         build_todo_runtime_shadow_projection(
@@ -121,8 +121,7 @@ def cli(*args: str) -> tuple[int, dict]:
             handoff_mode="hard_lease",
             leases=[],
         ),
-        state_path=state,
-        provider=provider,
+        operation_id="fixture-retained-frontier",
     )
 
     old_turn, new_turn = "interrupted-original", "current-after-interruption"
diff --git a/tests/control_plane/test_refresh_external_delivery.py b/tests/control_plane/test_refresh_external_delivery.py
index 4719c38550..691253a783 100644
--- a/tests/control_plane/test_refresh_external_delivery.py
+++ b/tests/control_plane/test_refresh_external_delivery.py
@@ -8,15 +8,17 @@
 from loopx.control_plane.quota import refresh_external_delivery as bridge
 from loopx.control_plane.quota.settlement import SettlementIdentity
 from loopx.state_refresh import refresh_state_run
+from tests.control_plane.canonical_authority_fixture import isolate_sqlite_runtime
 from tests.control_plane.test_quota_settlement_cli import (
     AGENT_ID, GOAL_ID, TODO_ID, TURN_ID, _write_fixture, _initialize_fixture_authority,
 )
 
 
 @pytest.fixture
-def session(tmp_path, monkeypatch, capsys):
+def session(tmp_path, monkeypatch, capsys, request):
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
     project, runtime, registry = _write_fixture(tmp_path)
-    _initialize_fixture_authority(project, runtime)
+    _initialize_fixture_authority(project, runtime, provider=getattr(request, "param", "file"))
     monkeypatch.setenv("LOOPX_RUNTIME_ROOT", str(tmp_path / "shared"))
     monkeypatch.chdir(project)
     prefix = ["--registry", str(registry), "--runtime-root", str(runtime)]
diff --git a/tests/control_plane/test_refresh_state_replan_gate.py b/tests/control_plane/test_refresh_state_replan_gate.py
index 66890149d6..0f586b679a 100644
--- a/tests/control_plane/test_refresh_state_replan_gate.py
+++ b/tests/control_plane/test_refresh_state_replan_gate.py
@@ -8,6 +8,7 @@
 from pathlib import Path
 
 import pytest
+from canonical_authority_fixture import initialize_canonical_source_fixture
 
 from loopx.cli import main as cli_main
 from loopx.control_plane.status.autonomous_replan_projection import (
@@ -353,6 +354,7 @@ def _replan_gate_tmp_fixture(
         runtime_root / "goals" / GOAL_ID / "runs" / "index.jsonl"
     )
     _write_typed_repeat_runs_index(runs_index)
+    initialize_canonical_source_fixture(runtime_root, GOAL_ID, state_path=state, provider="sqlite")
     return registry_path, project, runtime_root, runs_index
 
 
@@ -452,6 +454,7 @@ def test_legacy_material_run_cadence_rejects_maintenance_writeback(tmp_path: Pat
         for run in _durable_runs(AUTONOMOUS_REPLAN_PERIODIC_RUN_THRESHOLD):
             handle.write(json.dumps(run) + "\n")
 
+    initialize_canonical_source_fixture(runtime_root, GOAL_ID, state_path=state, provider="sqlite")
     with pytest.raises(ValueError) as exc:
         refresh_state_run(
             registry_path=registry_path,
@@ -690,6 +693,7 @@ def test_refresh_state_run_rejects_maintenance_after_vision_obligation_rotation(
         for run in reversed(_rotated_vision_runs()):
             handle.write(json.dumps(run) + "\n")
 
+    initialize_canonical_source_fixture(runtime_root, GOAL_ID, state_path=state, provider="sqlite")
     before = state.read_text(encoding="utf-8")
     with pytest.raises(ValueError, match="required vision outcome"):
         refresh_state_run(
diff --git a/tests/control_plane/test_refresh_todo_projection.py b/tests/control_plane/test_refresh_todo_projection.py
index 578ba33af9..3fe1b72cf9 100644
--- a/tests/control_plane/test_refresh_todo_projection.py
+++ b/tests/control_plane/test_refresh_todo_projection.py
@@ -3,10 +3,6 @@
 import json
 
 import pytest
-from canonical_authority_fixture import (
-    initialize_canonical_authority,
-    isolate_sqlite_runtime,
-)
 from tests.control_plane import test_refresh_external_delivery as refresh_fixtures
 from loopx.control_plane.coordination.local_authority import (
     read_canonical_todos_if_promoted,
@@ -162,29 +158,20 @@ def unexpected(**kw):
     assert state.read_bytes() == before
 
 
-@pytest.mark.parametrize("provider", ["file", "sqlite"])
+@pytest.mark.parametrize("session", ["file", "sqlite"], indirect=True)
 def test_cli_same_turn_replay_recovers_projection_without_duplicate_writeback(
-    session, monkeypatch, tmp_path, provider
+    session, monkeypatch
 ):
-    from loopx.control_plane.coordination.runtime_shadow import (
-        build_runtime_shadow_source_snapshot,
-    )
     from loopx.control_plane.todos import provider_projection
     from tests.control_plane.test_quota_settlement_cli import GOAL_ID
 
     project, runtime, registry, args, run, journal, index = session
-    isolate_sqlite_runtime(tmp_path, monkeypatch)
     initial = run(args)
     assert initial["appended"]
     data = json.loads(registry.read_text())
     goal = data["goals"][0]
     state = project / goal["state_file"]
-    projection, _ = build_runtime_shadow_source_snapshot(
-        goal=goal, runtime_root=runtime, state_path=state, registry_path=registry
-    )
-    seeded = initialize_canonical_authority(
-        runtime, GOAL_ID, projection, state_path=state, provider=provider
-    )
+    seeded = read_canonical_todos_if_promoted(runtime_root=runtime, goal_id=GOAL_ID)
     before = index.read_bytes(), journal.read_bytes()
     original = state.read_text()
     state.write_text("# Existing narrative\n\n## Agent Todo\n")
@@ -202,7 +189,7 @@ def fail(*a, **kw):
     assert recovered["idempotent_replay"] and not recovered["appended"]
     assert recovered["projection_delivery"] == "delivered"
     assert "Existing narrative" in state.read_text()
-    assert projection["todos"][0]["todo_id"] in state.read_text()
+    assert seeded["todos"][0]["todo_id"] in state.read_text()
     stable = run(args)
     assert stable["projection_delivery"] == "current"
     assert (index.read_bytes(), journal.read_bytes()) == before
diff --git a/tests/control_plane/test_registry_relative_runtime_resolution.py b/tests/control_plane/test_registry_relative_runtime_resolution.py
index 01408b9668..e4aa54d1f1 100644
--- a/tests/control_plane/test_registry_relative_runtime_resolution.py
+++ b/tests/control_plane/test_registry_relative_runtime_resolution.py
@@ -8,7 +8,7 @@
 
 import pytest
 
-from canonical_authority_fixture import initialize_canonical_authority
+from canonical_authority_fixture import advance_canonical_fixture_authority
 from examples.control_plane.quota_plan_fixtures import write_cli_fixture
 from loopx.capabilities.explore.result_log import (
     append_explore_result_event,
@@ -154,11 +154,8 @@ def test_todo_list_anchors_promoted_authority_to_registry_project(
             }
         ],
     )
-    initialize_canonical_authority(
-        target_runtime,
-        "half-speed",
-        projection,
-        state_path=state_path,
+    advance_canonical_fixture_authority(
+        target_runtime, "half-speed", projection, operation_id="canonical-relative-runtime",
     )
     independent_worktree = tmp_path / "independent-worktree"
     independent_worktree.mkdir()
diff --git a/tests/control_plane/test_replan_history_provider.py b/tests/control_plane/test_replan_history_provider.py
index 60ecaf4892..1e98f9e9d0 100644
--- a/tests/control_plane/test_replan_history_provider.py
+++ b/tests/control_plane/test_replan_history_provider.py
@@ -18,7 +18,7 @@
 @pytest.mark.parametrize("provider", ["file", "sqlite"])
 def test_complex_provider_snapshot_replan_and_quota_readback(tmp_path, monkeypatch, provider):
     isolate_sqlite_runtime(tmp_path, monkeypatch)
-    paths = _write_fixture(tmp_path, runs=[])
+    paths = _write_fixture(tmp_path, runs=[], initialize=False)
     projection = _fixture()
     initialize_canonical_authority(paths["runtime"], GOAL_ID, projection,
                                    state_path=paths["state_file"], provider=provider)
@@ -51,7 +51,7 @@ def test_complex_provider_snapshot_replan_and_quota_readback(tmp_path, monkeypat
 @pytest.mark.parametrize("provider", ["file", "sqlite"])
 def test_quota_shared_evidence_command_reads_exact_history_and_preserves_scope(tmp_path, monkeypatch, provider):
     isolate_sqlite_runtime(tmp_path, monkeypatch)
-    paths = _write_fixture(tmp_path, runs=[])
+    paths = _write_fixture(tmp_path, runs=[], initialize=False)
     initialize_canonical_authority(paths["runtime"], GOAL_ID, _fixture(),
                                    state_path=paths["state_file"], provider=provider)
     history = [{**row, "health_check": f"Observation {i}\nFull tail constraint."}
@@ -87,7 +87,7 @@ def test_explicit_completed_todo_fallback_retains_twenty_record_threshold(
     tmp_path, monkeypatch, provider,
 ):
     isolate_sqlite_runtime(tmp_path, monkeypatch)
-    paths = _write_fixture(tmp_path, runs=[])
+    paths = _write_fixture(tmp_path, runs=[], initialize=False)
     registry = json.loads(paths["registry"].read_text())
     registry["goals"][0]["adapter"] = {"kind": "harness_self_improvement"}
     paths["registry"].write_text(json.dumps(registry))
diff --git a/tests/control_plane/test_replan_history_transport.py b/tests/control_plane/test_replan_history_transport.py
index fbe17accd3..d6ecd90831 100644
--- a/tests/control_plane/test_replan_history_transport.py
+++ b/tests/control_plane/test_replan_history_transport.py
@@ -7,7 +7,7 @@
 
 
 def test_large_history_refresh_replay_and_spend_once(tmp_path):
-    project, runtime, registry = _write_fixture(tmp_path)
+    project, runtime, registry = _write_fixture(tmp_path, provider="sqlite")
     index = runtime / "goals" / GOAL_ID / "runs" / "index.jsonl"
     code, baseline = _run_cli(registry, runtime, "refresh-state", "--goal-id", GOAL_ID,
                               "--agent-id", AGENT_ID, "--no-global-sync",
diff --git a/tests/control_plane/test_replan_staged_read.py b/tests/control_plane/test_replan_staged_read.py
index d6c5cc2078..780a14138c 100644
--- a/tests/control_plane/test_replan_staged_read.py
+++ b/tests/control_plane/test_replan_staged_read.py
@@ -129,7 +129,7 @@ def test_pure_replan_does_not_advertise_unavailable_explore_attachment(tmp_path,
         # not suggest a new way around it or change its primary commit behavior.
         attachment = {**original["result_attachment_template"], "question": "Is the source valid?",
             "applicability": "Synthetic source", "input_revision": "source-v1", "observation": "Reviewed",
-            "interpretation": "Unresolved", "evidence_refs": ["evidence:source-review"]}
+            "interpretation": "Unresolved", "status": "tentative", "evidence_refs": ["evidence:source-review"]}
         with pytest.raises(ValueError, match="requires agent, Todo and Turn identity"):
             prepare_result_attachment(attachment, registry_path=tmp_path / "registry.json",
                 runtime_root=runtime, goal_id=goal, agent_id=agent, todo_id=None, turn_instance_id=turn)
diff --git a/tests/control_plane/test_replan_terminal_spend_ordering.py b/tests/control_plane/test_replan_terminal_spend_ordering.py
index 5d1631e256..fe1cdea212 100644
--- a/tests/control_plane/test_replan_terminal_spend_ordering.py
+++ b/tests/control_plane/test_replan_terminal_spend_ordering.py
@@ -17,7 +17,7 @@
 from tests.control_plane.test_quota_settlement_cli import (
     AGENT_ID,
     GOAL_ID,
-    _configure_autonomous_replan_fixture,
+    _configure_autonomous_replan_fixture, _initialize_fixture_authority,
     _projected_cli_args,
     _run_cli,
     _spend_run_count,
@@ -162,6 +162,7 @@ def test_coverage_backed_no_followup_replan_does_not_strand_its_quota_spend(
     project, runtime, registry_path = _write_fixture(tmp_path)
     _configure_autonomous_replan_fixture(project, runtime, registry_path)
     _write_terminal_frontier_state(project)
+    _initialize_fixture_authority(project, runtime, provider="sqlite")
     vision_path = _write_no_followup_vision(tmp_path)
 
     guard = _should_run(registry_path, runtime, project)
@@ -242,6 +243,7 @@ def test_unbound_terminal_spend_stays_rejected(tmp_path: Path) -> None:
     project, runtime, registry_path = _write_fixture(tmp_path)
     _configure_autonomous_replan_fixture(project, runtime, registry_path)
     _write_terminal_frontier_state(project)
+    _initialize_fixture_authority(project, runtime, provider="sqlite")
     vision_path = _write_no_followup_vision(tmp_path)
 
     guard = _should_run(registry_path, runtime, project)
diff --git a/tests/control_plane/test_resume_planning_projection.py b/tests/control_plane/test_resume_planning_projection.py
index 4f5e155f64..077fc15499 100644
--- a/tests/control_plane/test_resume_planning_projection.py
+++ b/tests/control_plane/test_resume_planning_projection.py
@@ -140,6 +140,12 @@ def test_public_quota_distinguishes_monitor_repair_from_generation_wait(
     assert (b"## Agent Todo" in before) == (provider is None)
     code, packet = run_json_cli_result("quota", "should-run", "--goal-id", "goal-a", "--agent-id", "agent-a",
         "--include-detail", "agent-todos",  registry_path=registry, runtime_root=runtime)
+    if provider is None:
+        assert code == 1 and packet["error_code"] == "canonical_authority_required", packet
+        assert packet["should_run"] is False
+        assert state.read_bytes() == before
+        assert not (runtime / "goals/goal-a/runs").exists()
+        return
     assert code == 0, packet
     summary = packet["agent_todo_summary"]
     condition = summary["resume_blocked_items"][0]["resume_condition"]
@@ -198,6 +204,12 @@ def test_public_quota_replans_completion_cycles_without_reassigning_work(
     before = state.read_bytes()
     code, packet = run_json_cli_result("quota", "should-run", "--goal-id", "goal-a", "--agent-id", "agent-a",
         "--include-detail", "agent-todos",  registry_path=registry, runtime_root=runtime)
+    if provider == "legacy":
+        assert code == 1 and packet["error_code"] == "canonical_authority_required", packet
+        assert packet["should_run"] is False
+        assert state.read_bytes() == before
+        assert not (runtime / "goals/goal-a/runs").exists()
+        return
     assert code == 0, packet
     summary = packet["agent_todo_summary"]
     blocked = next(row for row in summary["resume_blocked_items"] if row["todo_id"] == "todo_alpha")
@@ -271,6 +283,11 @@ def test_public_quota_capacity_wait_uses_provider_after_cutover_without_display_
             "quota", "should-run", "--goal-id", "goal-a", "--agent-id", "agent-a",
              *capabilities, registry_path=registry, runtime_root=runtime,
         )
+        if provider is None:
+            assert code == 1 and packet["error_code"] == "canonical_authority_required", packet
+            assert packet["should_run"] is False
+            assert state.read_bytes() == before
+            continue
         assert code == 0, packet
         assert packet["agent_todo_summary"]["current_agent_deferred_resume_count"] == count
         if count:
diff --git a/tests/control_plane/test_retired_todo_event_source.py b/tests/control_plane/test_retired_todo_event_source.py
index 3cf34c9a38..684d09bdb0 100644
--- a/tests/control_plane/test_retired_todo_event_source.py
+++ b/tests/control_plane/test_retired_todo_event_source.py
@@ -11,7 +11,7 @@
 import pytest
 
 from loopx.control_plane.goals.legacy_event_source import (
-    require_no_legacy_todo_events,
+    RetiredTodoEventSourceError, require_no_legacy_todo_events,
 )
 from loopx.control_plane.coordination.local_authority import (
     CanonicalAuthorityRequired,
@@ -22,7 +22,7 @@
 from loopx.control_plane.todos.todo_index import build_todo_index
 from loopx.control_plane.runtime.public_safety import public_safe_compact_text
 from tests.control_plane.canonical_authority_fixture import (
-    initialize_canonical_authority, isolate_sqlite_runtime,
+    initialize_canonical_authority, initialize_canonical_source_fixture, isolate_sqlite_runtime,
 )
 from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection
 from loopx.control_plane.effect_runtime import restart_effect_runtime
@@ -62,13 +62,15 @@ def test_all_old_source_selectors_refuse_reads_and_writes_without_data_loss(tmp_
     registry = tmp_path / "registry.json"
     registry.write_text(json.dumps({"common_runtime_root": str(tmp_path / "runtime"), "goals": [goal]}))
     before = state.read_bytes(), old.read_bytes()
+    with pytest.raises(RetiredTodoEventSourceError, match="legacy_todo_event_source_retired"):
+        require_no_legacy_todo_events(goal, state_path=state)
     for args in [
         ("todo", "list"),
         ("shared-goal-alignment", "--agent-id", "agent-a", "--project", str(tmp_path)),
     ]:
         code, result = run_json_cli_result(*args, "--goal-id", goal["id"], registry_path=registry)
         assert code != 0 or result.get("ok") is False, result
-        assert "legacy_todo_event_source_retired" in json.dumps(result), result
+        assert "canonical_authority_required" in json.dumps(result), result
         assert (state.read_bytes(), old.read_bytes()) == before
     for args in [
         ("todo", "add", "--text", "New work", "--role", "agent"),
@@ -131,12 +133,18 @@ def _portfolio_fixture(root: Path, alias: str | None = None):
     registry = root / "registry.json"
     registry.write_text(json.dumps({"schema_version": 1, "common_runtime_root": str(runtime), "goals": goals}))
     old = root / "retired" / ("aliased.jsonl" if alias else "events.jsonl")
+    initialize_canonical_source_fixture(runtime, "healthy",
+        state_path=Path(goals[1]["state_file"]), provider="sqlite")
     return registry, runtime, goals, old
 
 
 def _assert_portfolio_refusal(payload):
-    rows = {item["goal_id"]: item for item in payload["attention_queue"]["items"]}
-    assert rows["retired"]["status"] == "legacy_todo_event_source_retired"
+    items = payload["attention_queue"]["items"]
+    rows = {
+        "retired": next(item for item in items if item["goal_id"] == "retired" and item.get("todo_source") == "unavailable"),
+        "healthy": next(item for item in items if item["goal_id"] == "healthy" and "agent_todos" in item),
+    }
+    assert rows["retired"]["status"] == "canonical_authority_required"
     assert rows["retired"]["todo_source"] == "unavailable"
     assert rows["retired"]["waiting_on"] == "user_or_controller"
     assert "agent_todos" not in rows["retired"]
@@ -162,8 +170,10 @@ def test_retired_source_is_scoped_and_later_healthy_goal_remains_visible(tmp_pat
     data["goals"] = goals[:1]
     registry.write_text(json.dumps(data))
     code, first = run_json_cli_result("status", registry_path=registry)
-    assert code == 0, first
-    assert first["attention_queue"]["items"][0]["status"] == "legacy_todo_event_source_retired"
+    assert code == 1 and first["ok"] is False, first
+    refused = next(item for item in first["attention_queue"]["items"]
+        if item["goal_id"] == "retired" and item.get("todo_source") == "unavailable")
+    assert refused["status"] == "canonical_authority_required"
     data["goals"] = goals
     registry.write_text(json.dumps(data))
     for rewrite_annotation in (False, True):
@@ -171,7 +181,7 @@ def test_retired_source_is_scoped_and_later_healthy_goal_remains_visible(tmp_pat
             state = Path(goals[0]["state_file"])
             state.write_text(state.read_text().replace("task_class=advancement_task", "task_class=maintenance_task"))
         code, payload = run_json_cli_result("status", registry_path=registry)
-        assert code == 0, payload
+        assert code == 1 and payload["ok"] is False, payload
         _assert_portfolio_refusal(payload)
         assert old.read_bytes() == before
     code, mutation = run_json_cli_result("todo", "add", "--goal-id", "retired",
@@ -181,7 +191,12 @@ def test_retired_source_is_scoped_and_later_healthy_goal_remains_visible(tmp_pat
     assert mutation["write_admitted"] is False
     assert mutation["legacy_fallback_used"] is False
     assert old.read_bytes() == before
-    old.unlink()  # Explicit owner recovery in a disposable synthetic fixture.
+    old.unlink()
+    code, still_unmigrated = run_json_cli_result("status", registry_path=registry)
+    assert code == 1 and still_unmigrated["ok"] is False
+    _assert_portfolio_refusal(still_unmigrated)
+    initialize_canonical_source_fixture(runtime, "retired",
+        state_path=Path(goals[0]["state_file"]), provider="sqlite")
     code, recovered = run_json_cli_result("status", registry_path=registry)
     assert code == 0, recovered
     assert {item["goal_id"] for item in recovered["todo_index"]["items"]} == {"healthy", "retired"}
@@ -215,11 +230,9 @@ def test_http_status_preserves_healthy_goal_when_retired_source_appears(tmp_path
             with urllib.request.urlopen(base + "/status", timeout=90) as response:
                 assert response.status == 200
                 payload = json.load(response)
+            _assert_portfolio_refusal(payload)
             if present:
-                _assert_portfolio_refusal(payload)
                 assert old.read_bytes() == b"old source preserved\n"
-            else:
-                assert {item["goal_id"] for item in payload["todo_index"]["items"]} == {"healthy", "retired"}
     finally:
         server.terminate()
         server.wait(timeout=10)
diff --git a/tests/control_plane/test_scheduler_arbitration_transport.py b/tests/control_plane/test_scheduler_arbitration_transport.py
index 19e74697d3..a44b887f13 100644
--- a/tests/control_plane/test_scheduler_arbitration_transport.py
+++ b/tests/control_plane/test_scheduler_arbitration_transport.py
@@ -107,10 +107,14 @@ def observe(method, params, **kwargs):
     assert observed == [(params_bytes, params_bytes > 65536)]
     assert bool(snapshot_directories) is (params_bytes > 65536)
     assert all(not directory.exists() for directory in snapshot_directories)
-    assert reads == [read, {
+    assert len(reads) == 4 and reads[0] == read
+    required = reads[1]
+    assert {key: value for key, value in required.items() if key != "reason"} == {
         "command": "cat -- " + shlex.quote(state_file), "source": "goal_state", "ordering": "before_work",
-        "reason": "Read complete Goal intent, acceptance and stops. Exact Todo reads own task state/claims. Source failure or changed requirements require a fresh guard; summaries cannot replace this read. Only when explicit user direction invalidates the objective, use goal-acceptance prepare-objective; reuse pending proposals. Progress, diagnostics or age need no revision; documents/peers grant no authority.",
-    }, {
+    }
+    for obligation in ("full Goal intent", "acceptance", "stops", "No summaries", "fresh guard"):
+        assert obligation in required["reason"]
+    assert reads[2:] == [{
         "command": "loopx --format json goal-acceptance inspect --goal-id " + shlex.quote(goal_id),
         "source": "goal_acceptance", "ordering": "before_work",
         "reason": "Read full configured objective, non-goals and criteria with scope/revision. Scoped acceptance does not replace Goal intent or prove completion; Removed or changed checks require a fresh guard.",
@@ -119,6 +123,7 @@ def observe(method, params, **kwargs):
         "source": "selected_todo", "ordering": "before_work",
         "reason": "Read full current requirements and status/claim. Require one matching active Todo; missing, ambiguous or changed work requires a fresh guard. A summary cannot replace this read.",
     }]
+
     assert arbitration == SchedulerArbitration(SchedulerDisposition.ACTIVE_WORK,
         "interaction_agent_attempt_required", "bounded_delivery")
     assert_private_files_removed(Path(tempfile.gettempdir()))
diff --git a/tests/control_plane/test_scoped_gate_fallback_cli.py b/tests/control_plane/test_scoped_gate_fallback_cli.py
index cc6758680c..d3bac992e0 100644
--- a/tests/control_plane/test_scoped_gate_fallback_cli.py
+++ b/tests/control_plane/test_scoped_gate_fallback_cli.py
@@ -8,7 +8,7 @@
 from loopx.control_plane.testing.canary_harness import write_fixture_registry, run_json_cli_result
 
 
-@pytest.mark.parametrize("provider", ["markdown", "file", "sqlite"])
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
 @pytest.mark.parametrize("explicit_target", [False, True, None])
 def test_quota_fallback_uses_real_source_and_explicit_relations(tmp_path, monkeypatch, provider, explicit_target):
     if provider == "sqlite":
diff --git a/tests/control_plane/test_selection_replan_reentry.py b/tests/control_plane/test_selection_replan_reentry.py
index 04454d5905..7800819a3f 100644
--- a/tests/control_plane/test_selection_replan_reentry.py
+++ b/tests/control_plane/test_selection_replan_reentry.py
@@ -3,16 +3,27 @@
 import shlex
 
 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 test_quota_settlement_cli import (
     AGENT_ID, GOAL_ID, SELECTED_REPLAN_TODO_ID, TODO_ID,
     _configure_autonomous_replan_fixture, _configure_selected_todo_replan_fixture,
     _configure_selectable_alternative, _heartbeat_receipt_count,
-    _initialize_git_checkout, _projected_cli_args, _run_cli, _run_generated_cli,
+    _initialize_fixture_authority, _initialize_git_checkout, _projected_cli_args, _run_cli, _run_generated_cli,
     _spend_run_count, _write_fixture,
 )
 
 
+def _advance_frontier(project, runtime, operation_id):
+    state = project / f".codex/goals/{GOAL_ID}/ACTIVE_GOAL_STATE.md"
+    advance_canonical_fixture_authority(
+        runtime, GOAL_ID, build_todo_runtime_shadow_projection(
+            goal_id=GOAL_ID, todos=read_fixture_todos(state), handoff_mode="soft_claim",
+        ), operation_id=operation_id,
+    )
+
+
 @pytest.mark.parametrize(("binding", "initial_replan"), [
     ("todo", False), ("autonomous_replan", False), ("todo", True),
 ])
@@ -21,6 +32,7 @@ def test_replan_selection_recovers_inline_or_refuses_ineligible_choice_and_settl
     _configure_selectable_alternative(project)
     if initial_replan:
         _configure_selected_todo_replan_fixture(project, registry)
+    _initialize_fixture_authority(project, runtime, provider="sqlite")
     turn = "turn-selection-preempted"
     guard = ("quota", "should-run", "--codex-app", "--goal-id", GOAL_ID,
              "--agent-id", AGENT_ID, "--turn-instance-id", turn, )
@@ -42,6 +54,7 @@ def test_replan_selection_recovers_inline_or_refuses_ineligible_choice_and_settl
     else:
         _configure_autonomous_replan_fixture(project, runtime, registry)
         selected_id = TODO_ID
+    _advance_frontier(project, runtime, "fixture-frontier-after-selection")
     rc, selected = _run_cli(registry, runtime, *guard, "--todo-id", selected_id)
     selection_deferred = binding == "todo"
     expected_before_resume = 2 if selection_deferred else 1
@@ -170,6 +183,7 @@ def test_reentry_never_replaces_retained_selection_with_recommended_todo(tmp_pat
     _configure_selectable_alternative(project)
     if initial_replan:
         _configure_selected_todo_replan_fixture(project, registry)
+    _initialize_fixture_authority(project, runtime, provider="sqlite")
     turn = "turn-selection-recommendation-drift"
     guard = ("quota", "should-run", "--codex-app", "--goal-id", GOAL_ID,
              "--agent-id", AGENT_ID, "--turn-instance-id", turn, )
@@ -187,6 +201,7 @@ def test_reentry_never_replaces_retained_selection_with_recommended_todo(tmp_pat
     # The original explicit choice leaves the refreshed frontier while a
     # different recommended Todo becomes visible under a hard replan.
     _configure_selected_todo_replan_fixture(project, registry)
+    _advance_frontier(project, runtime, "fixture-recommendation-drift")
     retained_todo_id = "todo_chain_000000000001"
     rc, resumed = _run_cli(
         registry, runtime, *guard, "--todo-id", retained_todo_id
@@ -278,6 +293,7 @@ def test_replan_preempts_unselected_todo_guards_on_selection_reentry(tmp_path, w
     )
     registry_data["goals"][0]["coordination"]["write_scope"] = write_scope
     registry.write_text(json.dumps(registry_data) + "\n", encoding="utf-8")
+    _initialize_fixture_authority(project, runtime, provider="sqlite")
     _initialize_git_checkout(project)
 
     turn = "turn-selection-replan-workspace-guard"
@@ -329,6 +345,7 @@ def test_inline_reentry_reads_gates_changed_after_choice_receipt(
 
     project, runtime, registry = _write_fixture(tmp_path)
     _configure_selected_todo_replan_fixture(project, registry)
+    _initialize_fixture_authority(project, runtime, provider="sqlite")
     turn = "turn-inline-reentry-new-gate"
     guard = ("quota", "should-run", "--codex-app", "--goal-id", GOAL_ID,
              "--agent-id", AGENT_ID, "--turn-instance-id", turn,
@@ -344,7 +361,8 @@ def reconcile_then_change_gate(*args, **kwargs):
         if len(calls) == 1:
             assert result.rejected and result.receipt_status == "selection_retained"
             if changed_gate == "user_gate":
-                _append_blocking_user_gate(project)
+                _append_blocking_user_gate(project, global_gate=True)
+                _advance_frontier(project, runtime, "fixture-new-user-gate")
             else:
                 data = json.loads(registry.read_text())
                 data["goals"][0]["quota"]["compute"] = 0
@@ -386,6 +404,7 @@ def reconcile_then_change_gate(*args, **kwargs):
 def test_first_inline_result_has_matching_turn_envelope(tmp_path):
     project, runtime, registry = _write_fixture(tmp_path)
     _configure_selected_todo_replan_fixture(project, registry)
+    _initialize_fixture_authority(project, runtime, provider="sqlite")
     turn = "turn-inline-envelope"
     guard = ("quota", "should-run", "--codex-app", "--goal-id", GOAL_ID,
              "--agent-id", AGENT_ID, "--turn-instance-id", turn,
diff --git a/tests/control_plane/test_settlement_input_discovery_cli.py b/tests/control_plane/test_settlement_input_discovery_cli.py
index fd6cb90a49..95c1990d36 100644
--- a/tests/control_plane/test_settlement_input_discovery_cli.py
+++ b/tests/control_plane/test_settlement_input_discovery_cli.py
@@ -282,7 +282,9 @@ def run(*args):
     # Reading this view neither enables stride nor settles another Turn.
     ordinary = project_replan_context(
         goal_id=GOAL_ID, agent_id=AGENT_ID, runs=(), source_status=status, stride_strategy="off")
-    assert "decision_followthrough" not in ordinary
+    assert ordinary["decision_followthrough"] == follow
+    assert "stride_strategy" not in ordinary
+    assert "explore_basis" not in ordinary
     assert _spend_run_count(runtime) == 3
 
     # Preserve compatibility without turning a descriptive or unsupported
diff --git a/tests/control_plane/test_source_lease_reader_isolation.py b/tests/control_plane/test_source_lease_reader_isolation.py
index 32514ab51b..9d7de829fd 100644
--- a/tests/control_plane/test_source_lease_reader_isolation.py
+++ b/tests/control_plane/test_source_lease_reader_isolation.py
@@ -97,8 +97,19 @@ def lease(action, *args, success=True):
         "updated_at": "2026-09-01T00:00:00Z",
         "expires_at": "2099-01-01T00:00:00Z",
     }
+    def retained_source_record():
+        return json.loads(subprocess.check_output(
+            [sys.executable, "-S", "-c",
+             "import json,sys;from pathlib import Path;"
+             "from loopx.control_plane.work_items.local_lease_record import read_lease;"
+             "print(json.dumps(read_lease(Path(sys.argv[1]))))", str(lease_path)],
+            cwd=tmp_path, text=True, timeout=30,
+        ))
+
     lease_path.write_text(json.dumps(historical))
-    assert lease("inspect")["lease"] == historical
+    refused = lease("inspect", success=False)
+    assert refused["error_code"] == "canonical_authority_required"
+    assert retained_source_record() == historical
 
     renew = lease(
         "renew",
@@ -126,7 +137,9 @@ def lease(action, *args, success=True):
         success=False,
     )
     assert transfer["error_code"] == "canonical_authority_required"
-    assert lease("inspect")["lease"] == historical
+    refused = lease("inspect", success=False)
+    assert refused["error_code"] == "canonical_authority_required"
+    assert retained_source_record() == historical
 
     released = lease(
         "release",
@@ -137,8 +150,9 @@ def lease(action, *args, success=True):
         "1",
     )
     assert released["lease"]["status"] == "released"
-    retained = lease("inspect")
-    assert not retained["active"] and retained["lease"] == released["lease"]
+    retained = retained_source_record()
+    assert retained == released["lease"] and retained["status"] == "released"
+    assert lease("inspect", success=False)["error_code"] == "canonical_authority_required"
     assert (w.state.read_bytes(), w.registry.read_bytes()) == original
     if capture:
         # Read the retained journal after the lifecycle; no capture remains pending.
diff --git a/tests/control_plane/test_standing_decision_order.py b/tests/control_plane/test_standing_decision_order.py
index 5b505ba172..f54c939212 100644
--- a/tests/control_plane/test_standing_decision_order.py
+++ b/tests/control_plane/test_standing_decision_order.py
@@ -210,6 +210,11 @@ def test_public_quota_reads_revocation_from_full_history_without_mutating_displa
         registry_path=registry,
         runtime_root=runtime,
     )
+    if mode == "markdown":
+        assert code == 1 and packet["error_code"] == "canonical_authority_required", packet
+        assert packet["should_run"] is False
+        assert state.read_bytes() == before
+        return
     assert code == 0, packet
     authority = packet["standing_decision_authority"]
     assert authority["active_count"] == 0
diff --git a/tests/control_plane/test_status_rollout_event_snapshot.py b/tests/control_plane/test_status_rollout_event_snapshot.py
index f20b17c6bb..21b62ab319 100644
--- a/tests/control_plane/test_status_rollout_event_snapshot.py
+++ b/tests/control_plane/test_status_rollout_event_snapshot.py
@@ -67,6 +67,8 @@ def _write_fixture(tmp_path: Path) -> tuple[Path, Path, dict[str, Any]]:
         + "\n",
         encoding="utf-8",
     )
+    from canonical_authority_fixture import initialize_canonical_source_fixture
+    initialize_canonical_source_fixture(runtime_root, goal["id"], state_path=state_path, provider="sqlite")
     return registry_path, runtime_root, goal
 
 
@@ -136,7 +138,7 @@ def _indexed_todo_ids(payload: dict[str, Any]) -> set[str]:
     }
 
 
-def test_collect_status_reuses_display_snapshot_but_pr_wait_reads_complete_facts(
+def test_collect_status_freezes_complete_pr_facts_before_bounded_display_read(
     tmp_path: Path,
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
@@ -153,6 +155,10 @@ def test_collect_status_reuses_display_snapshot_but_pr_wait_reads_complete_facts
     )
     for event in initial_events:
         _append_event(runtime_root, event)
+        if event is initial_events[0]:
+            for index in range(600):
+                _append_event(runtime_root, {**_merge_event(900, recorded_at="2026-09-18T01:00:30Z"),
+                    "event_id": f"merge-noise-{index}"})
 
     loads: list[tuple[str, int | None]] = []
     original_load = rollout_event_log_module.load_rollout_events
@@ -180,13 +186,12 @@ def load_and_append(path: Path, *, limit: int | None = None):
 
     assert loads == [(GOAL_ID, 500)]
     assert _deferred_todo(first, "todo_wait_pr_41")["resume_ready"] is True
-    # The retained PR proof is read by the typed Todo owner, independently of
-    # the bounded display snapshot. Evidence receipts and Todo index still
-    # share the original snapshot within this status request.
+    # The typed owner captures complete PR facts before display enrichment.
+    # PR41 is older than the 500-event display tail and still resumes. PR42
+    # arriving after that request snapshot is visible on the next request.
     resumed = _deferred_todo(first, "todo_wait_pr_42")
-    assert resumed["resume_ready"] is True
-    assert resumed["resume_condition"]["matched_event_id"] == appended_events[0]["event_id"]
-    assert resumed["resume_condition"]["matched_pr_ref"] == f"{REPOSITORY}#42"
+    assert resumed["resume_ready"] is False
+    assert "matched_event_id" not in resumed["resume_condition"]
     assert [
         receipt["event_id"]
         for receipt in _goal_queue_item(first)["evidence_log_read_receipts"]
@@ -204,6 +209,7 @@ def load_and_append(path: Path, *, limit: int | None = None):
 
     assert loads == [(GOAL_ID, 500), (GOAL_ID, 500)]
     assert _deferred_todo(second, "todo_wait_pr_42")["resume_ready"] is True
+    assert _deferred_todo(second, "todo_wait_pr_42")["resume_condition"]["matched_event_id"] == appended_events[0]["event_id"]
     assert [
         receipt["event_id"]
         for receipt in _goal_queue_item(second)["evidence_log_read_receipts"]
diff --git a/tests/control_plane/test_succession_provider_readback.py b/tests/control_plane/test_succession_provider_readback.py
index ed4e0e9fe2..b12bbdb2d5 100644
--- a/tests/control_plane/test_succession_provider_readback.py
+++ b/tests/control_plane/test_succession_provider_readback.py
@@ -56,7 +56,7 @@ def selected_summary(registry: Path, runtime: Path, state: Path, todo_id: str, *
     return projected.summaries['agent_todos']
 
 
-@pytest.mark.parametrize('provider', ['legacy', 'file', 'sqlite'])
+@pytest.mark.parametrize('provider', ['file', 'sqlite'])
 def test_real_cli_historical_route_advisory_preserves_old_source_and_provider(tmp_path, monkeypatch, provider):
     isolate_sqlite_runtime(tmp_path, monkeypatch)
     runtime, state, registry = tmp_path / 'runtime', tmp_path / 'state.md', tmp_path / 'registry.json'
@@ -92,7 +92,7 @@ def test_real_cli_historical_route_advisory_preserves_old_source_and_provider(tm
     assert (state.read_bytes() if state.exists() else None) == source
 
 
-@pytest.mark.parametrize('provider', ['legacy', 'file', 'sqlite'])
+@pytest.mark.parametrize('provider', ['file', 'sqlite'])
 def test_real_cli_graph_selection_history_and_read_only_manager(tmp_path, monkeypatch, provider):
     isolate_sqlite_runtime(tmp_path, monkeypatch)
     fixture = fixture_projection()
diff --git a/tests/control_plane/test_task_graph_topology.py b/tests/control_plane/test_task_graph_topology.py
index 9caa57061f..70767a3a9d 100644
--- a/tests/control_plane/test_task_graph_topology.py
+++ b/tests/control_plane/test_task_graph_topology.py
@@ -73,7 +73,7 @@ def test_lineage_and_condition_between_same_pair_are_both_retained():
 
 
 @pytest.mark.parametrize("provider,display", [
-    ("legacy", "current"), ("file", "current"), ("sqlite", "current"),
+    ("file", "current"), ("sqlite", "current"),
     ("file", "missing"), ("sqlite", "missing"),
 ])
 def test_real_status_reader_uses_provider_snapshot_without_repair_writes(tmp_path, monkeypatch, provider, display):
diff --git a/tests/control_plane/test_todo_consumer_scope.py b/tests/control_plane/test_todo_consumer_scope.py
index 3730978a2f..1aaa694a02 100644
--- a/tests/control_plane/test_todo_consumer_scope.py
+++ b/tests/control_plane/test_todo_consumer_scope.py
@@ -41,7 +41,7 @@ def test_list_and_quota_share_addressed_scope(task_class, scope, visible):
     assert filtered_todo_summary(source, role="user")["total_count"] == 1
 
 
-@pytest.mark.parametrize("provider", ["legacy", "file", "sqlite"])
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
 def test_real_cli_scope_survives_missing_display_and_limits(tmp_path: Path, provider: str):
     state, registry, runtime = tmp_path / "STATE.md", tmp_path / "registry.json", tmp_path / "runtime"
     user_rows = [
diff --git a/tests/control_plane/test_todo_exact_detail.py b/tests/control_plane/test_todo_exact_detail.py
index 3280508e8c..df517459c7 100644
--- a/tests/control_plane/test_todo_exact_detail.py
+++ b/tests/control_plane/test_todo_exact_detail.py
@@ -7,7 +7,7 @@
 from pathlib import Path
 
 import pytest
-from canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime
+from canonical_authority_fixture import assert_todo_read_unchanged, 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_todo_source
@@ -40,7 +40,7 @@ def string_leaves(value):
             yield from string_leaves(child)
 
 
-@pytest.mark.parametrize("provider", ["legacy", "file", "sqlite"])
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
 def test_detail_retains_one_complete_body_and_current_source(tmp_path, monkeypatch, provider):
     isolate_sqlite_runtime(tmp_path, monkeypatch)
     project, runtime, registry, state = _write_fixture(tmp_path, SCENARIOS[0])
@@ -75,7 +75,8 @@ def test_detail_retains_one_complete_body_and_current_source(tmp_path, monkeypat
     assert "TAIL:" in markdown and '"source_complete": true' in markdown
     assert state.read_bytes() == before
     rc, repeated = cli(registry, runtime, "--todo-id", TODO, "--agent-id", AGENT_IDS[0])
-    assert rc == 0 and repeated == full
+    assert rc == 0
+    assert_todo_read_unchanged(repeated, full)
     rc, thin = cli(registry, runtime, "--thin")
     assert rc == 0 and "TAIL:" not in json.dumps(thin)
     rc, missing = cli(registry, runtime, "--todo-id", "todo_missing")
diff --git a/tests/control_plane/test_todo_list_agent_lane_projection.py b/tests/control_plane/test_todo_list_agent_lane_projection.py
index 365b9363c5..ff2ea55741 100644
--- a/tests/control_plane/test_todo_list_agent_lane_projection.py
+++ b/tests/control_plane/test_todo_list_agent_lane_projection.py
@@ -32,7 +32,7 @@ def _todo_line(
     ]
 
 
-def _write_fixture(tmp_path: Path) -> tuple[Path, Path]:
+def _write_fixture(tmp_path: Path, *, initialize: bool = True) -> tuple[Path, Path]:
     project = tmp_path / "project"
     runtime = tmp_path / "runtime"
     state_relative = Path(".local/goals") / GOAL_ID / "ACTIVE_GOAL_STATE.md"
@@ -195,6 +195,9 @@ def _write_fixture(tmp_path: Path) -> tuple[Path, Path]:
         + "\n",
         encoding="utf-8",
     )
+    from canonical_authority_fixture import initialize_canonical_source_fixture
+    if initialize:
+        initialize_canonical_source_fixture(runtime, GOAL_ID, state_path=state_file, provider="sqlite")
     return registry_path, state_file
 
 
@@ -264,6 +267,12 @@ def test_agent_lane_retains_bounded_blocked_advancement_reason(
     with state_file.open("a", encoding="utf-8") as stream:
         stream.write("\n".join(blocked) + "\n")
 
+    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
+    runtime = Path(json.loads(registry_path.read_text(encoding="utf-8"))["common_runtime_root"])
+    advance_canonical_fixture_authority(runtime, GOAL_ID,
+        build_todo_runtime_shadow_projection(goal_id=GOAL_ID, todos=read_fixture_todos(state_file),
+            handoff_mode="soft_claim"), operation_id="fixture-blocked-lane-added")
     payload = list_goal_todos(
         registry_path=registry_path,
         goal_id=GOAL_ID,
diff --git a/tests/control_plane/test_todo_list_record_references.py b/tests/control_plane/test_todo_list_record_references.py
index a2059d4706..834d23729d 100644
--- a/tests/control_plane/test_todo_list_record_references.py
+++ b/tests/control_plane/test_todo_list_record_references.py
@@ -50,7 +50,7 @@ def test_projection_failure_returns_the_standard_cli_error_without_source_change
     from loopx.control_plane import effect_runtime
 
     isolate_sqlite_runtime(tmp_path, monkeypatch)
-    registry, state = _write_fixture(tmp_path)
+    registry, state = _write_fixture(tmp_path, initialize=False)
     before = state.read_bytes()
 
     def unavailable(operation, payload, **kwargs):
@@ -70,13 +70,15 @@ def unavailable(operation, payload, **kwargs):
 @pytest.mark.parametrize("provider", ["legacy", "file", "sqlite"])
 def test_real_cli_references_reconstruct_all_scoped_lanes_and_counts(tmp_path, monkeypatch, provider):
     isolate_sqlite_runtime(tmp_path, monkeypatch)
-    registry, state = _write_fixture(tmp_path)
+    registry, state = _write_fixture(tmp_path, initialize=False)
     runtime = Path(json.loads(registry.read_text())["common_runtime_root"])
     if provider == "legacy":
-        from loopx.control_plane.coordination.local_authority import CanonicalAuthorityRequired
         before = state.read_bytes()
-        with pytest.raises(CanonicalAuthorityRequired):
+        from loopx.control_plane.coordination.local_authority import CanonicalAuthorityRequired
+        with pytest.raises(CanonicalAuthorityRequired) as error:
             list_goal_todos(registry_path=registry, goal_id=GOAL_ID, json_references=True)
+        assert error.value.payload["write_admitted"] is False
+        assert error.value.payload["legacy_fallback_used"] is False
         assert state.read_bytes() == before
         return
     if provider != "legacy":
diff --git a/tests/control_plane/test_todo_machine_section_projection.py b/tests/control_plane/test_todo_machine_section_projection.py
index 8f93ad05ea..e2593f80c0 100644
--- a/tests/control_plane/test_todo_machine_section_projection.py
+++ b/tests/control_plane/test_todo_machine_section_projection.py
@@ -846,7 +846,7 @@ def run(revision: str, *extra: str) -> tuple[int, dict]:
 
     code, result = run("unpromoted", "--execute")
     assert code == 1
-    assert "requires promoted canonical authority" in result["error"]
+    assert result["error_code"] == "canonical_authority_required"
     assert state.read_bytes() == source
 
     projection = build_todo_runtime_shadow_projection(goal_id="goal-a", todos=_records())
diff --git a/tests/control_plane/test_todo_projection_recovery.py b/tests/control_plane/test_todo_projection_recovery.py
index 1457cbe0cd..91f96c1e18 100644
--- a/tests/control_plane/test_todo_projection_recovery.py
+++ b/tests/control_plane/test_todo_projection_recovery.py
@@ -318,7 +318,7 @@ def test_unpromoted_missing_document_is_not_regenerated(canonical_display):
     state.unlink()
     legacy_coordination_writer_fence_path(runtime_root=runtime, goal_id="goal-a").unlink()
     code, result = _run(registry, revision, "--execute")
-    assert code == 1 and "requires promoted canonical authority" in result["error"]
+    assert code == 1 and result["error_code"] == "canonical_authority_required"
     assert not state.exists()
 
 
diff --git a/tests/control_plane/test_truncated_active_state_todos.py b/tests/control_plane/test_truncated_active_state_todos.py
index ece23f4859..f0021edcb1 100644
--- a/tests/control_plane/test_truncated_active_state_todos.py
+++ b/tests/control_plane/test_truncated_active_state_todos.py
@@ -9,6 +9,9 @@
 from loopx.control_plane.testing.canary_harness import run_json_cli_result
 from loopx.status import active_state_todo_fields, collect_status
 from loopx.todos import list_goal_todos
+from loopx.control_plane.coordination.local_authority import LocalCoordinationAuthorityUnavailable, CanonicalAuthorityRequired
+from loopx.control_plane.todos.active_state_todo_parser import active_state_incomplete_warning
+from tests.control_plane.canonical_authority_fixture import initialize_canonical_source_fixture
 
 
 GOAL_ID = "goal-truncated-state"
@@ -66,51 +69,28 @@ def _fixture(tmp_path: Path, state_text: str) -> tuple[dict[str, object], Path,
 def test_truncated_legacy_agent_todo_section_is_not_projected_as_empty(tmp_path: Path) -> None:
     healthy = _state_text(agent_heading="Agent Todo", agent_todo=True)
     wounded = healthy[: healthy.index("## Agent Todo")].rstrip() + "\n"
-    goal, runtime_root, registry_path, _state_path = _fixture(tmp_path, wounded)
-
-    listed = list_goal_todos(registry_path=registry_path, goal_id=GOAL_ID)
-    assert listed["ok"] is False
-    assert listed["error_code"] == "state_incomplete"
-    assert listed["todo_count"] == 0
-    assert listed["state_incomplete"]["kind"] == "state_incomplete"
-    assert listed["state_incomplete"]["roles"] == ["agent"]
-
-    fields = active_state_todo_fields(goal, runtime_root=runtime_root)
-    assert fields["state_incomplete"]["kind"] == "state_incomplete"
-    assert fields["state_incomplete"]["roles"] == ["agent"]
-
-    contract = check_contract(
-        registry_path=registry_path,
-        runtime_root_override=str(runtime_root),
-        scan_roots=[],
-        limit=5,
-        goal_id_filter=GOAL_ID,
-        include_public_boundary_scan=False,
-    )
+    goal, runtime_root, registry_path, state_path = _fixture(tmp_path, wounded)
+    original = state_path.read_bytes()
+    # Explicit source inspection still diagnoses the migration input. Ordinary
+    # readers must not use that input as a substitute for canonical authority.
+    source = active_state_incomplete_warning(wounded)
+    assert source["kind"] == "state_incomplete"
+    assert source["roles"] == ["agent"]
+    with pytest.raises(CanonicalAuthorityRequired) as error:
+        list_goal_todos(registry_path=registry_path, goal_id=GOAL_ID)
+    assert error.value.payload["write_admitted"] is False
+    assert error.value.payload["legacy_fallback_used"] is False
+    with pytest.raises(LocalCoordinationAuthorityUnavailable):
+        active_state_todo_fields(goal, runtime_root=runtime_root)
+    contract = check_contract(registry_path=registry_path, runtime_root_override=str(runtime_root),
+        scan_roots=[], limit=5, goal_id_filter=GOAL_ID, include_public_boundary_scan=False)
     assert contract["ok"] is False
-    assert any(
-        row["code"] == "state_incomplete"
-        and row["goal_id"] == GOAL_ID
-        for row in contract["error_diagnostics"]
-    )
-
-    status = collect_status(
-        registry_path=registry_path,
-        runtime_root_override=str(runtime_root),
-        limit=5,
-        goal_id=GOAL_ID,
-    )
-    assert status["ok"] is False
-    goal_attention = next(
-        item for item in status["attention_queue"]["items"] if item["goal_id"] == GOAL_ID
-    )
-    assert goal_attention["status"] == "state_incomplete"
-    assert goal_attention["waiting_on"] == "user_or_controller"
-    assert goal_attention["severity"] == "high"
-    assert goal_attention["state_incomplete"]["missing_sections"] == ["Agent Todo"]
-    assert goal_attention["project_asset"]["state_incomplete"]["roles"] == ["agent"]
-    assert "restore the missing Agent Todo section" in goal_attention["recommended_action"]
-    assert "first read-only adapter tick" not in goal_attention["recommended_action"]
+    assert any(row["code"] == "canonical_authority_required" and row["goal_id"] == GOAL_ID
+               for row in contract["error_diagnostics"])
+    with pytest.raises(LocalCoordinationAuthorityUnavailable):
+        collect_status(registry_path=registry_path, runtime_root_override=str(runtime_root),
+                       limit=5, goal_id=GOAL_ID)
+    assert state_path.read_bytes() == original
 
 
 @pytest.mark.parametrize("agent_heading", ["Agent Todo", "Codex Todo", "Project Agent Todo"])
@@ -122,6 +102,7 @@ def test_present_but_empty_agent_todo_section_remains_healthy(
         tmp_path,
         _state_text(agent_heading=agent_heading),
     )
+    initialize_canonical_source_fixture(runtime_root, GOAL_ID, state_path=_state_path)
 
     listed = list_goal_todos(registry_path=registry_path, goal_id=GOAL_ID)
     assert listed["ok"] is True
@@ -145,62 +126,25 @@ def test_present_but_empty_agent_todo_section_remains_healthy(
 def test_truncated_legacy_state_cli_paths_fail_closed(tmp_path: Path) -> None:
     healthy = _state_text(agent_heading="Agent Todo", agent_todo=True)
     wounded = healthy[: healthy.index("## Agent Todo")].rstrip() + "\n"
-    _goal, runtime_root, registry_path, _state_path = _fixture(tmp_path, wounded)
-
-    code, status = run_json_cli_result(
-        "status",
-        "--goal-id",
-        GOAL_ID,
-        registry_path=registry_path,
-        runtime_root=runtime_root,
-    )
-    assert code != 0
-    assert status["ok"] is False
-    assert any(
-        row["code"] == "state_incomplete" and row["goal_id"] == GOAL_ID
-        for row in status["contract"]["error_diagnostics"]
-    )
-    goal_attention = next(
-        item for item in status["attention_queue"]["items"] if item["goal_id"] == GOAL_ID
-    )
-    assert goal_attention["status"] == "state_incomplete"
-
-    code, listed = run_json_cli_result(
-        "todo",
-        "list",
-        "--goal-id",
-        GOAL_ID,
-        "--role",
-        "agent",
-        registry_path=registry_path,
-        runtime_root=runtime_root,
-    )
-    assert code != 0
-    assert listed["ok"] is False
-    assert listed["error_code"] == "state_incomplete"
-    assert listed["todo_count"] == 0
-
+    _goal, runtime_root, registry_path, state_path = _fixture(tmp_path, wounded)
+    original = state_path.read_bytes()
     registry = json.loads(registry_path.read_text(encoding="utf-8"))
     registry["goals"][0]["coordination"] = {
-        "registered_agents": ["agent-a"],
-        "agent_model": "peer_v1",
+        "registered_agents": ["agent-a"], "agent_model": "peer_v1",
     }
     registry_path.write_text(json.dumps(registry), encoding="utf-8")
-
-    code, quota = run_json_cli_result(
-        "quota",
-        "should-run",
-        "--goal-id",
-        GOAL_ID,
-        "--agent-id",
-        "agent-a",
-        registry_path=registry_path,
-        runtime_root=runtime_root,
-    )
-    assert code != 0
-    assert quota["should_run"] is False
-    assert quota["state"] == "blocked_health"
-    assert quota["status"] == "state_incomplete"
+    for arguments in [("status", "--goal-id", GOAL_ID),
+                      ("todo", "list", "--goal-id", GOAL_ID, "--role", "agent"),
+                      ("quota", "should-run", "--goal-id", GOAL_ID, "--agent-id", "agent-a")]:
+        code, payload = run_json_cli_result(*arguments, registry_path=registry_path,
+                                           runtime_root=runtime_root)
+        assert code != 0 and payload["ok"] is False
+        assert payload["error_code"] == "canonical_authority_required", payload
+        proof = payload.get("authority_read", payload)
+        assert proof["write_admitted"] is False and proof["legacy_fallback_used"] is False
+        if arguments[0] == "quota":
+            assert payload["should_run"] is False and payload["state"] == "blocked_health"
+        assert state_path.read_bytes() == original
 
 
 def test_malformed_todo_region_is_a_contract_error_not_a_crash(tmp_path: Path) -> None:
@@ -221,7 +165,7 @@ def test_malformed_todo_region_is_a_contract_error_not_a_crash(tmp_path: Path) -
     )
     assert contract["ok"] is False
     assert any(
-        row["code"] == "active_state_read_failed"
+        row["code"] == "canonical_authority_required"
         and row["goal_id"] == GOAL_ID
         for row in contract["error_diagnostics"]
     )
diff --git a/tests/control_plane/test_turn_completion_source.py b/tests/control_plane/test_turn_completion_source.py
index 64d373671f..69732e7d34 100644
--- a/tests/control_plane/test_turn_completion_source.py
+++ b/tests/control_plane/test_turn_completion_source.py
@@ -149,6 +149,11 @@ def legacy_context(*args, **kwargs):
 
     monkeypatch.setattr(context_readback, "attach_work_context", legacy_context)
     host, validator = _completion_host_and_validation_scripts()
+    host = host.replace(
+        "request = json.load(sys.stdin)",
+        'request = json.load(sys.stdin)\ncalls = pathlib.Path("host-calls.txt")\n'
+        'calls.write_text(calls.read_text() + "x" if calls.exists() else "x")',
+    )
     argv = _turn_run_once_completion_argv(project, runtime, registry, host, validator)
     actual_write = command.write_turn_validated_completion
     results = []
@@ -199,7 +204,22 @@ def recover_old_writer(**kwargs):
     )
     assert resumed_code == 0 and resumed["status"] == "committed", resumed
     assert resumed["effects"]["host_invoked"] is False
-    assert results[1]["provider_status"] == "replayed", results[1]
+    # Recovery reads the original accepted completion; it must not invoke the
+    # lifecycle writer again after its execution lease was retired.
+    assert len(results) == 1
+    recovered = _turn_journal(runtime)
+    assert recovered["writeback"]["completion"] == {
+        "todo_id": "todo_fixture0001", "continuation": "active_goal",
+    }
+    assert _todo(prefix)["todo"]["completion_turn_key"] == journal["turn_key"]
+    assert (project / "host-calls.txt").read_text() == "x"
+    replay_code, replay = _cli(
+        [], *argv[:-1], "--resume-turn-key", str(journal["turn_key"]), "--execute"
+    )
+    assert replay_code == 0 and replay["replayed"] is True, replay
+    assert not any(replay["effects"].values())
+    assert len(results) == 1
+    assert _turn_journal(runtime) == recovered
     assert _todo(prefix)["todo"]["status"] == "done"
 
 
@@ -409,6 +429,11 @@ def test_native_turn_recovers_accepted_completion_before_checking_new_text(
         "json",
     ]
     host, validator = _completion_host_and_validation_scripts()
+    host = host.replace(
+        "request = json.load(sys.stdin)",
+        'request = json.load(sys.stdin)\ncalls = pathlib.Path("host-calls.txt")\n'
+        'calls.write_text(calls.read_text() + "x" if calls.exists() else "x")',
+    )
     argv = _turn_run_once_completion_argv(project, runtime, registry, host, validator)
     actual_completion = command.write_turn_validated_completion
     completions: list[dict] = []
@@ -474,5 +499,20 @@ def lose_response_after_commit(**kwargs):
     )
     assert resumed_code == 0 and resumed["status"] == "committed", resumed
     assert resumed["effects"]["host_invoked"] is False
-    assert completions[1]["provider_status"] == "replayed", completions[1]
+    # Recovery reads the original accepted completion; it must not invoke the
+    # lifecycle writer again after its execution lease was retired.
+    assert len(completions) == 1
+    recovered = _turn_journal(runtime)
+    assert recovered["writeback"]["completion"] == {
+        "todo_id": "todo_fixture0001", "continuation": "active_goal",
+    }
+    assert _todo(prefix)["todo"]["completion_turn_key"] == journal["turn_key"]
+    assert (project / "host-calls.txt").read_text() == "x"
+    replay_code, replay = _cli(
+        [], *argv[:-1], "--resume-turn-key", str(journal["turn_key"]), "--execute"
+    )
+    assert replay_code == 0 and replay["replayed"] is True, replay
+    assert not any(replay["effects"].values())
+    assert len(completions) == 1
+    assert _turn_journal(runtime) == recovered
     assert _todo(prefix)["todo"]["text"] == "[P0] A later requirement after acceptance."
diff --git a/tests/control_plane/test_turn_envelope_capability_facts.py b/tests/control_plane/test_turn_envelope_capability_facts.py
index 107ccf0a0f..daec89b1e2 100644
--- a/tests/control_plane/test_turn_envelope_capability_facts.py
+++ b/tests/control_plane/test_turn_envelope_capability_facts.py
@@ -14,7 +14,7 @@
 from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection
 from loopx.control_plane.turn_driver.driver import build_loopx_turn_plan
 from tests.control_plane.canonical_authority_fixture import (
-    initialize_canonical_authority, isolate_sqlite_runtime,
+    initialize_canonical_authority, isolate_sqlite_runtime, read_fixture_todos,
 )
 from tests.control_plane.reward_memory_host_fixture import enable_live_memory
 from tests.control_plane.test_quota_settlement_cli import (
@@ -34,10 +34,8 @@ def test_real_captured_guard_preserves_required_and_missing_capabilities(
     state = project / f".codex/goals/{GOAL_ID}/ACTIVE_GOAL_STATE.md"
     state.write_text(state.read_text().replace("action_kind=validate ",
         "action_kind=validate continuation_policy=same_agent_non_delivery "))
-    code, listed = _run_cli(registry, runtime, "todo", "list", "--goal-id", GOAL_ID)
-    assert code == 0, listed
     projection = build_todo_runtime_shadow_projection(
-        goal_id=GOAL_ID, handoff_mode="soft_claim", todos=listed["todos"],
+        goal_id=GOAL_ID, handoff_mode="soft_claim", todos=read_fixture_todos(state),
     )
     initialize_canonical_authority(runtime, GOAL_ID, projection, state_path=state, provider=provider)
     if mode != "off":
@@ -88,10 +86,8 @@ def test_captured_vision_contract_preserves_closeout_and_original_observation(
     state = project / f".codex/goals/{GOAL_ID}/ACTIVE_GOAL_STATE.md"
     state.write_text(state.read_text().replace("action_kind=validate ",
         "action_kind=validate continuation_policy=same_agent_non_delivery "))
-    code, listed = _run_cli(registry, runtime, "todo", "list", "--goal-id", GOAL_ID)
-    assert code == 0, listed
     initialize_canonical_authority(runtime, GOAL_ID, build_todo_runtime_shadow_projection(
-        goal_id=GOAL_ID, handoff_mode="soft_claim", todos=listed["todos"],
+        goal_id=GOAL_ID, handoff_mode="soft_claim", todos=read_fixture_todos(state),
     ), state_path=state, provider=provider)
 
     # A synthetic retained checkpoint drives the real frontier and CLI owners.
@@ -185,7 +181,7 @@ def test_captured_vision_contract_preserves_closeout_and_original_observation(
     plan = build_loopx_turn_plan(envelope, host="dsh", execution_mode="isolated-headless",
                                  turn_instance_id="vision-contract-turn")
     assert plan["route"]["would_invoke_host"] is True
-    assert envelope["action"]["delivery_allowed"] is True
+    assert envelope["action"]["delivery_allowed"] is (not replan)
     request = {"schema_version": "loopx_turn_host_request_v0", "turn_key": plan["transaction"]["turn_key"],
                "route": "replan_required" if replan else "primary_delivery", "turn_envelope": envelope}
     result = subprocess.run(command, input=json.dumps(request), text=True, capture_output=True, timeout=30)
@@ -226,7 +222,8 @@ def test_captured_vision_contract_preserves_closeout_and_original_observation(
     read = subprocess.run(shlex.split(detail["full_decision"]), capture_output=True, check=True).stdout
     assert read == original
     assert build_turn_envelope(json.loads(read))["action_signature"]["source_decision_hash"] == authority["action_signature"]["source_decision_hash"]
-    assert "writeback" not in authority and "scheduler" not in authority
+    assert "writeback" not in authority
+    assert authority["scheduler"] == envelope["scheduler"]
     assert "Do not execute its state or scheduler commands" in prompt
     for field, value in (
         ("required_before_closeout", list(reversed(projected["required_before_closeout"]))),
diff --git a/tests/control_plane/test_turn_settlement_owed_signal.py b/tests/control_plane/test_turn_settlement_owed_signal.py
index bf71d77488..65d386766b 100644
--- a/tests/control_plane/test_turn_settlement_owed_signal.py
+++ b/tests/control_plane/test_turn_settlement_owed_signal.py
@@ -7,6 +7,7 @@
 
 from loopx.control_plane.quota.effect_program import SettlementIdentity
 from loopx.rollout_event_log import rollout_event_log_path
+from tests.control_plane.canonical_authority_fixture import initialize_canonical_source_fixture
 from loopx.state_refresh import refresh_state_run
 
 GOAL_ID = "goal-owed-spend"
@@ -74,6 +75,7 @@ def _fixture(tmp_path: Path) -> tuple[Path, Path, Path]:
         ),
         encoding="utf-8",
     )
+    initialize_canonical_source_fixture(tmp_path / "runtime", GOAL_ID, state_path=state_path)
     return registry_path, project, tmp_path / "runtime"
 
 
diff --git a/tests/control_plane/test_warm_continuation_readback.py b/tests/control_plane/test_warm_continuation_readback.py
index c7ca5d548f..9df671d948 100644
--- a/tests/control_plane/test_warm_continuation_readback.py
+++ b/tests/control_plane/test_warm_continuation_readback.py
@@ -88,7 +88,7 @@ def interleaved(*args):
                      "--claimed-by", "different-agent")
         return result
 
-    with pytest.raises(ValueError, match="preparation.*changed|not open"):
+    with pytest.raises(ValueError, match="preparation.*changed|not open|no actionable current preparation"):
         _read_preparation(interleaved, GOAL, AGENT, todo, {"deadline_epoch": int(time.time()) + 600})
     assert changed
     final = call("todo", "list", "--goal-id", GOAL, "--todo-id", todo)["todo"]
diff --git a/tests/control_plane/test_work_lane_narrative_cli.py b/tests/control_plane/test_work_lane_narrative_cli.py
index e4d011bc5c..d73021c253 100644
--- a/tests/control_plane/test_work_lane_narrative_cli.py
+++ b/tests/control_plane/test_work_lane_narrative_cli.py
@@ -9,6 +9,7 @@
 from pathlib import Path
 
 import pytest
+from tests.control_plane.canonical_authority_fixture import initialize_canonical_source_fixture
 
 
 GOAL_ID = "work-lane-narrative-fixture"
@@ -65,6 +66,7 @@ def _guard(
     (runs / "index.jsonl").write_text(json.dumps({
         **run, "json_path": str(run_path), "markdown_path": str(markdown_path),
     }) + "\n", encoding="utf-8")
+    initialize_canonical_source_fixture(runtime, GOAL_ID, state_path=state)
     result = subprocess.run(
         [sys.executable, "-c", "from loopx.cli import main; raise SystemExit(main())",
          "--registry", str(registry), "--runtime-root", str(runtime), "--format", "json",
diff --git a/tests/extensions/test_lark_direct_group_dispatch.py b/tests/extensions/test_lark_direct_group_dispatch.py
index 4a17a53867..882770488f 100644
--- a/tests/extensions/test_lark_direct_group_dispatch.py
+++ b/tests/extensions/test_lark_direct_group_dispatch.py
@@ -107,10 +107,15 @@ def reply(args):
         result = transport(args)
         if current_source in returns:
             returns[current_source] = reply_state["reply_text"]
-        result["stdout"] = (
-            result["stdout"].replace("linkmacbot", "LoopX Mew")
-            .replace("om_reply_fixture", "om_return_" + current_source)
-        )
+        payload = json.loads(result["stdout"].replace("linkmacbot", "LoopX Mew"))
+        if ("+messages-reply" in args or "+messages-send" in args) and "--dry-run" not in args:
+            original_id = payload["data"]["message_id"]
+            returned_id = "om_return_" + current_source
+            message = reply_state["messages"].pop(original_id)
+            message["message_id"] = returned_id
+            reply_state["messages"][returned_id] = message
+            payload["data"]["message_id"] = returned_id
+        result["stdout"] = json.dumps(payload)
         return result
 
     for event in events:
diff --git a/tests/extensions/test_lark_realtime_dispatch.py b/tests/extensions/test_lark_realtime_dispatch.py
index 661b84e09e..f94385b206 100644
--- a/tests/extensions/test_lark_realtime_dispatch.py
+++ b/tests/extensions/test_lark_realtime_dispatch.py
@@ -368,9 +368,9 @@ def unexpected(*_args):
 def test_native_source_lanes_preserve_topic_and_private_sender_order(tmp_path, monkeypatch):
     snapshot = _snapshot(tmp_path)
     snapshot["private_profiles"] = {
-        "mew": {"provider_ref": "a" * 24},
-        "alias": {"provider_ref": "a" * 24},
-        "other": {"provider_ref": "b" * 24},
+        "mew": {"provider_ref": "a" * 24, "audience": "group"},
+        "alias": {"provider_ref": "a" * 24, "audience": "group"},
+        "other": {"provider_ref": "b" * 24, "audience": "group"},
     }
 
     def unexpected(**_kwargs):
diff --git a/tests/test_chat_delegation_wake.py b/tests/test_chat_delegation_wake.py
index 9be4d42fc3..cecd7524ab 100644
--- a/tests/test_chat_delegation_wake.py
+++ b/tests/test_chat_delegation_wake.py
@@ -901,7 +901,6 @@ def test_lost_receipt_then_new_conversation_cannot_dispatch_twice(mode, monkeypa
 def test_the_in_turn_tool_pins_the_conversation_it_runs_in(mode, monkeypatch):  # noqa: F811
     """The model supplies the brief, never the wake routing: the host fixes it."""
     from loopx.collaboration_mcp import Delegations
-    from loopx.control_plane.testing.canary_harness import import_fixture_authority
 
     pinned = []
 
@@ -911,7 +910,6 @@ def start(self, binding_id, operation_id, brief, parent_request_id=None, *, conv
 
     monkeypatch.setattr(Delegations, "start", start)
     service, sid, _, settings, _ = mode
-    import_fixture_authority(service.controller.registry_path, "research")
     apply(mode, "start", settings=settings)
     adapter = type("A", (), {})()
     adapter.session = type("S", (), {})()
diff --git a/tests/test_cli_argument_diagnostics.py b/tests/test_cli_argument_diagnostics.py
index 3c7e312ad8..c3b06e4847 100644
--- a/tests/test_cli_argument_diagnostics.py
+++ b/tests/test_cli_argument_diagnostics.py
@@ -503,7 +503,7 @@ def test_quota_action_selection_requires_turn_identity() -> None:
         validate_quota_command_request(args)
 
     assert str(exc_info.value) == (
-        "`loopx quota should-run --todo-id` requires --turn-instance-id"
+        "`loopx quota should-run --todo-id` requires --turn-instance-id or --begin-turn"
     )
 
 
diff --git a/tests/test_delegation_stop_recovery.py b/tests/test_delegation_stop_recovery.py
index 7bc7c45721..38bf0d9c8b 100644
--- a/tests/test_delegation_stop_recovery.py
+++ b/tests/test_delegation_stop_recovery.py
@@ -50,14 +50,15 @@ def test_missing_cutover_marker_does_not_erase_an_unannotated_lease(service, mon
     assert inspect(runner)["lease"]["status"] == "released"
 
 
-def test_never_promoted_authority_needs_no_delegation_lease(service):
+def test_unpromoted_authority_cannot_prove_no_delegation_lease(service):
     from types import SimpleNamespace
 
     root, runner = service
     unused_runtime = root / "never-promoted"
     observer = SimpleNamespace(root=unused_runtime, goal_id=runner.goal_id)
     row = {"identity": {"binding": {"todo_id": "todo_analyst-initial"}}}
-    assert stop_lease.obligation(observer, row) is None
+    with pytest.raises(ValueError, match="canonical authority absence unproven"):
+        stop_lease.obligation(observer, row)
     assert not unused_runtime.exists(), "absence inspection must not create an authority"
 
 
diff --git a/tests/test_host_parity_smoke.py b/tests/test_host_parity_smoke.py
index 66a4d60553..d1aa3a3a61 100644
--- a/tests/test_host_parity_smoke.py
+++ b/tests/test_host_parity_smoke.py
@@ -162,7 +162,7 @@ def test_generic_cli_types_share_profile(self):
 
 class TestTurnHostIdentities:
     def test_supported_hosts(self):
-        assert SUPPORTED_HOSTS == {"codex-cli", "claude-code", "dsh", "generic-cli"}
+        assert SUPPORTED_HOSTS == {"codex-cli", "codex-app-server", "claude-code", "dsh", "generic-cli"}
         assert set(SUPPORTED_TURN_HOST_IDENTITIES) == {
             "codex-cli",
             "claude-code",
diff --git a/tests/test_lark_private_progress.py b/tests/test_lark_private_progress.py
index c8276a57fb..dc875ba8de 100644
--- a/tests/test_lark_private_progress.py
+++ b/tests/test_lark_private_progress.py
@@ -703,7 +703,19 @@ def inbox():
             sent = len(provider.writes)
             assert transport._deliver(path, record, "terminal", draft + "Final.", inbox=inbox)
             assert len(provider.writes) == sent
-            final_parts = [text.split("\n\n", 1)[1] for _, text in provider.writes[count:]]
+            from loopx.extensions.lark.post_delivery import _path
+
+            attempt = record["deliveries"]["terminal"]["attempt"]
+            assert attempt["message_ref"].startswith("lark-parts:")
+            parts = _read_json(_path(transport.runtime_root, inbox(), attempt["intent_digest"]))
+            concrete = [parts["delivery_part_attempts"][str(index)]["message_ref"]
+                        for index in range(len(parts["delivery_parts"]))]
+            assert len(concrete) == len(set(concrete))
+            # Inspect every retained provider post, including a verified draft
+            # prefix reused by the final delivery, against the authored text.
+            final_parts = [json.loads(provider.messages[ref]["body"]["content"])
+                           ["zh_cn"]["content"][0][0]["text"].split("\n\n", 1)[1]
+                           for ref in concrete]
             assert "".join(final_parts) == draft + "Final."
         assert all(message["msg_type"] == "post" for ref, message in provider.messages.items() if ref.startswith("om_out_"))
     finally:
diff --git a/tests/test_lark_source_authority_reply.py b/tests/test_lark_source_authority_reply.py
index 22ac9ab352..f8f010f581 100644
--- a/tests/test_lark_source_authority_reply.py
+++ b/tests/test_lark_source_authority_reply.py
@@ -21,7 +21,7 @@ def _deliver(transport, record, *, phase="admission"):
     return transport._deliver(path, record, phase, "Original source result", inbox=lambda: config)
 
 
-def test_send_then_lost_readback_recovers_with_one_fresh_source_authority_each_time(completed_group_stop):  # noqa: F811
+def test_send_fences_preflight_and_effect_but_readback_needs_no_second_send(completed_group_stop):  # noqa: F811
     _, provider, transport, root, _, record = completed_group_stop
     transport._inbox(record)
     provider.calls.clear()
@@ -30,7 +30,9 @@ def test_send_then_lost_readback_recovers_with_one_fresh_source_authority_each_t
     assert len(provider.writes) == 1
     before_send = next(i for i, args in enumerate(provider.calls)
                        if "+messages-reply" in args and "--dry-run" not in args)
-    assert sum("auth" in args for args in provider.calls[:before_send]) == 1
+    # One proof joins sender/source preflight; another fences the write after
+    # readonly provider preview. Recovery without a new write needs only one.
+    assert sum("auth" in args for args in provider.calls[:before_send]) == 2
     assert record["deliveries"]["admission"]["attempt"]["message_ref"] == "om_out_0"
     provider.verify_replies = True
     provider.calls.clear()
@@ -57,12 +59,41 @@ def test_ordinary_result_uses_the_same_fresh_app_proof_for_sender_and_source(ord
         transport._inbox(record)
         provider.calls.clear()
         assert _deliver(transport, record)
-        assert sum("auth" in args for args in provider.calls) == 1
+        assert sum("auth" in args for args in provider.calls) == 2
         assert provider.writes == [("notes-app", "Original source result")]
     finally:
         runtime.close()
 
 
+@pytest.mark.parametrize("change", ["app", "revoke"])
+def test_successful_preflight_cannot_authorize_a_send_after_authority_changes(
+    completed_group_stop, monkeypatch, change,  # noqa: F811
+):
+    _, provider, transport, _, _, record = completed_group_stop
+    transport._inbox(record)
+    original = transport._source_authority
+    proofs = []
+
+    def change_after_proof(*args):
+        proof = original(*args)
+        proofs.append(proof)
+        if len(proofs) == 1:
+            assert proof[0] and proof[1] is True
+            if change == "app":
+                provider.profile_apps["notes-app"] = "cli_other"
+            else:
+                transport.bindings.disconnect(
+                    record["binding_id"], expected_revision=transport.bindings.read()["revision"]
+                )
+        return proof
+
+    monkeypatch.setattr(transport, "_source_authority", change_after_proof)
+    assert not _deliver(transport, record)
+    assert len(proofs) == 2 and proofs[1][1] is False
+    assert not provider.writes and not provider.reaction_creates
+    assert not record["deliveries"]["admission"].get("started")
+
+
 @pytest.mark.parametrize("change", ["app", "bot", "owner", "revoke", "root", "sender", "chat", "parent", "thread"])
 def test_current_source_denial_never_becomes_a_send_or_receipt_intent(completed_group_stop, change):  # noqa: F811
     _, provider, transport, root, event, record = completed_group_stop
diff --git a/tests/test_projection_envelope.py b/tests/test_projection_envelope.py
index 5f69fbe4d8..20607d8d36 100644
--- a/tests/test_projection_envelope.py
+++ b/tests/test_projection_envelope.py
@@ -33,6 +33,7 @@
 from loopx.presentation.renderers.status_markdown import render_status_markdown
 from loopx.quota import build_quota_should_run
 from loopx.status import collect_status
+from tests.control_plane.canonical_authority_fixture import initialize_canonical_source_fixture
 
 
 def _registry(tmp_path: Path, goal_ids: tuple[str, ...] = ("goal-a", "goal-b")) -> Path:
@@ -61,6 +62,9 @@ def _registry(tmp_path: Path, goal_ids: tuple[str, ...] = ("goal-a", "goal-b"))
         ),
         encoding="utf-8",
     )
+    for goal_id in goal_ids:
+        initialize_canonical_source_fixture(tmp_path / "runtime", goal_id,
+            state_path=tmp_path / "ACTIVE_GOAL_STATE.md")
     return registry_path
 
 
@@ -1095,6 +1099,7 @@ def test_real_cli_global_membership_failure_and_recovery(tmp_path):
     )
     global_registry["goals"].append({**global_registry["goals"][0], "id": "goal-b",
                                      "state_file": str(second_state)})
+    initialize_canonical_source_fixture(runtime, "goal-b", state_path=second_state)
 
     def cli(command, registry_path=registry):
         result = subprocess.run(
diff --git a/tests/test_quota_bound_wait_recovery.py b/tests/test_quota_bound_wait_recovery.py
index f452d20f60..298708b653 100644
--- a/tests/test_quota_bound_wait_recovery.py
+++ b/tests/test_quota_bound_wait_recovery.py
@@ -10,7 +10,7 @@
 import pytest
 
 from control_plane.canonical_authority_fixture import (
-    initialize_canonical_authority,
+    initialize_canonical_authority, read_fixture_todos,
     isolate_sqlite_runtime,
 )
 from loopx.cli import main as cli_main
@@ -116,14 +116,12 @@ def cli(*args: str, cwd: Path = project) -> tuple[int, dict]:
             f"claimed_by={dependency_owner} {dependency_metadata} -->\n"
         )
     )
-    rc, listed = cli("todo", "list", "--goal-id", GOAL_ID)
-    assert rc == 0, listed
     initialize_canonical_authority(
         runtime,
         GOAL_ID,
         build_todo_runtime_shadow_projection(
             goal_id=GOAL_ID,
-            todos=listed["todos"],
+            todos=read_fixture_todos(state),
             handoff_mode="soft_claim",
             leases=[],
         ),
diff --git a/tests/test_runtime_archive_concurrency.py b/tests/test_runtime_archive_concurrency.py
index af9809ea4b..cc57245441 100644
--- a/tests/test_runtime_archive_concurrency.py
+++ b/tests/test_runtime_archive_concurrency.py
@@ -300,11 +300,11 @@ def paused_maintenance_lock(
         registry_path = tmp_path / "registry.json"
         _write_legacy_registry(registry_path, runtime_root)
 
-        def paused_effect(method: str, params: dict[str, Any]) -> Any:
+        def paused_effect(method: str, params: dict[str, Any], **kwargs: Any) -> Any:
             writer_paused.set()
             if not continue_writer.wait(timeout=10):
                 raise TimeoutError("legacy quota writer barrier was not released")
-            return actual_effect_runtime_result(method, params)
+            return actual_effect_runtime_result(method, params, **kwargs)
 
         monkeypatch.setattr(spend_commit, "effect_runtime_result", paused_effect)
 
diff --git a/tests/test_state_refresh_agent_lane_action.py b/tests/test_state_refresh_agent_lane_action.py
index aac244265a..aaa2d3c1ff 100644
--- a/tests/test_state_refresh_agent_lane_action.py
+++ b/tests/test_state_refresh_agent_lane_action.py
@@ -10,13 +10,21 @@
     resolve_refresh_recommendation,
 )
 from loopx.rollout_event_log import rollout_event_log_path
+from loopx.control_plane.todos.active_state_todo_parser import parse_active_state_todos
+from tests.control_plane.canonical_authority_fixture import initialize_canonical_source_fixture
 from loopx.state_refresh import (
     RECOMMENDED_ACTION_SOURCE_ACTIVE_NEXT_ACTION,
     RECOMMENDED_ACTION_SOURCE_AGENT_LANE_SELECTED_TODO,
-    derive_recommended_action_with_source,
     refresh_state_run,
 )
 
+def _recommendation(state_text, *, agent_id=None, todo_fields):
+    result = resolve_refresh_recommendation(
+        state_text, agent_id=agent_id, todo_fields=todo_fields,
+    )
+    return result["recommended_action"], result["recommended_action_source"]
+
+
 TWO_AGENT_STATE = """# Active Goal State
 
 ## Next Action
@@ -35,31 +43,31 @@
 
 
 def test_agent_scoped_derivation_prefers_own_lane_todo() -> None:
-    action, source = derive_recommended_action_with_source(
+    action, source = _recommendation(
         TWO_AGENT_STATE, agent_id="agent-a"
-    )
+    , todo_fields=parse_active_state_todos(TWO_AGENT_STATE, item_limit=None))
     assert action == "[P1] continue agent A scheduler coverage fix"
     assert source == RECOMMENDED_ACTION_SOURCE_AGENT_LANE_SELECTED_TODO
 
 
 def test_peer_lane_never_shadows_own_selection() -> None:
-    action, source = derive_recommended_action_with_source(
+    action, source = _recommendation(
         TWO_AGENT_STATE, agent_id="agent-b"
-    )
+    , todo_fields=parse_active_state_todos(TWO_AGENT_STATE, item_limit=None))
     assert action == "[P1] polish agent B release notes"
     assert source == RECOMMENDED_ACTION_SOURCE_AGENT_LANE_SELECTED_TODO
 
 
 def test_agent_without_claimed_todo_falls_back_to_shared_section() -> None:
-    action, source = derive_recommended_action_with_source(
+    action, source = _recommendation(
         TWO_AGENT_STATE, agent_id="agent-c"
-    )
+    , todo_fields=parse_active_state_todos(TWO_AGENT_STATE, item_limit=None))
     assert action == "Publish agent B's release checklist"
     assert source == RECOMMENDED_ACTION_SOURCE_ACTIVE_NEXT_ACTION
 
 
 def test_unscoped_derivation_keeps_shared_section_priority() -> None:
-    action, source = derive_recommended_action_with_source(TWO_AGENT_STATE)
+    action, source = _recommendation(TWO_AGENT_STATE, todo_fields=parse_active_state_todos(TWO_AGENT_STATE, item_limit=None))
     assert action == "Publish agent B's release checklist"
     assert source == RECOMMENDED_ACTION_SOURCE_ACTIVE_NEXT_ACTION
 
@@ -79,7 +87,7 @@ def test_blocked_p0_never_shadows_runnable_p1() -> None:
   <!-- loopx:todo status=open task_class=advancement_task claimed_by=agent-a todo_id=todo_runnable_delivery -->
 """
 
-    action, source = derive_recommended_action_with_source(state, agent_id="agent-a")
+    action, source = _recommendation(state, agent_id="agent-a", todo_fields=parse_active_state_todos(state, item_limit=None))
 
     assert action == "[P1] runnable delivery"
     assert source == RECOMMENDED_ACTION_SOURCE_AGENT_LANE_SELECTED_TODO
@@ -94,7 +102,7 @@ def test_agent_scope_without_own_candidate_never_falls_into_peer_lane() -> None:
   <!-- loopx:todo status=open task_class=advancement_task claimed_by=agent-b todo_id=todo_peer_delivery -->
 """
 
-    action, source = derive_recommended_action_with_source(state, agent_id="agent-a")
+    action, source = _recommendation(state, agent_id="agent-a", todo_fields=parse_active_state_todos(state, item_limit=None))
 
     assert "peer-only" not in action
     assert source == RECOMMENDED_ACTION_SOURCE_DEFAULT
@@ -119,7 +127,7 @@ def test_exact_settlement_todo_outranks_newer_lane_priority() -> None:
         state,
         agent_id="agent-a",
         settlement_identity=identity.as_dict(),
-    )
+     todo_fields=parse_active_state_todos(state, item_limit=None))
 
     assert (
         result["recommended_action"] == "[P1] continue agent A scheduler coverage fix"
@@ -144,7 +152,7 @@ def test_missing_settlement_todo_reports_typed_gap_before_lane_fallback() -> Non
         TWO_AGENT_STATE,
         agent_id="agent-a",
         settlement_identity=identity.as_dict(),
-    )
+     todo_fields=parse_active_state_todos(TWO_AGENT_STATE, item_limit=None))
 
     assert (
         result["recommended_action"] == "[P1] continue agent A scheduler coverage fix"
@@ -166,7 +174,7 @@ def test_unclaimed_lane_candidate_exposes_claim_prerequisite() -> None:
   <!-- loopx:todo status=open task_class=advancement_task todo_id=todo_unclaimed_slice -->
 """
 
-    result = resolve_refresh_recommendation(state, agent_id="agent-a")
+    result = resolve_refresh_recommendation(state, agent_id="agent-a", todo_fields=parse_active_state_todos(state, item_limit=None))
 
     assert result["recommended_action"] == "[P1] claim and continue this slice"
     assert (
@@ -182,7 +190,7 @@ def _write_turn_receipt(
     identity: SettlementIdentity,
 ) -> None:
     path = rollout_event_log_path(runtime_root, identity.goal_id)
-    path.parent.mkdir(parents=True)
+    path.parent.mkdir(parents=True, exist_ok=True)
     path.write_text(
         json.dumps(
             {
@@ -207,7 +215,7 @@ def test_refresh_state_run_reuses_exact_turn_selection(tmp_path: Path) -> None:
     goal_id = "goal-shared-refresh"
     project = tmp_path / "project"
     state_path = project / ".codex" / "goals" / goal_id / "ACTIVE_GOAL_STATE.md"
-    state_path.parent.mkdir(parents=True)
+    state_path.parent.mkdir(parents=True, exist_ok=True)
     state_path.write_text(TWO_AGENT_STATE, encoding="utf-8")
     registry_path = tmp_path / "registry.json"
     registry_path.write_text(
@@ -233,6 +241,7 @@ def test_refresh_state_run_reuses_exact_turn_selection(tmp_path: Path) -> None:
         encoding="utf-8",
     )
     runtime_root = tmp_path / "runtime"
+    initialize_canonical_source_fixture(runtime_root, goal_id, state_path=state_path)
     identity = SettlementIdentity(
         goal_id=goal_id,
         agent_id="agent-a",
diff --git a/tests/test_thread_agent_binding.py b/tests/test_thread_agent_binding.py
index a29b19b87b..4977b67044 100644
--- a/tests/test_thread_agent_binding.py
+++ b/tests/test_thread_agent_binding.py
@@ -921,6 +921,15 @@ def test_agent_directory_withholds_a_credential_shaped_thread_from_the_packet(
     """
 
     registry = _registry(tmp_path, ["agent-a"])
+    from tests.control_plane.canonical_authority_fixture import initialize_canonical_source_fixture
+    runtime = tmp_path / "runtime"
+    state = tmp_path / "ACTIVE_GOAL_STATE.md"
+    state.write_text("# Goal\n\n## Agent Todo\n", encoding="utf-8")
+    config = json.loads(registry.read_text())
+    config["common_runtime_root"] = str(runtime)
+    config["goals"][0].update(repo=str(tmp_path), state_file=str(state))
+    registry.write_text(json.dumps(config), encoding="utf-8")
+    initialize_canonical_source_fixture(runtime, "goal", state_path=state)
     secret_thread = "ghp_" + "1234567890abcdefghijklmnopqrstuvwxyz1234"
     assert (
         bind_thread_agent_in_registry(