diff --git a/loopx/control_plane/turn_driver/codex_context.py b/loopx/control_plane/turn_driver/codex_context.py index 6773998caa..2a286a174d 100644 --- a/loopx/control_plane/turn_driver/codex_context.py +++ b/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 " diff --git a/loopx/control_plane/turn_driver/decision_capture.py b/loopx/control_plane/turn_driver/decision_capture.py index 1521219608..d15ef1df34 100644 --- a/loopx/control_plane/turn_driver/decision_capture.py +++ b/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 " diff --git a/tests/control_plane/codex_checkpoint_fixture.py b/tests/control_plane/codex_checkpoint_fixture.py index c852b5d7ef..9ae3c197c0 100644 --- a/tests/control_plane/codex_checkpoint_fixture.py +++ b/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 diff --git a/tests/control_plane/test_codex_context_lifecycle.py b/tests/control_plane/test_codex_context_lifecycle.py index 4f6e4d1f27..dc63f08b0f 100644 --- a/tests/control_plane/test_codex_context_lifecycle.py +++ b/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") diff --git a/tests/control_plane/test_context_output_forwarding.py b/tests/control_plane/test_context_output_forwarding.py new file mode 100644 index 0000000000..e2cecfb0f1 --- /dev/null +++ b/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]] diff --git a/tests/control_plane/test_saved_context_project.py b/tests/control_plane/test_saved_context_project.py index 3d1cc25ed1..d00c414eef 100644 --- a/tests/control_plane/test_saved_context_project.py +++ b/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()