test: repair native recovery and staged settlement readers

tests/control_plane/codex_checkpoint_fixture.py

@@ -114,6 +114,26 @@ def reference_metadata_oracle(view):
     return view
 
 
+def read_cli_policy_payload(stdout, *, returncode):
+    """Read the current policy capture for tests, without claiming model retention.
+
+    Preserve failed commands and inline/off output. Successful native full/delta
+    navigation uses its complete current recovery, never a cached or delta body.
+    """
+    from loopx.control_plane.turn_driver.decision_capture import (
+        read_context_capture_file,
+        read_context_read_reference,
+    )
+
+    payload = json.loads(stdout)
+    if returncode == 0 and payload.get("mode") in {"full", "delta"} and isinstance(payload.get("recovery"), dict):
+        recovery = payload["recovery"]
+        if "reference_file" in recovery:
+            recovery = read_context_read_reference(recovery["reference_file"])[1]
+        return read_context_capture_file(recovery["path"], recovery["sha256"], private=True)[2]
+    return payload
+
+
 class NativeContextReader:
     """Static inference which follows only model-visible native reader cursors.
 

tests/control_plane/test_context_read_reference.py

@@ -11,6 +11,7 @@
 from loopx.control_plane.turn_driver.decision_capture import (
     capture_page_range, read_context_read_reference, write_context_read_reference,
 )
+from tests.control_plane.codex_checkpoint_fixture import read_cli_policy_payload
 
 
 def read(command, **overrides):
@@ -31,6 +32,46 @@ def fixture(tmp_path):
     return payload, view, model_context_reference(view)
 
 
+@pytest.mark.parametrize("mode", ["full", "delta"])
+@pytest.mark.parametrize("reference", [False, True], ids=["explicit", "navigation"])
+def test_policy_oracle_reads_complete_current_recovery(tmp_path, mode, reference):
+    payload, machine, model = fixture(tmp_path)
+    # Full and delta transports both expose the current complete recovery.
+    # A decoy transport/baseline must never replace those facts in a policy test.
+    view = dict(model if reference else machine, mode=mode)
+    view["path"] = str(tmp_path / "unavailable-delta.json")
+    view["baseline_reference"] = {"path": str(tmp_path / "old-capture.json")}
+    before = {path: path.read_bytes() for path in tmp_path.iterdir() if path.is_file()}
+    assert read_cli_policy_payload(json.dumps(view), returncode=0) == payload
+    assert {path: path.read_bytes() for path in tmp_path.iterdir() if path.is_file()} == before
+
+
+@pytest.mark.parametrize("returncode", [0, 1])
+def test_policy_oracle_preserves_inline_output_and_failed_commands(tmp_path, returncode):
+    inline = {"ok": returncode == 0, "decision": "skip", "error": "original failure"}
+    assert read_cli_policy_payload(json.dumps(inline), returncode=returncode) == inline
+    if returncode:
+        _, _, model = fixture(tmp_path)
+        failed = {**model, "ok": False, "error": "command failed after capture"}
+        # Keep the failure even if a complete successful capture is available.
+        assert read_cli_policy_payload(json.dumps(failed), returncode=returncode) == failed
+
+
+@pytest.mark.parametrize("damage", ["digest", "capture_symlink", "reference_symlink"])
+def test_policy_oracle_rejects_replaced_current_capture(tmp_path, damage):
+    _, machine, model = fixture(tmp_path)
+    if damage == "digest":
+        capture = Path(machine["path"])
+        capture.write_bytes(capture.read_bytes() + b" ")
+    else:
+        target = Path(machine["path"] if damage == "capture_symlink" else model["recovery"]["reference_file"])
+        saved = target.with_suffix(".original")
+        target.rename(saved)
+        target.symlink_to(saved)
+    with pytest.raises(ValueError, match="digest mismatch|physical.*symlink"):
+        read_cli_policy_payload(json.dumps(model), returncode=0)
+
+
 def test_native_navigation_keeps_digests_in_immutable_reference(tmp_path):
     payload, machine, model = fixture(tmp_path)
     raw = Path(machine["path"]).read_bytes()

tests/control_plane/test_quota_settlement_cli.py

@@ -26,10 +26,6 @@
 )
 from loopx.control_plane.quota.settlement import render_settlement_progress_markdown
 from loopx.control_plane.todos.active_state_todo_parser import parse_active_state_todos
-from loopx.control_plane.turn_driver.decision_capture import (
-    read_context_capture_file,
-    read_context_read_reference,
-)
 from loopx.control_plane.scheduler.state import (
     APP_AUTOMATION_STATEFUL_BACKOFF_STATE_KEY,
     load_scheduler_state,
@@ -41,6 +37,7 @@
     initialize_canonical_source_fixture,
     read_fixture_todos,
 )
+from tests.control_plane.codex_checkpoint_fixture import read_cli_policy_payload
 
 REPO_ROOT = Path(__file__).resolve().parents[2]
 GOAL_ID = "settlement-cli-fixture"
@@ -281,16 +278,7 @@ def _run_generated_cli(
         text=True,
         env={**os.environ, "PYTHONPATH": str(REPO_ROOT)},
     )
-    payload = json.loads(result.stdout)
-    if payload.get("mode") in {"full", "delta"} and isinstance(payload.get("recovery"), dict):
-        # Execute the generated default-on command unchanged. These policy
-        # oracles inspect its current immutable capture without another guard,
-        # admission or retention claim; native reader coverage lives separately.
-        recovery = payload["recovery"]
-        if "reference_file" in recovery:
-            recovery = read_context_read_reference(recovery["reference_file"])[1]
-        _, _, payload = read_context_capture_file(recovery["path"], recovery["sha256"], private=True)
-    return result.returncode, payload
+    return result.returncode, read_cli_policy_payload(result.stdout, returncode=result.returncode)
 
 
 def _spend_run_count(runtime: Path) -> int:
@@ -6877,13 +6865,14 @@ def test_todoless_blocked_replan_settles_read_only_external_evidence_without_wor
     assert replay.get("unsettled_host_turn_recovery") is None
 
 
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
 def test_unbound_visible_goal_todoless_replan_reenters_through_guided_turn(
-    tmp_path: Path,
+    tmp_path: Path, provider: str,
 ) -> None:
     project, runtime, registry_path = _write_fixture(tmp_path)
     _configure_autonomous_replan_fixture(project, runtime, registry_path)
 
-    _initialize_fixture_authority(project, runtime)
+    _initialize_fixture_authority(project, runtime, provider=provider)
     unbound_rc, unbound = _run_cli(
         registry_path,
         runtime,
@@ -6920,11 +6909,42 @@ def test_unbound_visible_goal_todoless_replan_reenters_through_guided_turn(
     assert identity["replan_obligation_id"] == obligation_id
     assert identity["turn_instance_id"].startswith("guided-start:")
     cli_channel = bound["interaction_contract"]["cli_channel"]
-    assert cli_channel["settlement_plan"]["identity"] == identity
-    assert len(cli_channel["next_cli_actions"]) == 2
-    for command in cli_channel["next_cli_actions"]:
+    assert cli_channel["settlement_identity"] == identity
+    assert cli_channel["settlement_read_ref"] == "$.replan_action_packet.writeback"
+    assert "settlement_plan" not in cli_channel
+    # Native admission defers the plan; consume its exact captured reader
+    # without another guard or a second guided-start Turn.
+    reader_argv = shlex.split(bound["replan_action_packet"]["writeback"])
+    source_paths = (
+        registry_path,
+        project / f".codex/goals/{GOAL_ID}/ACTIVE_GOAL_STATE.md",
+        runtime / "goals" / GOAL_ID / "runs/index.jsonl",
+        Path(reader_argv[reader_argv.index("--decision-file") + 1]),
+    )
+    original_files = {path: path.read_bytes() for path in source_paths}
+    reader = subprocess.run(
+        reader_argv,
+        cwd=REPO_ROOT,
+        env={**os.environ, "PYTHONPATH": str(REPO_ROOT)},
+        capture_output=True,
+        text=True,
+        timeout=60,
+    )
+    assert reader.returncode == 0, (reader.stdout, reader.stderr)
+    writeback = json.loads(reader.stdout)
+    assert writeback["obligation_id"] == obligation_id
+    plan = writeback["writeback"]["settlement_plan"]
+    assert plan["identity"] == identity
+    assert [step["kind"] for step in plan["ordered_steps"]] == [
+        "validation", "durable_writeback", "quota_spend",
+    ]
+    commands = [step["command_template"] for step in plan["ordered_steps"][1:]]
+    for command in commands:
         assert f"--replan-obligation-id {obligation_id}" in command
         assert f"--turn-instance-id {identity['turn_instance_id']}" in command
+    assert "--source visible-goal" in commands[1]
+    assert _spend_run_count(runtime) == 0
+    assert {path: path.read_bytes() for path in source_paths} == original_files
 
 
 def test_autonomous_replan_semantic_delta_keeps_accountable_receipt_chain(

tests/control_plane/test_refresh_checkpoint_isolation.py

@@ -18,6 +18,7 @@
 from loopx.extensions.runtime import install_extension
 from loopx.global_registry import sync_project_registry_to_global
 from loopx.state_refresh import render_state_refresh_markdown
+from tests.control_plane.codex_checkpoint_fixture import read_cli_policy_payload
 from tests.control_plane.test_refresh_checkpoint_recovery import first_refresh as first_refresh
 from tests.control_plane.test_quota_settlement_cli import (
     AGENT_ID, GOAL_ID, REPO_ROOT, TODO_ID, TURN_ID,
@@ -123,7 +124,7 @@ def run(argv, output="json", expected_rc=0):
         rc = cli.main(["--format", output, *argv])
         stdout = capsys.readouterr().out
         assert rc == expected_rc, stdout
-        return json.loads(stdout) if output == "json" else stdout
+        return read_cli_policy_payload(stdout, returncode=rc) if output == "json" else stdout
 
     if baseline:
         run([*prefix, "refresh-state", "--goal-id", GOAL_ID, "--agent-id", AGENT_ID,

tests/control_plane/test_refresh_external_delivery.py

@@ -9,6 +9,7 @@
 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.codex_checkpoint_fixture import read_cli_policy_payload
 from tests.control_plane.test_quota_settlement_cli import (
     AGENT_ID, GOAL_ID, TODO_ID, TURN_ID, _write_fixture, _initialize_fixture_authority,
 )
@@ -29,7 +30,7 @@ def run(args, expected=0):
         rc = cli.main(["--format", "json", *prefix, *args])
         stdout = capsys.readouterr().out
         assert rc == expected, stdout
-        return json.loads(stdout)
+        return read_cli_policy_payload(stdout, returncode=rc)
 
     assert run(["quota", "should-run", "--codex-app", *binding,
                 ])["decision"] == "run"