fix(context): preserve async reader handles in code-mode guidance

loopx/control_plane/turn_driver/codex_context.py

@@ -18,7 +18,7 @@
 from .decision_capture import (
     write_decision_capture, capture_context_projection, CONTEXT_CAPTURE_PAGE_GUIDANCE,
     capture_page_range, read_context_capture_file, CONTEXT_CAPTURE_NAVIGATION_GUIDANCE,
-    CONTEXT_CAPTURE_SELECTION_GUIDANCE,
+    CONTEXT_CAPTURE_SELECTION_GUIDANCE, CONTEXT_CAPTURE_CODE_MODE_GUIDANCE,
     context_capture_read_command,
 )
 from .native_context_profile import (
@@ -267,10 +267,10 @@ def codex_context_recovery_instructions(directory: str, *, goal_id: str, agent_i
         "--max-page-output-bytes", str(page_output_bytes)])
     return (
         _context_continuation_instructions(command) +
-        CONTEXT_CAPTURE_NAVIGATION_GUIDANCE +
+        CONTEXT_CAPTURE_NAVIGATION_GUIDANCE + CONTEXT_CAPTURE_CODE_MODE_GUIDANCE +
         ("The generated native configuration supports 64000-byte context-read outputs and a 20000-token history limit. "
          + _NATIVE_CONTEXT_OUTPUT_GUIDANCE +
-         "Read one continuation per tool call; in code-mode, forward successful stdout directly with text(result.output). "
+         "Read one continuation per tool call. "
          if page_output_bytes == NATIVE_CONTEXT_LARGE_OUTPUT_BYTES else "") +
         "For code-mode reads, size batches against both direct-result and retained-history output limits; "
         "max_output_tokens alone does not establish the latter. Read one output at a time with "

loopx/control_plane/turn_driver/decision_capture.py

@@ -207,10 +207,18 @@ def validate_handoff_socket(value: str) -> Path:
     return path
 
 
+CODE_MODE_EXEC_RESULT_FORWARDER = "text(result.exit_code === 0 ? result.output : result)"
+CONTEXT_CAPTURE_CODE_MODE_GUIDANCE = (
+    "In code-mode, forward exec_command results with `" + CODE_MODE_EXEC_RESULT_FORWARDER + "`. "
+    "Keep a running command's session_id and finish it with write_stdin; "
+    "for a yielded functions.exec cell_id use functions.wait. "
+    "Running output or empty stdout does not prove completion; preserve failures. "
+)
+
 _CONTEXT_CAPTURE_PAGE_START = (
     "Follow next_command through eof and combine content in byte-offset order. "
-    "In code-mode, forward successful exec stdout directly with text(result.output); "
-    "retain errors. The reader verifies the original capture SHA-256; keep its returned digest and byte cursors. "
+    + CONTEXT_CAPTURE_CODE_MODE_GUIDANCE +
+    "The reader verifies the original capture SHA-256; keep its returned digest and byte cursors. "
     "Do not hash or retype the packet. "
 )
 _CONTEXT_CAPTURE_PAGE_END = (
@@ -251,8 +259,8 @@ def validate_handoff_socket(value: str) -> Path:
 ) + _CONTEXT_CAPTURE_PAGE_END
 CONTEXT_CAPTURE_READ_GUIDANCE = (
     "Follow next_command through eof and combine content in byte-offset order. "
-    "In code-mode, forward successful exec stdout directly with text(result.output); "
-    "retain errors. Small batches can save model round trips; start with at most two default pages "
+    + CONTEXT_CAPTURE_CODE_MODE_GUIDANCE +
+    "Small batches can save model round trips; start with at most two default pages "
     "and inspect every output. A host may truncate inside valid JSON or omit a whole output. "
     "Check each model-visible content's UTF-8 byte length against next_offset (total_bytes at eof), "
     "complete coverage and the original SHA-256. If incomplete, reread the affected range "

tests/control_plane/codex_checkpoint_fixture.py

@@ -100,22 +100,27 @@ def native_process_continuation(request, identity, *, code_mode=False):
         for part in reversed(parts):
             if not isinstance(part, str):
                 continue
+            pending_session = None
             try:
                 wrapper = json.loads(part)
+                if (isinstance(wrapper, dict) and type(wrapper.get("session_id")) is int
+                        and wrapper.get("exit_code") is None):
+                    pending_session = wrapper["session_id"]
                 if isinstance(wrapper, dict) and isinstance(wrapper.get("output"), str):
                     part = wrapper["output"]
             except ValueError:
                 pass
             match = re.search(r"(?m)^Process running with session ID (\d+)\s*$", part)
-            if match:
-                options = {"session_id": int(match[1]), "chars": "", "yield_time_ms": 1000,
+            if match or pending_session is not None:
+                options = {"session_id": int(match[1]) if match else pending_session, "chars": "", "yield_time_ms": 1000,
                            "max_output_tokens": 30000}
                 base = {"id": f"fc_process_{identity}", "call_id": f"call_process_{identity}"}
                 if code_mode:
+                    from loopx.control_plane.turn_driver.decision_capture import CODE_MODE_EXEC_RESULT_FORWARDER
                     options["max_output_tokens"] = 22000
                     return {**base, "type": "custom_tool_call", "name": "exec",
-                            "input": '// @exec: {"max_output_tokens":24000}\ntext((await tools.write_stdin('
-                                + json.dumps(options) + ')).output);'}
+                            "input": '// @exec: {"max_output_tokens":24000}\nconst result=await tools.write_stdin('
+                                + json.dumps(options) + ');' + CODE_MODE_EXEC_RESULT_FORWARDER + ';'}
                 return {**base, "type": "function_call", "name": "write_stdin", "arguments": json.dumps(options)}
         return None  # Never borrow a session from an older result.
     return None

tests/control_plane/test_codex_context_lifecycle.py

@@ -184,12 +184,12 @@ def test_lifecycle_invalidates_and_recovers_current_capture_without_bodies(tmp_p
     if event_name == "SessionStart":
         body = result["hookSpecificOutput"]["additionalContext"]
         assert first["recovery"]["sha256"] in body and "quota context-read" in body
-        # The shared selection advice adds about 400 characters to recovery,
-        # avoiding mandatory discovery and duplicate/oversized read batches.
-        # Preserve identity/retention clauses; this is a synthetic regression
-        # allowance, not permission to raise any host output/context limit.
+        # Async handle/error preservation adds 212 characters to the shared
+        # read guide. Retain the complete identity/retention clauses; this
+        # synthetic allowance does not raise any host output/context limit.
         assert "max_output_tokens=22000" in body and "max_output_tokens=24000" in body
-        assert payload["requirements"] not in body and len(body) < 3100
+        assert "session_id" in body and "write_stdin" in body and "functions.wait" in body
+        assert payload["requirements"] not in body and len(body) < 3350
     else:
         assert result == {}
     _, second = project(tmp_path, "two")

tests/control_plane/test_context_output_forwarding.py

@@ -0,0 +1,25 @@
+"""Execute the shipped code-mode example against its result-state contract."""
+import json
+import shutil
+import subprocess
+
+import pytest
+
+from loopx.control_plane.turn_driver.decision_capture import CODE_MODE_EXEC_RESULT_FORWARDER
+
+
+def test_code_mode_example_keeps_pending_handles_and_failed_output():
+    node = shutil.which("node")
+    if node is None:
+        pytest.skip("Node runtime is required")
+    results = [
+        {"exit_code": 0, "output": '{"ok":true,"content":"complete🙂"}'},
+        {"session_id": 41, "output": "", "chunk_id": "pending"},
+        {"exit_code": 19, "output": "capture digest mismatch", "chunk_id": "failure"},
+    ]
+    script = ("const results=JSON.parse(process.argv[1]);const visible=[];"
+              "const text=value=>visible.push(value);for(const result of results){"
+              + CODE_MODE_EXEC_RESULT_FORWARDER + ";}process.stdout.write(JSON.stringify(visible));")
+    completed = subprocess.run([node, "-e", script, json.dumps(results)],
+                               capture_output=True, text=True, check=True, timeout=10)
+    assert json.loads(completed.stdout) == [results[0]["output"], results[1], results[2]]

tests/control_plane/test_saved_context_project.py

@@ -26,7 +26,8 @@ def test_saved_projection_checks_original_before_any_publication(tmp_path, failu
         path.rename(source)
         path.symlink_to(source)
     args = ["quota", "context-project", "--capture-file", str(path), "--capture-sha256", digest,
-        "--goal-id", "saved-goal", "--agent-id", "saved-agent", "--turn-instance-id", "saved-turn"]
+        "--goal-id", "saved-goal", "--agent-id", "saved-agent", "--turn-instance-id", "saved-turn",
+        "--context-delivery", "off"]
     for name in ("digest", "goal", "agent", "turn"):
         if failure == name:
             option = {"digest": "--capture-sha256", "goal": "--goal-id", "agent": "--agent-id", "turn": "--turn-instance-id"}[name]
@@ -500,7 +501,7 @@ def consume(wire):
 
 
 @pytest.mark.skipif(not os.environ.get("LOOPX_NATIVE_CHECKPOINT_BIN"), reason="Explicit native Codex; loopback inference only")
-@pytest.mark.parametrize("host", ["app_server", "benchmark", "benchmark_progressive", "app_machine", "app_code_mode", "app_progressive", "app_progressive_large", "app_program_route", "app_code_mode_program_route"])
+@pytest.mark.parametrize("host", ["app_server", "benchmark", "benchmark_progressive", "app_machine", "app_code_mode", "app_code_mode_async", "app_progressive", "app_progressive_large", "app_program_route", "app_code_mode_program_route"])
 def test_native_saved_preflight_uses_lifecycle_without_guard_replay(tmp_path, monkeypatch, host):
     from canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime, read_fixture_todos
     from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection
@@ -521,6 +522,7 @@ def test_native_saved_preflight_uses_lifecycle_without_guard_replay(tmp_path, mo
     large_capture = host == "app_progressive_large"
     routed_program = host.endswith("_program_route")
     code_mode = host.startswith("app_code_mode")
+    force_async = host == "app_code_mode_async"
     if routed_program and os.name == "nt":
         pytest.skip("POSIX console alias and controlled PATH")
     unicode_requirement = "保留原始目标、范围与停止条件🙂\n" * 120
@@ -591,12 +593,21 @@ def test_native_saved_preflight_uses_lifecycle_without_guard_replay(tmp_path, mo
     rejected_read = False
     reading_initial = False
     initial_content = {}
+    read_gate = workspace / "reader-may-finish"
+    read_gate_armed = False
+    async_observed = False
 
     def response(request, number, message):
-        nonlocal phase, queue, actual_turn, receipts, rejected_read, reading_initial
+        nonlocal phase, queue, actual_turn, receipts, rejected_read, reading_initial, read_gate_armed, async_observed
         try:
             if continuation := native_process_continuation(request, f"projection_{number}",
                     code_mode=code_mode):
+                if read_gate_armed:
+                    # The child cannot finish before its handle reaches this
+                    # next real native request. No timing luck decides success.
+                    async_observed = True
+                    read_gate.touch()
+                    read_gate_armed = False
                 return continuation
             if reading_initial:
                 name = phase.removesuffix("_read")
@@ -744,14 +755,23 @@ def response(request, number, message):
                         return message
                     return response(request, number, message)
             arguments = {"cmd": command, "max_output_tokens": 22000}
+            if force_async and not async_observed and " quota context-read " in command:
+                wait_code = ("import pathlib,time; p=pathlib.Path(" + repr(str(read_gate)) + "); "
+                    "deadline=time.monotonic()+20\n"
+                    "while not p.exists() and time.monotonic()<deadline: time.sleep(.05)\n"
+                    "assert p.exists(), 'native output handle was not delivered'")
+                arguments["cmd"] = shlex.join([sys.executable, "-c", wait_code]) + " && " + command
+                arguments["yield_time_ms"] = 1000
+                read_gate_armed = True
             if routed_program:
                 # A login shell may replace the controlled PATH before it
                 # executes the otherwise unchanged generated reader command.
                 arguments["login"] = False
             if code_mode:
+                from loopx.control_plane.turn_driver.decision_capture import CODE_MODE_EXEC_RESULT_FORWARDER
                 return {"type": "custom_tool_call", "id": f"fc_{number}", "call_id": f"call_{number}",
-                    "name": "exec", "input": '// @exec: {"max_output_tokens":24000}\ntext((await tools.exec_command('
-                        + json.dumps(arguments) + ')).output);'}
+                    "name": "exec", "input": '// @exec: {"max_output_tokens":24000}\nconst result=await tools.exec_command('
+                        + json.dumps(arguments) + ');' + CODE_MODE_EXEC_RESULT_FORWARDER + ';'}
             # Match the generated native profile's 64000-byte page transport;
             # the old 3000-token tool cap truncated its valid JSON first page.
             return {"type": "function_call", "id": f"fc_{number}", "call_id": f"call_{number}",
@@ -808,6 +828,8 @@ def response(request, number, message):
                 config.write_text(config.read_text().replace("[features]\n", "[features]\ncode_mode=true\n"))
             _app_server(binary, env, workspace, home, True, sandbox="workspace-write")
     assert not errors and phase == "done", (errors, phase, wrong_runtime.exists())
+    if force_async:
+        assert async_observed and read_gate.is_file()
     if routed_program:
         assert set(initial_content) == {"cold"} and not wrong_runtime.exists()
     if benchmark:
@@ -920,7 +942,7 @@ def request(method, params):
     assert not list(home.rglob("*.pack")), "discovery must not clone a plugin catalog"
 
 
-def test_saved_projection_default_off_ignores_available_native_cache(tmp_path, monkeypatch):
+def test_saved_projection_explicit_off_ignores_available_native_cache(tmp_path, monkeypatch):
     from loopx.cli import build_parser
     from loopx.cli_commands.quota_capture import project_saved_context_capture
     from loopx.control_plane.turn_driver import codex_context
@@ -930,11 +952,12 @@ def test_saved_projection_default_off_ignores_available_native_cache(tmp_path, m
         "heartbeat_receipt": {"turn_instance_id": "t"}}).encode()
     path.write_bytes(raw)
     def unexpected(*args, **kwargs):
-        pytest.fail("default-off saved presentation must not inspect/create native state")
+        pytest.fail("explicit-off saved presentation must not inspect/create native state")
     monkeypatch.setattr(codex_context, "codex_context_directory", unexpected)
     args = build_parser().parse_args(["quota", "context-project", "--capture-file", str(path),
         "--capture-sha256", hashlib.sha256(raw).hexdigest(), "--goal-id", "g", "--agent-id", "a",
-        "--turn-instance-id", "t", "--codex-context-dir", str(root), "--decision-output-root", str(root)])
+        "--turn-instance-id", "t", "--context-delivery", "off",
+        "--codex-context-dir", str(root), "--decision-output-root", str(root)])
     result = project_saved_context_capture(args)
     assert result["ok"] and result["mode"] == "full" and result["path"] == str(path)
     assert not root.exists() and path.read_bytes() == raw