test: align release fixtures with canonical authority and replan contracts

examples/backlog-hygiene-smoke.py

@@ -11,6 +11,9 @@
 
 
 REPO_ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(REPO_ROOT / "tests" / "control_plane"))
+from canonical_authority_fixture import initialize_canonical_source_fixture  # noqa: E402
+
 GOAL_ID = "backlog-hygiene-goal"
 
 
@@ -112,6 +115,7 @@ def write_fixture(
         + "\n",
         encoding="utf-8",
     )
+    initialize_canonical_source_fixture(runtime, GOAL_ID, state_path=state_path, provider="sqlite")
     return registry_path, runtime
 
 

loopx/control_plane/testing/actual_default_model_behavior_portfolio.py

@@ -336,11 +336,11 @@ class _ContrastSpec:
             "decision",
             "user_action_required",
             "must_attempt_work",
-            "delivery_allowed",
             "quiet_noop_allowed",
             "external_write_requested",
         ),
-        ("selected_todo_id",),
+        # The Host must attempt the replan, without ordinary delivery authority.
+        ("selected_todo_id", "delivery_allowed"),
     ),
 )
 ACTUAL_DEFAULT_MODEL_BEHAVIOR_CONTRAST_COUNT = len(_CONTRASTS)
@@ -411,42 +411,29 @@ def actual_default_model_behavior_scenario_catalog() -> dict[str, Any]:
 
 
 def _write_scenario_project(root: Path) -> tuple[Path, Path]:
+    root = root.resolve()
+    from ..coordination.new_goal_storage_transport import initialize_goal_storage_target
+    from ..goals.source_session_registration import configure_source_creation_goal
+
     project = root / "project"
-    state_file = (
-        project
-        / ".codex"
-        / "goals"
-        / ACTUAL_DEFAULT_MODEL_BEHAVIOR_FIXTURE_GOAL_ID
-        / "ACTIVE_GOAL_STATE.md"
-    )
+    state_file = project / ".codex" / "goals" / ACTUAL_DEFAULT_MODEL_BEHAVIOR_FIXTURE_GOAL_ID / "ACTIVE_GOAL_STATE.md"
     state_file.parent.mkdir(parents=True)
     state_file.write_text("# Active Goal State\n", encoding="utf-8")
     registry_path = project / ".loopx" / "registry.json"
     registry_path.parent.mkdir(parents=True)
-    registry_path.write_text(
-        json.dumps(
-            {
-                "schema_version": "0.1",
-                "goals": [
-                    {
-                        "id": ACTUAL_DEFAULT_MODEL_BEHAVIOR_FIXTURE_GOAL_ID,
-                        "status": "active",
-                        "repo": str(project),
-                        "state_file": str(state_file.relative_to(project)),
-                        "coordination": {
-                            "agent_model": "peer_v1",
-                            "registered_agents": [
-                                ACTUAL_DEFAULT_MODEL_BEHAVIOR_FIXTURE_AGENT_ID
-                            ],
-                        },
-                    }
-                ],
-            },
-            indent=2,
-        )
-        + "\n",
-        encoding="utf-8",
-    )
+    runtime_root = root / "runtime"
+    goal = {
+        "id": ACTUAL_DEFAULT_MODEL_BEHAVIOR_FIXTURE_GOAL_ID,
+        "status": "active", "repo": str(project),
+        "state_file": str(state_file.relative_to(project)),
+    }
+    # This is a new empty synthetic Goal, not an imported legacy authority.
+    # Reuse the first-party typed creation owner before requesting its frontier.
+    configure_source_creation_goal(goal, [ACTUAL_DEFAULT_MODEL_BEHAVIOR_FIXTURE_AGENT_ID],
+                                   "sqlite", "portfolio-entry-fixture")
+    registry_path.write_text(json.dumps({"schema_version": "0.1",
+        "common_runtime_root": str(runtime_root), "goals": [goal]}, indent=2) + "\n", encoding="utf-8")
+    initialize_goal_storage_target(runtime_root, goal, registry_path=registry_path)
     return project, registry_path
 
 

loopx/control_plane/testing/capability_monitor_repair_tool_behavior.py

@@ -156,6 +156,12 @@ def _build_capability_repair_fixture(root: Path) -> _SelectedTodoToolFixture:
         registry_path.parent.mkdir(parents=True, exist_ok=True)
         registry_path.write_text(registry_text + "\n", encoding="utf-8")
 
+    from .canary_harness import import_fixture_authority
+
+    import_fixture_authority(
+        global_registry_path, SELECTED_TODO_TOOL_FIXTURE_GOAL_ID, provider="sqlite",
+    )
+
     prompt = build_heartbeat_prompt(
         goal_id=SELECTED_TODO_TOOL_FIXTURE_GOAL_ID,
         thin=True,

loopx/control_plane/testing/replan_semantic_action_behavior.py

@@ -642,16 +642,16 @@ def _quota_behavior_observation(packet: Mapping[str, Any]) -> dict[str, Any]:
         raise ValueError("quota does not expose the semantic replan contract")
     if not (
         action.get("must_attempt") is True
-        and action.get("delivery_allowed") is True
+        and action.get("delivery_allowed") is False
         and action.get("quiet_noop_allowed") is False
     ):
-        raise ValueError("semantic replan must require one executable action")
+        raise ValueError("semantic replan must require its Host action without ordinary delivery authority")
     return {
         "decision": "execute",
         "selected_todo_id": None,
         "user_action_required": bool(user.get("action_required")),
         "must_attempt_work": True,
-        "delivery_allowed": True,
+        "delivery_allowed": False,
         "quiet_noop_allowed": False,
         "external_write_requested": False,
         "replan_obligation_id": obligation.get("obligation_id"),

loopx/control_plane/testing/replan_vision_closeout_behavior.py

@@ -72,6 +72,7 @@ def required_vision_scenario_contract(
         "selected_todo_id": None,
         "user_action_required": False,
         "must_attempt_work": True,
+        "delivery_allowed": False,
         "quiet_noop_allowed": False,
     }
     if any(contract.get(field) != value for field, value in required.items()):

loopx/control_plane/testing/scoped_gate_successor_tool_behavior.py

@@ -131,7 +131,7 @@ def _build_scoped_gate_fixture(root: Path) -> _SelectedTodoToolFixture:
         "## Completed Work Archive\n\n"
         "- [x] [P0] Complete the quality prerequisite.\n"
         "  <!-- loopx:todo "
-        f"todo_id={SCOPED_GATE_PREREQUISITE_TODO_ID} status=done "
+        f"todo_id={SCOPED_GATE_PREREQUISITE_TODO_ID} role=agent status=done "
         "task_class=advancement_task priority=P0 -->\n",
         encoding="utf-8",
     )
@@ -168,6 +168,12 @@ def _build_scoped_gate_fixture(root: Path) -> _SelectedTodoToolFixture:
         registry_path.parent.mkdir(parents=True, exist_ok=True)
         registry_path.write_text(registry_text + "\n", encoding="utf-8")
 
+    from .canary_harness import import_fixture_authority
+
+    import_fixture_authority(
+        global_registry_path, SELECTED_TODO_TOOL_FIXTURE_GOAL_ID, provider="sqlite",
+    )
+
     prompt = build_heartbeat_prompt(
         goal_id=SELECTED_TODO_TOOL_FIXTURE_GOAL_ID,
         thin=True,

loopx/control_plane/testing/selected_todo_tool_behavior.py

@@ -220,6 +220,12 @@ def _build_fixture(
         registry_path.parent.mkdir(parents=True, exist_ok=True)
         registry_path.write_text(registry_text + "\n", encoding="utf-8")
 
+    from .canary_harness import import_fixture_authority
+
+    import_fixture_authority(
+        global_registry_path, SELECTED_TODO_TOOL_FIXTURE_GOAL_ID, provider="sqlite",
+    )
+
     if prior_in_flight_progress:
         run = {
             "generated_at": "2026-08-12T00:01:00+00:00",

tests/control_plane/canonical_authority_fixture.py

@@ -99,6 +99,38 @@ def read_fixture_todos(state_path: Path) -> list[dict]:
     ]
 
 
+def advance_canonical_fixture_authority(
+    runtime_root: Path, goal_id: str, projection: dict, *, operation_id: str,
+) -> dict:
+    """Advance an already initialized disposable store through its real CAS.
+
+    Consumer fixtures may author a later synthetic state, but must not replace
+    storage bytes, reset lineage, re-import Markdown or re-engage its fence.
+    """
+    module = Path(__file__).resolve().parents[2] / "loopx/control_plane/coordination/local_authority_provider.ts"
+    script = (
+        f"import {{openLocalAuthorityStore}} from {json.dumps(module.as_uri())};"
+        "let input='';for await(const chunk of process.stdin)input+=chunk;"
+        "const request=JSON.parse(input);"
+        "const store=await openLocalAuthorityStore(request.root,request.goal);"
+        "const head=await store.loadAuthority();"
+        "if(head.status!=='loaded')throw new Error(JSON.stringify(head));"
+        "const result=await store.commitAuthority({expected_provider_revision:head.provider_revision,"
+        "operation_id:request.operation_id,events:[],next_projection:request.projection,receipts:[]});"
+        "process.stdout.write(JSON.stringify(result));"
+    )
+    process = subprocess.run(
+        ["node", "--no-warnings", "--experimental-strip-types", "--input-type=module", "-e", script],
+        input=json.dumps({"root": str(runtime_root), "goal": goal_id,
+                         "projection": projection, "operation_id": operation_id}),
+        capture_output=True, text=True, check=False, timeout=45,
+    )
+    assert process.returncode == 0, process.stderr
+    result = json.loads(process.stdout)
+    assert result["status"] == "applied", result
+    return result
+
+
 def initialize_canonical_source_fixture(
     runtime_root: Path, goal_id: str, *, state_path: Path,
     provider: str = "file", handoff_mode: str = "soft_claim",

tests/control_plane/test_actual_default_model_behavior_portfolio.py

@@ -348,7 +348,7 @@ def _replan_semantic_action_actor(_: str) -> dict[str, Any]:
         "replan_semantic_action_behavior_receipt_v0",
         selected_todo_id=None,
         user_action_required=False,
-        delivery_allowed=True,
+        delivery_allowed=False,
         semantic_action_accepted=True,
         context_delivery="host_projected",
         selected_semantic_outcomes=["fresh_vision_path_outcome"],
@@ -1659,7 +1659,10 @@ def test_portfolio_real_tool_scenarios_choose_from_latest_quota_result(
         **_real_tool_actors(tmp_path / "real-tool-actors"),
     )
 
-    assert result["qualification_passed"] is True
+    assert result["qualification_passed"] is True, [
+        (row["scenario_id"], row.get("failure_codes"), row.get("repeat_diagnostics"))
+        for row in result["scenarios"] if row["status"] != "passed"
+    ]
     boundary = result["boundary"]
     assert boundary["tools_enabled"] is True
     assert boundary["tool_enabled_scenario_count"] == 5

tests/control_plane/test_agent_capability_memory.py

@@ -9,7 +9,7 @@
 
 
 def test_next_turn_reuses_declared_runtime_capability(tmp_path):
-    project, runtime, registry = _write_fixture(tmp_path, required_capability="network")
+    project, runtime, registry = _write_fixture(tmp_path, required_capability="network", provider="sqlite")
     before = registry.read_bytes()
     args = ("quota", "should-run", "--goal-id", GOAL_ID, "--agent-id", AGENT_ID)
     rc, first = _run_cli(registry, runtime, *args, "--available-capability", "network")
@@ -29,7 +29,7 @@ def test_next_turn_reuses_declared_runtime_capability(tmp_path):
 
 
 def test_goal_inheritance_local_failure_recovery_and_forget(tmp_path):
-    _, runtime, registry = _write_fixture(tmp_path, required_capability="network", claimed=False)
+    _, runtime, registry = _write_fixture(tmp_path, required_capability="network", claimed=False, provider="sqlite")
     data = json.loads(registry.read_text())
     data["goals"][0]["coordination"]["available_capabilities"] = ["network"]
     data["goals"][0]["coordination"]["registered_agents"].append("agent-b")
@@ -58,7 +58,7 @@ def test_goal_inheritance_local_failure_recovery_and_forget(tmp_path):
 
 
 def test_owner_authority_and_optional_enablement_are_not_remembered(tmp_path):
-    _, runtime, registry = _write_fixture(tmp_path, required_capability="credentials")
+    _, runtime, registry = _write_fixture(tmp_path, required_capability="credentials", provider="sqlite")
     args = ("quota", "should-run", "--goal-id", GOAL_ID, "--agent-id", AGENT_ID)
     rc, first = _run_cli(registry, runtime, *args, "--available-capability", "credentials",
                          "--available-capability", "material_lifecycle")
@@ -69,7 +69,7 @@ def test_owner_authority_and_optional_enablement_are_not_remembered(tmp_path):
 
 
 def test_observation_does_not_cross_registry_or_agent(tmp_path):
-    _, runtime, registry = _write_fixture(tmp_path, required_capability="network", claimed=False)
+    _, runtime, registry = _write_fixture(tmp_path, required_capability="network", claimed=False, provider="sqlite")
     data = json.loads(registry.read_text())
     data["goals"][0]["coordination"]["registered_agents"].append("agent-b")
     registry.write_text(json.dumps(data))
@@ -85,7 +85,7 @@ def test_observation_does_not_cross_registry_or_agent(tmp_path):
 
 
 def test_turn_plan_stays_read_only_and_settlement_rechecks_memory(tmp_path):
-    _, runtime, registry = _write_fixture(tmp_path, required_capability="network")
+    _, runtime, registry = _write_fixture(tmp_path, required_capability="network", provider="sqlite")
     scope = ("--goal-id", GOAL_ID, "--agent-id", AGENT_ID)
     rc, plan = _run_cli(registry, runtime, "turn", "plan", *scope, "--available-capability", "network")
     assert rc == 0, plan
@@ -109,7 +109,7 @@ def test_executing_managed_turn_remembers_through_turn_start_hook(
     tmp_path, capsys, monkeypatch
 ):
     project, runtime, registry = _write_fixture(
-        tmp_path, required_capability="network"
+        tmp_path, required_capability="network", provider="sqlite",
     )
 
     def completed_turn(*args, **kwargs):
@@ -173,7 +173,7 @@ def completed_turn(*args, **kwargs):
 
 
 def test_agent_alias_uses_existing_identity_normalization(tmp_path):
-    _, runtime, registry = _write_fixture(tmp_path, required_capability="network")
+    _, runtime, registry = _write_fixture(tmp_path, required_capability="network", provider="sqlite")
     rc, first = _run_cli(registry, runtime, "quota", "should-run", "--goal-id", GOAL_ID,
                          "--agent-id", AGENT_ID.upper(), "--available-capability", "network")
     assert rc == 0, first

tests/control_plane/test_agent_context.py

@@ -159,12 +159,10 @@ def test_shared_host_delivers_signed_context_and_original_legacy_detail(legacy_r
     prompt = render_prompt(authority)
     assert json.dumps(envelope["agent_context"], ensure_ascii=False,
                       sort_keys=True, separators=(",", ":")) in prompt
-    if legacy_reference:
-        assert authority["detail_ref"] == envelope["detail_ref"]
-        assert authority["action_signature"] == envelope["action_signature"]
-    else:
-        assert "detail_ref" not in authority
-        assert "action_signature" not in authority
+    # The captured source accompanies current inline work context as well as
+    # historical context references; both retain the original signed carrier.
+    assert authority["detail_ref"] == envelope["detail_ref"]
+    assert authority["action_signature"] == envelope["action_signature"]
     invalid = deepcopy(envelope)
     invalid["agent_context"]["scope"]["agent_id"] = "different-agent"
     with pytest.raises(ValueError, match="action signature"):