fix(context): keep saved projections independent of Goal discovery

loopx/cli_runtime.py

@@ -179,12 +179,12 @@ def resolve_cli_registry(
 		# Service activation reads its route from the loaded original manager.
 		# It neither opens Goal authority nor guesses a default Chat store.
 		return registry_path, registry_was_configured
-	if args.command == "quota" and getattr(args, "quota_command", None) in {"context-read", "replan-view"}:
+	if args.command == "quota" and getattr(args, "quota_command", None) in {"context-read", "context-project", "replan-view"}:
 		from .cli_commands.context_machine import has_explicit_context_read_source
 
-		# 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):
+		# Saved views own source validation (including missing references) and
+		# native-session scope; machine defaults keep their separate route owner.
+		if args.quota_command in {"context-project", "replan-view"} or has_explicit_context_read_source(args):
 			return registry_path, registry_was_configured
 	if str(args.registry) == HOST_GLOBAL_REGISTRY_SELECTOR:
 		try:

skills/loopx-self-repair/references/repair-patterns.md

@@ -5,6 +5,7 @@ teaches a reusable control-plane lesson.
 
 | Pattern | Symptoms | Evidence To Read | Likely Root | Durable Repair |
 | --- | --- | --- | --- | --- |
+| `saved_projection_goal_route_coupling` | A saved context projection fails on conflicting default runtime roots even with native transport disabled, while the same capture remains readable. | Run the shipped projection and returned reader against one digest-bound synthetic capture, with implicit, missing and global registry selectors; retain digest/identity refusals and unchanged runtime snapshots. | Shared CLI registry discovery runs before the saved-view owner validates its source. | Classify the saved presentation at the existing CLI route owner; preserve machine-configuration routing, live authority conflict checks and source/session validation. Reuse the registry-independence integration tests rather than adding a second recovery path or repairing live state. |
 | `host_bridge_reader_orphan` | A rejected bridge observation leaves its reader thread and output pipe alive after the child has exited. | Original framing error, cleanup result, child exit, bounded-queue backpressure and reader/pipe state across repeated failures. | Cancellation rejects another malformed frame; the sole consumer leaves while the producer blocks on a full queue, and joining alone cannot release it. | Stop publication, release the waiting producer, join and close through the existing transport helper. Retain cleanup-unconfirmed results and original Host drain ownership; reclaiming a reader does not prove Host exit. Cover malformed JSON and invalid typed observations with real subprocesses, plus existing real Host cancellation and success paths. Do not increase timeouts or fabricate a successful result. |
 | `codex_strict_result_schema_gap` | A Codex Turn fails with invalid_json_schema before execution, or a strict-schema stop result fails internal evidence validation. | Generated Codex schema, required properties, unsupported conditional keywords, and both CLI/app-server result boundaries. | Optional evidence and conditional schema requirements conflict with strict Structured Outputs; internal stop results prohibit evidence fields. | Require nullable evidence fields on the Codex wire and remove unsupported conditionals. Normalize only stop null placeholders in both Codex transports; reject missing/null profile material evidence and preserve non-null evidence for the existing validator. Cover legacy absence, invalid arrays and an isolated schema-acceptance probe without changing other hosts. |
 | `initial_preference_view_omission` | A fulfilled preference hook leaves corrections, retirement or advisory instructions absent from the recommended initial read. | The preference owner's complete current scoped view, actual selected reader/native input, empty and unavailable states, and same-capture continuation/recovery. | A hand-picked projection uses incomplete or stale child field names after the owner has already delivered the current view. | Reuse the complete owned current view in the existing initial selector; retain provenance, advisory authority, exact-scope freshness and immutable capture checks. Cover real File/SQLite lifecycle reads, empty replacements, unknown values and native consumer readback. Preserve changed-profile refusal and fresh recovery; disclose restored bytes as information cost, not compression or model benefit. |

tests/control_plane/test_context_capture_registry_independence.py

@@ -65,6 +65,71 @@ def test_explicit_capture_and_selected_details_ignore_conflicting_defaults(tmp_p
     assert capture.read_bytes() == raw
 
 
+@pytest.mark.parametrize("module", ["loopx.entrypoint", "loopx.cli"])
+@pytest.mark.parametrize("route", ["implicit", "missing", "global"])
+def test_saved_projection_without_native_transport_needs_no_goal_route(
+    tmp_path, conflicting_defaults, module, route,
+):
+    original = {"ok": True, "goal_id": "fixture", "agent_identity": {"agent_id": "agent"},
+        "heartbeat_receipt": {"turn_instance_id": "original-turn"},
+        "requirement": "Preserve the original source and final stop.", "counter": 9007199254740993}
+    capture = tmp_path / "original.json"
+    raw = json.dumps(original, ensure_ascii=False).encode()
+    capture.write_bytes(raw)
+    digest = hashlib.sha256(raw).hexdigest()
+    missing = tmp_path / "unrelated-registry.json"
+    options = {"implicit": (), "missing": ("--registry", str(missing)),
+        "global": ("--registry", HOST_GLOBAL_REGISTRY_SELECTOR)}[route]
+    command = ("quota", "context-project", "--capture-file", str(capture),
+        "--capture-sha256", digest, "--goal-id", "fixture", "--agent-id", "agent",
+        "--turn-instance-id", "original-turn", "--context-delivery", "off")
+    result = run_cli(module, tmp_path, *options, *command)
+    assert result.returncode == 0, result.stdout + result.stderr
+    projected = json.loads(result.stdout)
+    assert projected["ok"] and projected["mode"] == "full"
+    assert projected["recovery"]["path"] == str(capture)
+    assert projected["recovery"]["sha256"] == digest
+    reader = shlex.split(projected["command"])
+    result = run_cli(module, tmp_path, *options, *reader[reader.index("quota"):])
+    assert result.returncode == 0, result.stdout + result.stderr
+    page = json.loads(result.stdout)
+    assert page["ok"] and page["eof"] and json.loads(page["content"]) == original
+
+    # Routing cannot hide the saved reader's digest, identity or source errors.
+    for option, value, reason in [("--capture-sha256", "0" * 64, "digest mismatch"),
+        ("--goal-id", "other", "exact Goal, Agent and Turn"),
+        ("--agent-id", "other", "exact Goal, Agent and Turn"),
+        ("--turn-instance-id", "other", "exact Goal, Agent and Turn"),
+        ("--capture-file", str(tmp_path / "absent.json"), "original capture could not be read")]:
+        invalid = list(command)
+        invalid[invalid.index(option) + 1] = value
+        rejected = run_cli(module, tmp_path, *options, *invalid)
+        assert rejected.returncode == 1, rejected.stdout + rejected.stderr
+        payload = json.loads(rejected.stdout)
+        assert not payload["ok"] and reason in payload["reason"]
+    assert capture.read_bytes() == raw and not missing.exists()
+
+
+@pytest.mark.parametrize("module", ["loopx.entrypoint", "loopx.cli"])
+@pytest.mark.parametrize("mode", [None, "on"])
+def test_saved_projection_keeps_machine_configuration_route_conflicts(
+    tmp_path, conflicting_defaults, module, mode,
+):
+    capture = tmp_path / "original.json"
+    capture.write_text(json.dumps({"ok": True, "goal_id": "fixture",
+        "agent_identity": {"agent_id": "agent"},
+        "heartbeat_receipt": {"turn_instance_id": "original-turn"}}))
+    command = ["quota", "context-project", "--capture-file", str(capture),
+        "--capture-sha256", hashlib.sha256(capture.read_bytes()).hexdigest(),
+        "--goal-id", "fixture", "--agent-id", "agent", "--turn-instance-id", "original-turn",
+        "--codex-app", "--codex-context-dir", str(tmp_path / "native")]
+    if mode is not None:
+        command.extend(["--context-delivery", mode])
+    result = run_cli(module, tmp_path, *command)
+    assert result.returncode != 0
+    assert "Both default LoopX runtime roots contain state" in result.stdout + result.stderr
+
+
 @pytest.mark.parametrize("module", ["loopx.entrypoint", "loopx.cli"])
 def test_explicit_native_location_recovers_without_goal_routing(tmp_path, monkeypatch, conflicting_defaults, module):
     native = tmp_path / "native"

tests/test_cli_entrypoint.py

@@ -378,10 +378,21 @@ def test_steward_endpoint_vocabulary_is_read_without_the_chat_layer() -> None:
     steward_executor_endpoints,
 )
 
-assert steward_executor_endpoints() == frozenset({"codex", "dsh"})
-normalize_steward_executor_machine_defaults(
-    {"schema_version": "steward_executor_machine_defaults_v0", "executor_endpoint": "dsh"}
-)
+expected_kinds = {"codex": "individual", "dsh": "managed", "antigravity": "individual"}
+assert steward_executor_endpoints() == frozenset(expected_kinds)
+for endpoint in expected_kinds:
+    normalized = normalize_steward_executor_machine_defaults(
+        {"schema_version": "steward_executor_machine_defaults_v0", "executor_endpoint": endpoint}
+    )
+    assert normalized["executor_endpoint"] == endpoint
+try:
+    normalize_steward_executor_machine_defaults(
+        {"schema_version": "steward_executor_machine_defaults_v0", "executor_endpoint": "unknown"}
+    )
+except ValueError:
+    pass
+else:
+    raise AssertionError("an unknown executor endpoint must be rejected")
 for name in ("loopx.chat_manager", "loopx.chat_agent", "loopx.chat_store"):
     assert name not in sys.modules, name
 
@@ -392,7 +403,7 @@ def test_steward_endpoint_vocabulary_is_read_without_the_chat_layer() -> None:
     "MANAGER_EXECUTOR_KIND_INDIVIDUAL", "MANAGER_EXECUTOR_KIND_MANAGED",
 ):
     assert getattr(manager, name) is getattr(machine_defaults, name), name
-assert manager.MANAGER_ENDPOINT_KINDS == {"codex": "individual", "dsh": "managed"}
+assert manager.MANAGER_ENDPOINT_KINDS == expected_kinds
 """
 	completed = run_isolated_script(script)
 	assert completed.returncode == 0, completed.stderr