test: restore canonical fixture and durable recovery regression coverage

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.

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.

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

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
 
 

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

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"

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.

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

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

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

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)

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

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
 
 

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,

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,