fix(context): recover captured replans independently of registry routing

loopx/cli_runtime.py

@@ -173,12 +173,12 @@ def resolve_cli_registry(
 	registry_was_configured = user_supplied_registry(raw_argv) or bool(
 		os.environ.get("LOOPX_REGISTRY")
 	)
-	if args.command == "quota" and getattr(args, "quota_command", None) == "context-read":
+	if args.command == "quota" and getattr(args, "quota_command", None) in {"context-read", "replan-view"}:
 		from .cli_commands.context_machine import has_explicit_context_read_source
 
-		# The reader owns capture integrity and native-session scope. It cannot
-		# reach Goal authority, so default registry conflicts are unrelated.
-		if has_explicit_context_read_source(args):
+		# Captured readers own source validation (including missing references)
+		# and native-session scope; neither reaches live Goal authority.
+		if args.quota_command == "replan-view" or has_explicit_context_read_source(args):
 			return registry_path, registry_was_configured
 	if str(args.registry) == HOST_GLOBAL_REGISTRY_SELECTOR:
 		try:

tests/control_plane/test_context_capture_registry_independence.py

@@ -3,12 +3,14 @@
 import json
 import os
 from pathlib import Path
+import shlex
 import subprocess
 import sys
 
 import pytest
 
 from loopx import paths
+from loopx.cli_runtime import HOST_GLOBAL_REGISTRY_SELECTOR
 from test_codex_context_lifecycle import project, setup
 
 
@@ -93,3 +95,111 @@ def test_implicit_source_and_authority_commands_preserve_route_conflict(tmp_path
     result = run_cli("loopx.entrypoint", tmp_path, *command)
     assert result.returncode != 0
     assert "Both default LoopX runtime roots contain state" in result.stdout + result.stderr
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_replan_stages_read_original_authority_capture_without_registry(
+    tmp_path, monkeypatch, conflicting_defaults, provider,
+):
+    from canonical_authority_fixture import isolate_sqlite_runtime
+    from loopx.control_plane.effect_runtime import restart_effect_runtime
+    from test_replan_progress_native_cli import (
+        AGENT, GOAL, TODO, TURN, call_progress_cli, make_progress_cli_fixture,
+    )
+
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    fixture = make_progress_cli_fixture(tmp_path, provider, "advanced")
+    registry = json.loads(fixture["registry"].read_bytes())
+    registry["goals"][0]["explore_graph"] = {"enabled": True}
+    fixture["registry"].write_text(json.dumps(registry))
+    document = tmp_path / "outcome.json"
+    document.write_text(json.dumps({"outcome_id": "original-artifact", "owner_agent_id": AGENT,
+        "objective": "Validate the original artifact", "todo_ids": [TODO],
+        "requirements": [{"id": "retained-stop", "description": "Stop before publication."}],
+        "source": {"ref": "message:original-request", "summary": "Original accepted boundary"}}))
+    try:
+        revision = call_progress_cli(fixture, "goal-acceptance", "inspect", "--goal-id", GOAL)["provider_revision"]
+        call_progress_cli(fixture, "goal-acceptance", "refine", "--goal-id", GOAL, "--agent-id", AGENT,
+            "--document", str(document), "--expected-provider-revision", revision,
+            "--operation-id", "original-outcome", "--execute")
+        guard = call_progress_cli(fixture, "quota", "should-run", "--codex-app",
+            "--goal-id", GOAL, "--agent-id", AGENT, "--turn-instance-id", TURN,
+            "--todo-id", TODO, "--decision-output-dir", str(tmp_path / "capture"))
+        capture = tmp_path / "capture" / "decision.json"
+        source = json.loads(capture.read_bytes())
+        identity = source["heartbeat_receipt"]["settlement_identity"]
+        assert (identity["goal_id"], identity["agent_id"], identity["turn_instance_id"]) == (GOAL, AGENT, TURN)
+        missing = tmp_path / "unrelated-missing-registry.json"
+
+        def snapshot():
+            return {str(p.relative_to(tmp_path)): p.read_bytes()
+                for root in (fixture["project"], fixture["runtime"], capture.parent)
+                for p in root.rglob("*") if p.is_file()} | {"registry": fixture["registry"].read_bytes()}
+
+        before = snapshot()
+        for module in ("loopx.entrypoint", "loopx.cli"):
+            for route in ((), ("--registry", str(missing)), ("--registry", HOST_GLOBAL_REGISTRY_SELECTOR)):
+                def read(command):
+                    argv = shlex.split(command)
+                    result = run_cli(module, fixture["project"], *route, *argv[argv.index("quota"):])
+                    assert result.returncode == 0, result.stdout + result.stderr
+                    value = json.loads(result.stdout)
+                    assert value["ok"] and (value["goal_id"], value["agent_id"]) == (GOAL, AGENT)
+                    assert value["obligation_id"] == source["replan_action_packet"]["obligation_id"]
+                    return value
+
+                review = read(guard["replan_action_packet"]["review"])
+                writer = read(review["next"]["writeback"])
+                assert writer["context_id"] == review["context_id"]
+                assert writer["source_hash"] == review["source_hash"]
+                assert writer["writeback"]["settlement_plan"]["identity"] == identity
+                index = review["acceptance_index"]
+                assert len(index["outcomes"]) == 1
+                acceptance = read(index["read_action_template"].replace("{outcome_ref}", index["outcomes"][0]["outcome_ref"]))
+                assert acceptance["outcome"]["requirements"][0]["description"] == "Stop before publication."
+                explore = read(guard["replan_action_packet"]["review"].replace("--replan-stage review", "--replan-stage explore"))
+                assert explore["explore_basis"] == source["autonomous_replan_obligation"]["replan_context"].get("explore_basis")
+        assert snapshot() == before, "captured readers cannot admit, settle, repair routes or rewrite authority"
+        assert not missing.exists()
+    finally:
+        restart_effect_runtime(wait_for_exit=True)
+
+
+@pytest.mark.parametrize("module", ["loopx.entrypoint", "loopx.cli"])
+def test_replan_capture_failures_belong_to_reader_not_default_route(tmp_path, conflicting_defaults, module):
+    from loopx.control_plane.quota.turn_envelope import _turn_envelope_result
+    from test_replan_host_context_projection import _quota_payload
+
+    capture = tmp_path / "source.json"
+    source = _quota_payload()
+    source["replan_read_path"] = str(capture)
+
+    def command(value, stage="review"):
+        capture.write_text(json.dumps(value))
+        packet = _turn_envelope_result("replan_cli_view", payload=value)["replan_action_packet"]
+        argv = shlex.split(packet[stage])
+        return argv[argv.index("quota"):]
+
+    def rejected(argv, reason):
+        result = run_cli(module, tmp_path, *argv)
+        assert result.returncode == 1, result.stdout + result.stderr
+        value = json.loads(result.stdout)
+        assert not value["ok"] and reason in value["reason"], value
+
+    argv = command(source)
+    rejected(["quota", "replan-view"], "requires --decision-file and --source-hash")
+    rejected(["quota", "replan-view", "--decision-file", str(capture)], "requires --decision-file and --source-hash")
+    rejected(["quota", "replan-view", "--source-hash", "sha256:" + "0" * 64], "requires --decision-file and --source-hash")
+    wrong = list(argv)
+    wrong[wrong.index("--source-hash") + 1] = "sha256:" + "0" * 64
+    rejected(wrong, "source hash")
+    capture.write_text("{")
+    rejected(argv, "Expecting")
+    capture.unlink()
+    rejected(argv, "No such file")
+    source["autonomous_replan_obligation"]["replan_context"]["obligation_id"] = "different-obligation"
+    rejected(command(source), "matching decision context")
+    source = _quota_payload()
+    source["replan_read_path"] = str(capture)
+    source["interaction_contract"]["schema_version"] = "unsupported-contract-version"
+    rejected(command(source, "writeback"), "schema_version")