fix(context): recover full captures before native initialization

docs/reference/required-work-context.md

@@ -443,6 +443,16 @@ disable. Other benchmark modes remain off. Workers pin the resolved trial choice
 in preflight and solver subprocesses; existing saved/frozen trials must not be
 rewritten or reused to qualify the new default.
 
+If a supported declared Codex caller has not initialized native continuity yet
+(including a missing native session id),
+the guard still delivers its complete current capture through the file reader.
+It does not create a substitute session, journal, epoch or retained baseline;
+progressive selections remain available from that same current source. Native
+reuse starts only through the existing session/lifecycle contract. Missing
+session identity does not excuse unsafe context directories: their private-file
+checks still fail, preserving the original decision and receipt for recovery
+without another admission. Explicit off retains ordinary inline delivery.
+
 Machine consumers that parse `should_run`, selected work or settlement identities
 must explicitly request raw output on internal `quota should-run` calls using
 `--context-delivery off` or the subprocess `LOOPX_CONTEXT_DELIVERY=off` override,
@@ -789,6 +799,12 @@ Codex App SSH Goal（`--runtime-profile codex_app_ssh_goal`），以及现有 Co
 不证明宿主身份或上下文保留。
 当前原生连续性依赖 POSIX；其他平台显式拒绝开启。
 
+受支持的 Codex 声明尚未完成原生连续性初始化时（包括尚未提供会话 id），
+guard 仍通过文件 reader 交付本次完整捕获，
+可以从同一当前来源渐进选读；不会制造替代会话、journal、epoch 或已保留基线。
+原生复用仍须由既有会话与生命周期契约建立。缺少会话标识不豁免私有目录检查：
+不安全目录仍失败，保留原决策与回执用于恢复，不重新准入。显式关闭仍完整内联交付。
+
 机器继承与显式开启的入口共用已解析 machine runtime 下的私有 `context-delivery` 目录，并按原生
 会话隔离 journal。缺失末级目录以 `0700` 创建；不跟随末级符号链接、不修改不安全目录
 权限。显式 on 未指定路径时也自动选择目录，仍支持手工覆盖；handoff socket 保留

loopx/cli_commands/quota.py

@@ -18,7 +18,6 @@
     native_context_activation, model_context_reference,
 )
 from ..control_plane.turn_driver.decision_capture import (
-    context_capture_read_command,
     project_captured_context,
 )
 from ..usage_goal import observe_quota_result
@@ -410,26 +409,30 @@ def _emit_quota_result(
     if (not deferred_context and args.format == "json" and succeeded and capture_directory
             and native_context):
         import hashlib
-        from ..control_plane.turn_driver.codex_context import codex_context_directory, project_codex_context
+        from ..control_plane.turn_driver.codex_context import (
+            NativeContextSessionUnavailable, codex_context_directory, project_codex_context,
+        )
 
         try:
             capture_id = hashlib.sha256((capture_directory / "decision.json").read_bytes()).hexdigest()
-            directory = codex_context_directory(directory=args.codex_context_dir,
-                root=getattr(args, "codex_context_root", None), create=True)
+            try:
+                directory = codex_context_directory(directory=args.codex_context_dir,
+                    root=getattr(args, "codex_context_root", None), create=True)
+            except NativeContextSessionUnavailable:
+                # A declared host may not have initialized its native session.
+                # Deliver this complete capture without a session journal or
+                # baseline claim; do not repeat the already-committed guard.
+                directory = None
             payload = project_codex_context(payload, directory=directory,
                 capture_directory=capture_directory, goal_id=decision_identity[0], agent_id=decision_identity[1],
                 surface="turn_envelope" if getattr(args, "turn_envelope", False) else "quota_packet",
                 capture_id=capture_id, continuity_generation=getattr(args, "_context_delivery_generation", None))
             recovery = payload.get("recovery")
-            if recovery is None:
-                original = capture_directory / "decision.json"
-                recovery = {
-                    "path": str(original),
-                    "sha256": capture_id,
-                    "command": context_capture_read_command(original, capture_id),
-                }
+            # Without a prepared native projection, the view adapter publishes
+            # this complete CLI view and its reader commands as current recovery.
+            # The original authority decision remains separately captured.
             payload = native_context_capture_view(payload, capture_directory, context_directory=directory,
-                metadata={"capture_sha256": capture_id, "recovery": recovery},
+                metadata={"capture_sha256": capture_id, **({"recovery": recovery} if recovery is not None else {})},
                 max_page_output_bytes=getattr(args, "max_page_output_bytes", None))
             payload = model_context_reference(payload)
         except (QuotaDecisionTransportError, OSError, ValueError) as exc:

loopx/control_plane/turn_driver/codex_context.py

@@ -31,6 +31,10 @@
 CONTEXT_LIFECYCLE_HOOKS = ("SessionStart", "PreCompact", "PostCompact", "PostToolUse", "Interrupt")
 
 
+class NativeContextSessionUnavailable(ValueError):
+    """No native identity for a session journal; explicit file reads still work."""
+
+
 def context_delivery_for_arguments(args: Any, *, codex_machine_preference: bool = False) -> dict[str, Any]:
     """One declaration adapter for explicit CLI and inherited machine opt-ins."""
     fields = {key: getattr(args, key, None) for key in ("host_surface", "scheduler_owner", "execution_mode")}
@@ -149,7 +153,7 @@ def codex_context_directory(*, directory: str | None = None, root: str | None =
     parent = _private_root(root)
     session = session if session is not None else os.environ.get("CODEX_THREAD_ID")
     if not isinstance(session, str) or not session:
-        raise ValueError("native session context requires CODEX_THREAD_ID or a native hook session")
+        raise NativeContextSessionUnavailable("native session context requires CODEX_THREAD_ID or a native hook session")
     child = parent / hashlib.sha256(session.encode()).hexdigest()
     if create:
         child.mkdir(mode=0o700, exist_ok=True)

tests/control_plane/test_context_capture_registry_independence.py

@@ -129,6 +129,21 @@ def test_replan_stages_read_original_authority_capture_without_registry(
         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)
+        if guard.get("schema_version") == "context_capture_navigation_v1":
+            command, observed = guard["recovery"]["command"], bytearray()
+            while True:
+                argv = shlex.split(command)
+                result = run_cli("loopx.cli", fixture["project"], *argv[argv.index("quota"):])
+                assert result.returncode == 0, result.stdout + result.stderr
+                page = json.loads(result.stdout)
+                assert page["ok"] and page["offset"] == len(observed)
+                observed.extend(page["content"].encode("utf-8"))
+                if page["eof"]:
+                    assert page["next_offset"] is None and len(observed) == page["total_bytes"]
+                    break
+                assert page["next_offset"] == len(observed)
+                command = page["next_command"]
+            guard = json.loads(observed)
         missing = tmp_path / "unrelated-missing-registry.json"
 
         def snapshot():

tests/control_plane/test_context_machine_configuration.py

@@ -372,6 +372,57 @@ def test_explicit_on_first_use_uses_same_private_default_without_machine_setting
     assert not list(root.iterdir()), "setup does not manufacture retained-input evidence"
 
 
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+@pytest.mark.parametrize("native_id", [None, "uninitialized-native-session"])
+@pytest.mark.skipif(os.name != "posix", reason="Native captured delivery uses POSIX private directories")
+def test_default_guard_without_native_session_delivers_current_source_once(tmp_path, monkeypatch, provider, native_id):
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    monkeypatch.delenv("CODEX_THREAD_ID", raising=False)
+    if native_id is not None:
+        monkeypatch.setenv("CODEX_THREAD_ID", native_id)
+    monkeypatch.delenv("LOOPX_CONTEXT_DELIVERY", raising=False)
+    project, runtime, registry = _write_fixture(tmp_path)
+    _initialize_fixture_authority(project, runtime, provider=provider)
+    state = project / f".codex/goals/{GOAL_ID}/ACTIVE_GOAL_STATE.md"
+    original = state.read_bytes()
+    turn = "sessionless-original-turn"
+    code, wire = _run_cli(registry, runtime, "quota", "should-run", "--codex-app",
+        "--goal-id", GOAL_ID, "--agent-id", AGENT_ID, "--todo-id", TODO_ID,
+        "--turn-instance-id", turn, cwd=project, raw_guard=False)
+    assert code == 0 and wire["mode"] == "full", wire
+    assert "baseline_reference" not in wire
+    code, page = _run_cli(registry, runtime, "quota", "context-read",
+        "--reference-file", wire["recovery"]["reference_file"],
+        "--select", "/goal_id", "--select", "/heartbeat_receipt",
+        "--select", "/interaction_contract/cli_channel", "--page-bytes", "64000",
+        "--max-page-output-bytes", "64000", cwd=project)
+    assert code == 0 and page["ok"] and page["eof"], page
+    values = {row["pointer"]: row["value"] for row in json.loads(page["content"])["selections"]}
+    assert values["/goal_id"] == GOAL_ID
+    assert values["/heartbeat_receipt"]["turn_instance_id"] == turn
+    assert values["/interaction_contract/cli_channel"]["settlement_plan"]
+    assert _heartbeat_receipt_count(runtime, turn) == 1 and _spend_run_count(runtime) == 0
+    assert state.read_bytes() == original
+    assert not list((runtime / "context-delivery").rglob("session.json"))
+
+
+@pytest.mark.skipif(os.name != "posix", reason="Native captured delivery checks POSIX private directories")
+def test_sessionless_fallback_does_not_hide_an_unsafe_context_root(tmp_path, monkeypatch):
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    monkeypatch.delenv("CODEX_THREAD_ID", raising=False)
+    project, runtime, registry = _write_fixture(tmp_path, provider="sqlite")
+    unsafe = tmp_path / "unsafe-context"
+    unsafe.mkdir()
+    unsafe.chmod(0o755)
+    turn = "unsafe-sessionless-turn"
+    code, failure = _run_cli(registry, runtime, "quota", "should-run", "--codex-app",
+        "--goal-id", GOAL_ID, "--agent-id", AGENT_ID, "--todo-id", TODO_ID,
+        "--turn-instance-id", turn, "--codex-context-root", str(unsafe), cwd=project, raw_guard=False)
+    assert code == 1 and failure["error_code"] == "quota_decision_transport_failed", failure
+    assert _heartbeat_receipt_count(runtime, turn) == 1 and _spend_run_count(runtime) == 0
+    assert not list(unsafe.iterdir())
+
+
 def test_explicit_on_unknown_host_fails_before_first_use_directory_creation(tmp_path):
     registry, runtime = tmp_path / "absent.json", tmp_path / "runtime"
     code, result = _run_cli(registry, runtime, "quota", "context-config", "--context-delivery", "on",