From 6a55306bcdc1086e1f07fe4e195c88b17a23f8f9 Mon Sep 17 00:00:00 2001 From: YEONGUK LEE Date: Thu, 1 Oct 2026 09:27:46 +0900 Subject: [PATCH 1/3] feat(experiment): validate opt-in Codex instant steering --- experiments/codex_steering/README.md | 90 ++++++ experiments/codex_steering/RESULTS.md | 131 ++++++++ experiments/codex_steering/native_fixture.py | 320 +++++++++++++++++++ experiments/codex_steering/steer.py | 250 +++++++++++++++ experiments/codex_steering/test_steer.py | 162 ++++++++++ 5 files changed, 953 insertions(+) create mode 100644 experiments/codex_steering/README.md create mode 100644 experiments/codex_steering/RESULTS.md create mode 100644 experiments/codex_steering/native_fixture.py create mode 100644 experiments/codex_steering/steer.py create mode 100644 experiments/codex_steering/test_steer.py diff --git a/experiments/codex_steering/README.md b/experiments/codex_steering/README.md new file mode 100644 index 00000000..5c3f2c9a --- /dev/null +++ b/experiments/codex_steering/README.md @@ -0,0 +1,90 @@ +# Opt-in Codex steering experiment (#247) + +This is an **unreleased local experiment**, not a new installed CLI mode. +Existing package rules exclude `experiments/`. Canonical queue/send/wake, +generated runtime, Relay/SSH, and the Side Session experiment (#245/#246) +are unchanged. + +## Technical review + +Codex 0.159.0 introduced default-off `instant_interrupt`, still under development +in 0.159.2. It makes current-turn user input preempt sampling and yield +foreground code-mode observations. It does not terminate all external work, +undo earlier side effects, or convert durable queue submissions into steering. + +Native queued-item dispatch uses `start_turn_if_idle`. The prototype instead +connects to the **existing owner** via private Unix-domain WebSocket and calls +`turn/steer` with exact `threadId` and nonempty `expectedTurnId`. It never uses +`turn/start`, `turn/interrupt`, resume, fallback, auto-approval or resend. + +Operator opt-in and exact server/home/target are required. The owning thread's +feature snapshot is read using `experimentalFeature/list(threadId=...)`; +`--require-instant` refuses disabled/unknown snapshots. This read is not a +promise that every pending step reflects a recently changed feature or that +latency meets a deadline. The operator must configure the destination; the +sender never enables the feature. Target permission overrides are not accepted. + +Dry-run verifies metadata but does **not** claim native validation of the +expected turn ID; that atomic guard is enforced during actual `turn/steer`. +Accepted is not consumed/ACK/completed. RPC uncertainty is never auto-retried. + +## Pre-registered native experiment + +Actual Codex 0.159.2 executes against a **loopback synthetic Responses API**. +No real account credentials or model calls are used. No existing session or +home is loaded. The first response emits an initial text delta then waits at +a controlled gate. Input is submitted only after that delta is observed. + +| Mode | `instant_interrupt` | Expected second model request before first gate release | +| --- | --- | --- | +| queue | OFF | No | +| queue | ON | No | +| steer | OFF | No | +| steer | ON | Yes | + +ON steering must produce the second request within 5 seconds while the first +gate stays closed; other cases must not produce it during a 0.75-second window. +After gate release, all four cases must include the new input in the next +request and finish. Steering must retain the same turn; queue must start a new +turn. These are deterministic synthetic protocol observations, not model ACKs +or a real-provider latency SLA. Each case has a 45-second deadline, first +scenario assertion failure stops the batch. Failed gates are findings, not +redefined expectations. Fixture/setup defects may be corrected separately. + +After the first four-case batch passed, a three-repeat confirmation batch was +specified with unchanged ordering windows and new wrong-home/missing-target +negative controls, before executing that confirmation. Native scenario results +and setup failures are recorded separately in `RESULTS.md`. + +The version-gated native harness covers SSE response preemption. Actual +code-mode cell yielding, real model/SSH/Windows/remote listeners, simultaneous +senders, and hosted permission/allowlist integration remain follow-up gates. + +## Usage (dummy values) + +Requires Python 3.11+ and websockets 17.1 for this experiment only. + +```sh +python3 experiments/codex_steering/steer.py --experimental-steering \ + --socket /private-test/app.sock --expected-home /private-test/home \ + --to codex:00000000-0000-4000-8000-000000000001 \ + --expected-turn-id turn-example --require-instant --dry-run \ + --message 'Change the review focus' +python3 -m unittest discover -s experiments/codex_steering -v +``` + +Native fixture execution is a separate opt-in command documented with its +results. It never silently changes a feature/config in an existing daemon. + +```sh +python3 experiments/codex_steering/native_fixture.py --run-native-fixture \ + --temp-root /approved-private-scratch --repetitions 3 +``` + +References (reviewed 2026-10-01): +[changelog](https://learn.chatgpt.com/docs/changelog), +[App Server](https://learn.chatgpt.com/docs/app-server). +Pinned source: openai/codex ff6aec96948b70d94983af2641a6b67c94faeff5 +(`rust-v0.159.2`), particularly `features/src/lib.rs`, +`core/src/session/input_queue.rs`, `core/src/session/turn.rs`, +`ext/queue/src/service.rs`, and `app-server/.../turn_processor.rs`. diff --git a/experiments/codex_steering/RESULTS.md b/experiments/codex_steering/RESULTS.md new file mode 100644 index 00000000..6bf085d1 --- /dev/null +++ b/experiments/codex_steering/RESULTS.md @@ -0,0 +1,131 @@ +# Codex steering / instant_interrupt — 2026-10-01 + +## Verdict + +**Experimental native protocol feasibility: PASS. General-use CLI support is +not implemented or accepted.** Issue #247 remains open. + +Queue-only default delivery must remain unchanged. An explicit native +`turn/steer` path is technically viable and the default-off feature changes +how promptly an active turn handles that input. No existing session was steered +or interrupted, and no installed package, permissions, global feature config, +hosted service or release was changed. + +## Scope and source-grounded findings + +- Base: session-peer 90489fccc87fe2588a2546aaaf7f285429d2bcfd. +- Codex CLI/server 0.159.2; source ff6aec96948b70d94983af2641a6b67c94faeff5. +- macOS arm64; Python **3.13.7**, websockets **17.1**. +- Native feature list reports `instant_interrupt` under development, default off. + A command-local `--enable` override is not a persistent config change. +- Native queue service calls `start_turn_if_idle`; it does not steer busy turns. +- `turn/steer` requires the exact active `expectedTurnId`; no separate user turn + is created. It is not `turn/interrupt`, which cancels a turn. +- The prototype connects to an operator-selected existing private Unix socket, + checks advertised native version, exact home, loaded target, active status, + direct-input capability and a thread-specific feature snapshot. The native + expected-turn guard handles the final submission, not the dry-run. +- Known RPC refusal vs post-attempt uncertainty are separate; neither is retried + or silently converted into queue/start/resume/interrupt. Approval requests + are not answered. Message input has a 64 KiB UTF-8 cap. + +## Actual native test, synthetic provider + +This used **actual Codex processes**, native WebSocket-over-Unix JSON-RPC, +thread/turn lifecycle and SSE parsing. A local synthetic Responses API emits +fixed text and holds the first streaming response at a gate. It does not invoke +an OpenAI model and its final text is **not an agent-generated ACK**. + +Each case had a newly generated private Codex home/cwd and loopback provider, +with no real keys/login files copied or supplied. Child environment was +allowlisted; existing credential-related environment variables were not inherited. +Thread permissions were read-only/on-request. The owner configured the feature +before starting the test thread; the adapter never enabled it. + +Inputs were submitted after observing the first native assistant text delta. +Only input-marker booleans were retained from provider requests; original +request bodies, histories, home paths, IDs and credentials are not in this report. + +### Pre-registered ordering expectations + +For ON steering, the replacement request must arrive before first-stream gate +release within 5 seconds. The other three cases must not issue that request +during a 0.75-second observation window; then the original gate is released. +Queue must use a new turn; steer must keep the initial turn. Each case has a +45-second hard deadline. First scenario assertion failure stops the batch. + +An initial four-case native batch passed. Before the confirmation, three +repetitions and additional wrong-home/missing-target controls were specified; +the ordering thresholds/expectations did not change. Final confirmation: + +| Mode | Feature | Second request before release | Same turn | n / PASS | Second request after input, ms min–max | +| --- | --- | --- | --- | --- | --- | +| queue | OFF | No | No | 3 / 3 | 809.70–815.18 | +| queue | ON | No | No | 3 / 3 | 792.40–808.92 | +| steer | OFF | No | Yes | 3 / 3 | 780.21–785.17 | +| steer | ON | Yes | Yes | 3 / 3 | 5.20–5.96 | + +**12/12 final native cases passed.** Each case made exactly two synthetic model +requests, with the new input present only in the second. Rejected control input +never appeared. Wrong expected turn (-32600), mismatched home, missing target, +and completed/idle-target steering were refused. Disabled-feature refusal for +`--require-instant` was exercised in OFF cases. No input was auto-retried. + +The approximately 0.8-second OFF/queue timings include the deliberately imposed +0.75-second gate. They are not typical Codex latency measurements. The ON +5.20–5.96 ms result is a localhost ordering observation, not a real-provider SLA +or guarantee that running tools stop. + +## Setup failures, not product findings + +Early fixture initialization failed before any scenario input because the +native advertised version prefix depends on the client originator. The first +attempt's teardown also hit macOS EPERM for a zombie-only process group, +skipping `wait()` and producing a subprocess ResourceWarning. Read-only PID +checks subsequently showed that process had exited. These were harness defects, +not product preemption failures. + +The accepted known version identifiers were corrected, and teardown now only +tolerates EPERM after verifying every group member is a zombie, then reaps its +leader. Provider cleanup executes independently in `finally`. Final 12-case +execution under `-X dev -W error::ResourceWarning` emitted no warnings/errors. +Expectations were not changed to turn a scenario failure into a pass. + +## Other checks and cleanup + +- 24 scoped unit tests pass: opt-in, dry-run truth, disabled/unknown feature, + identity/capability/approval guards, input limits, wrong turn, uncertain + responses, socket access and pagination bounds. +- Generator `--check` and diff whitespace check pass; runtime diff is zero. +- Artifact-content tests were invoked but **all 3 skipped** because artifacts + were not supplied. The local environment lacks `build`; no dependencies were + installed just for this experiment. `pyproject.toml` statically excludes the + experiment from wheel/sdist; actual artifact verification is left to CI. +- Each final native case reaped its owned process group and verified no live + group members; temporary fixture homes were removed. Known test-created + native socket/lock paths were cleaned by exact path, never their shared parent. +- Final checks found no test home directories or Codex cwd/command matching + the fixture prefix. The unused worktree shell was closed. Other worktrees, + existing daemons, sessions and account files were preserved. + +## Remaining gates before a supported delivery mode + +1. Real-provider ON/OFF behavior, real code-mode cell/command yielding and + remaining external-work semantics; these were reviewed in source, not executed. +2. Safe owner endpoint registration/generation and authorization for existing + app-hosted sessions; a thread-specific config snapshot is not a guarantee + of a step's effective preemption settings after a mid-turn config change. +3. Explicit steering capability/policy, sender/untrusted-message framing and + return-route/ACK behavior. The prototype only forwards operator-authorized + text and does not provide the canonical envelope or generic completion wait. +4. Genuine turn-replacement interleavings, disconnect/timeout after acceptance, + SSH, Linux/Windows, MCP/paired exposure and multi-client approval behavior. +5. Stable CLI spelling/schema, regression and independent review. The prototype + remains excluded from distributions; this is not release/merge approval. + +Unlike `turn/start`, which can implicitly steer on an idle-check race, native +`turn/steer` provides an explicit expected-turn guard. It does **not** solve +the `/btw` UI-open vs server-loaded lifetime issue in #245 / Draft #246. + +References: [official changelog](https://learn.chatgpt.com/docs/changelog), +[App Server](https://learn.chatgpt.com/docs/app-server). diff --git a/experiments/codex_steering/native_fixture.py b/experiments/codex_steering/native_fixture.py new file mode 100644 index 00000000..96ddb6ed --- /dev/null +++ b/experiments/codex_steering/native_fixture.py @@ -0,0 +1,320 @@ +"""Actual native Codex + loopback synthetic SSE provider; no account credentials. + +First response gates prove ordering, not real-model latency/ACK. Stops on first +assertion failure. Fixture-created homes/sockets are temporary and private. +""" +import argparse +import contextlib +import http.server +import json +import os +import shutil +import signal +import subprocess +import sys +import tempfile +import threading +import time +from pathlib import Path + +from steer import Rpc, Refused, Rejected, pages, steer + + +class Provider: + def __init__(self): + self.release = threading.Event() + self.second = threading.Event() + self.lock = threading.Lock() + self.requests = [] + self.second_at = None + self.started_at = time.monotonic() + self.unexpected_paths = 0 + provider = self + + class Handler(http.server.BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + + def log_message(self, *args): + pass + + def do_GET(self): + self.send_response(404) + self.send_header("Content-Length", "0") + self.end_headers() + + def do_POST(self): + if self.path != "/v1/responses": + provider.unexpected_paths += 1 + self.send_error(404) + return + n = int(self.headers.get("Content-Length", "0")) + assert 0 < n < 4 * 1024 * 1024 + data = self.rfile.read(n) + if self.headers.get("Content-Encoding") == "zstd": + raise RuntimeError("fixture needs compression disabled") + payload = json.loads(data) + # Only synthetic marker booleans are retained; no request dump. + rendered = json.dumps(payload.get("input", [])) + with provider.lock: + number = len(provider.requests) + 1 + provider.requests.append({"newInput": "UPDATE-STEER247" in rendered, + "rejectedInput": "MUST-NOT-ACCEPT-247" in rendered}) + if number == 2: + provider.second_at = time.monotonic() + provider.second.set() + self.send_response(200) + self.send_header("Content-Type", "text/event-stream") + self.send_header("Connection", "close") + self.end_headers() + self.close_connection = True + + def emit(value): + self.wfile.write(("data: " + json.dumps(value) + "\n\n").encode()) + self.wfile.flush() + + try: + emit({"type": "response.created", "response": {"id": f"resp-{number}"}}) + if number == 1: + emit({"type": "response.output_item.added", "item": { + "type": "message", "role": "assistant", "id": "initial-msg", + "content": [{"type": "output_text", "text": ""}]}}) + emit({"type": "response.output_text.delta", "delta": "ORIGINAL-PREFIX"}) + if not provider.release.wait(40): + return + message = "ORIGINAL-FINISHED" if number == 1 else "SYNTHETIC-UPDATED-RESPONSE" + emit({"type": "response.output_item.done", "item": { + "type": "message", "role": "assistant", "id": f"msg-{number}", + "content": [{"type": "output_text", "text": message}]}}) + emit({"type": "response.completed", "response": { + "id": f"resp-{number}", "usage": { + "input_tokens": 1, "input_tokens_details": None, + "output_tokens": 1, "output_tokens_details": None, "total_tokens": 2}}}) + except (BrokenPipeError, ConnectionResetError): + pass # Expected when ON steering preempts the initial stream. + + self.server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler) + self.server.daemon_threads = True + self.thread = threading.Thread(target=self.server.serve_forever, daemon=True) + self.thread.start() + + def close(self): + self.release.set() + self.server.shutdown() + self.server.server_close() + self.thread.join(timeout=2) + assert not self.thread.is_alive() + + +def kill_owned(process): + # Keep the leader PID unreaped through group cleanup, never reuse an old pgid. + for sig, grace in ((signal.SIGTERM, 0.2), (signal.SIGKILL, 0)): + try: + os.killpg(process.pid, sig) + except ProcessLookupError: + pass + except PermissionError: + # macOS reports EPERM for groups containing only zombies. Do not + # ignore denial while any member is still running. + members = subprocess.run(["ps", "-o", "stat=", "-g", str(process.pid)], + capture_output=True, text=True, timeout=3) + if sys.platform != "darwin" or members.returncode not in (0, 1) or any( + line.strip() and not line.strip().startswith("Z") for line in members.stdout.splitlines()): + raise + if grace: + time.sleep(grace) + process.wait(timeout=5) + check = subprocess.run(["ps", "-o", "stat=", "-g", str(process.pid)], + capture_output=True, text=True, timeout=3) + assert check.returncode in (0, 1) + assert not any(line.strip() and not line.strip().startswith("Z") for line in check.stdout.splitlines()) + + +def run_case(executable, root, mode, enabled): + t0 = time.monotonic() + case = root / (mode + ("-on" if enabled else "-off")) + case.mkdir(mode=0o700) + home, cwd = case / "home", case / "cwd" + home.mkdir(mode=0o700) + cwd.mkdir(mode=0o700) + sock = case / "app.sock" + provider = Provider() + port = provider.server.server_address[1] + # This configuration is generated fixture state. It contains no credentials. + config = f'''model = "gpt-6.1-sol" +model_provider = "local_fixture" +sandbox_mode = "read-only" +approval_policy = "on-request" +[analytics] +enabled = false +[features] +instant_interrupt = {str(enabled).lower()} +enable_request_compression = false +[model_providers.local_fixture] +name = "Loopback synthetic fixture" +base_url = "http://127.0.0.1:{port}/v1" +wire_api = "responses" +requires_openai_auth = false +supports_websockets = false +''' + (home / "config.toml").write_text(config) + (home / "config.toml").chmod(0o600) + env = {k: v for k, v in os.environ.items() if k in ("PATH", "HOME", "TMPDIR", "LANG", "LC_ALL")} + env["CODEX_HOME"] = str(home) + process = None + physical = None + deadline_hit = threading.Event() + watchdog_done = threading.Event() + + def watchdog(): + if not watchdog_done.wait(45) and process is not None: + deadline_hit.set() + with contextlib.suppress(ProcessLookupError): + os.killpg(process.pid, signal.SIGTERM) + + watcher = threading.Thread(target=watchdog, daemon=True) + watcher.start() + try: + process = subprocess.Popen([executable, "app-server", "--listen", "unix://" + str(sock)], + cwd=cwd, env=env, stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, + start_new_session=True) + deadline = t0 + 15 + while not sock.exists(): + if time.monotonic() > deadline: + raise RuntimeError("fixture_native_startup_deadline") + time.sleep(0.05) + physical = sock.resolve() + observer = Rpc(sock, timeout=10) + with observer: + target = observer.call("thread/start", { + "cwd": str(cwd), "model": "gpt-6.1-sol", "sandbox": "read-only", + "approvalPolicy": "on-request", "ephemeral": False, + })["thread"]["id"] + turn = observer.call("turn/start", {"threadId": target, + "input": [{"type": "text", "text": "INITIAL-STEER247 synthetic protocol scenario"}]})["turn"]["id"] + observer.wait_event(lambda e: e.get("method") == "item/agentMessage/delta" + and "ORIGINAL-PREFIX" in e.get("params", {}).get("delta", "")) + with Rpc(sock, timeout=10) as sender: + flag_rows = list(pages(sender, "experimentalFeature/list", {"threadId": target})) + flag = next(f["enabled"] for f in flag_rows if f["name"] == "instant_interrupt") + assert flag is enabled, "feature_snapshot_mismatch" + for target_id, home_path in ((target, str(home) + "-wrong"), + ("00000000-0000-4000-8000-000000000001", str(home))): + try: + steer(sender, target_id, turn, "MUST-NOT-ACCEPT-247", + expected_home=home_path, opt_in=True) + except Refused: + pass + else: + raise AssertionError("identity_guard_failed") + if not enabled: + try: + steer(sender, target, turn, "MUST-NOT-ACCEPT-247", expected_home=str(home), + opt_in=True, require_instant=True) + except Refused: + pass + else: + raise AssertionError("disabled_instant_accepted") + # Wrong expected turn must reject before any input; no retry. + try: + sender.call("turn/steer", {"threadId": target, "expectedTurnId": "wrong-turn-id", + "input": [{"type": "text", "text": "MUST-NOT-ACCEPT-247"}]}) + except Rejected as exc: + assert exc.code == -32600 + else: + raise AssertionError("wrong_turn_accepted") + submitted_at = time.monotonic() + if mode == "steer": + result = steer(sender, target, turn, "UPDATE-STEER247 synthetic direction change", + expected_home=str(home), opt_in=True, require_instant=enabled) + assert result["submitted"] and not result["consumptionConfirmed"] + assert result["turnId"] == turn + else: + sender.call("thread/queue/add", {"threadId": target, + "clientUserMessageId": str(__import__("uuid").uuid4()), + "input": [{"type": "text", "text": "UPDATE-STEER247 synthetic queued follow-up"}]}) + before_release = provider.second.wait(5 if mode == "steer" and enabled else 0.75) + expected = mode == "steer" and enabled + assert before_release is expected, "preemption_gate_mismatch" + provider.release.set() + assert provider.second.wait(10), "updated_request_missing" + observer.wait_event(lambda e: e.get("method") == "item/completed" + and e.get("params", {}).get("item", {}).get("text") == "SYNTHETIC-UPDATED-RESPONSE", timeout=10) + done = observer.wait_event(lambda e: e.get("method") == "turn/completed" + and e.get("params", {}).get("turn", {}).get("status") == "completed" + and (mode == "queue" or e.get("params", {}).get("turn", {}).get("id") == turn), timeout=10) + completion_turn = done["params"]["turn"]["id"] + if mode == "queue" and completion_turn == turn: + done = observer.wait_event(lambda e: e.get("method") == "turn/completed" + and e.get("params", {}).get("turn", {}).get("id") != turn, timeout=10) + completion_turn = done["params"]["turn"]["id"] + assert (completion_turn == turn) is (mode == "steer") + assert provider.requests == [{"newInput": False, "rejectedInput": False}, + {"newInput": True, "rejectedInput": False}] + assert provider.unexpected_paths == 0 + # Completed turn refuses steering; there is no queued fallback. + try: + steer(sender, target, completion_turn, "DO-NOT-SEND-AFTER-COMPLETION", + expected_home=str(home), opt_in=True) + except Refused: + pass + else: + raise AssertionError("idle_steer_accepted") + assert len(provider.requests) == 2 + result = {"mode": mode, "instantInterrupt": enabled, "pass": True, + "secondRequestBeforeRelease": before_release, + "secondRequestAfterSubmitMs": round((provider.second_at - submitted_at) * 1000, 2), + "sameTurn": completion_turn == turn, "modelRequests": len(provider.requests), + "syntheticResponseObserved": True, "wrongTurnRejected": True, + "idleRejected": True, "homeMismatchRejected": True, + "missingTargetRejected": True, "elapsedMs": round((time.monotonic() - t0) * 1000, 2)} + return result + except Refused as exc: + if str(exc) == "unsupported_server_version" and hasattr(observer, "info"): + brand = observer.info.get("userAgent", "").split(" ", 1)[0] + print(json.dumps({"fixtureSetupFailed": True, "nativeBrand": brand}), flush=True) + raise + finally: + provider.release.set() + watchdog_done.set() + watcher.join(timeout=2) + try: + if process is not None: + kill_owned(process) + finally: + provider.close() + if physical is not None: + # Exact unique test-created socket/lock, never its shared parent. + for path in (physical, Path(str(physical) + ".lock")): + if path.exists() and path.stat().st_uid == os.getuid(): + path.unlink() + if sock.is_symlink() or sock.exists(): + sock.unlink() + assert not deadline_hit.is_set(), "batch_deadline_hit" + + +def main(): + p = argparse.ArgumentParser(description=__doc__) + p.add_argument("--run-native-fixture", action="store_true", required=True) + p.add_argument("--temp-root", required=True) + p.add_argument("--repetitions", type=int, default=1, choices=range(1, 6)) + args = p.parse_args() + executable = shutil.which("codex") + version = subprocess.check_output([executable, "--version"], text=True).strip() + if version != "codex-cli 0.159.2": + raise RuntimeError("unsupported_fixture_version") + # tempfile is strictly within the operator-selected approved scratch root. + with tempfile.TemporaryDirectory(prefix="codex-steer247-", dir=args.temp_root) as tmp: + root = Path(tmp) + for repetition in range(1, args.repetitions + 1): + repeat_root = root / str(repetition) + repeat_root.mkdir(mode=0o700) + for mode, enabled in (("queue", False), ("queue", True), ("steer", False), ("steer", True)): + print(json.dumps(dict(run_case(executable, repeat_root, mode, enabled), + repetition=repetition)), flush=True) + print(json.dumps({"complete": True, "nativeVersion": version, "cases": 4 * args.repetitions, + "temporaryHomesRemoved": True, "credentialsUsed": False}), flush=True) + + +if __name__ == "__main__": + main() diff --git a/experiments/codex_steering/steer.py b/experiments/codex_steering/steer.py new file mode 100644 index 00000000..d92c37a9 --- /dev/null +++ b/experiments/codex_steering/steer.py @@ -0,0 +1,250 @@ +"""Opt-in local steering prototype (#247), excluded from installed packages. + +This connects to an existing owner; it never starts/resumes a destination, +enables a feature, cancels a turn, answers approvals, queues or retries input. +""" +import argparse +import json +import os +import re +import stat +import sys +import time +import uuid +from pathlib import Path + + +class Refused(Exception): + pass + + +class Unknown(Exception): + pass + + +class Rejected(Refused): + def __init__(self, method, code): + super().__init__("rpc_rejected:" + method) + self.code = code + + +class Rpc: + LIMIT = 1024 * 1024 + + def __init__(self, path, timeout=10): + self.path, self.timeout = Path(path), timeout + self.sequence = 0 + self.events = [] + + def __enter__(self): + if not self.path.is_absolute(): + raise Refused("absolute_socket_required") + original = self.path.lstat() + resolved = self.path.resolve(strict=True) + target = resolved.lstat() + if original.st_uid != os.getuid() or target.st_uid != os.getuid() or not stat.S_ISSOCK(target.st_mode): + raise Refused("owned_socket_required") + if target.st_mode & 0o077: + raise Refused("private_socket_required") + for parent in (self.path.parent, resolved.parent): + value = parent.stat() + if value.st_uid != os.getuid() or value.st_mode & 0o077: + raise Refused("private_directory_required") + try: + from websockets.sync.client import unix_connect + except ImportError as exc: + raise Refused("websockets_dependency_missing") from exc + self.connection = unix_connect(str(resolved), open_timeout=self.timeout, + close_timeout=1, max_size=self.LIMIT) + try: + self.socket = self.connection.__enter__() + except Exception as exc: + raise Refused("connection_failed") from exc + try: + self.info = self.call("initialize", { + "clientInfo": {"name": "session_peer_steer_experiment", "version": "0.0.0"}, + "capabilities": {"experimentalApi": True}, + }) + if not re.match(r"(?:codex-tui|codex_cli_rs|codex-cli|session_peer_steer_experiment)/0\.159\.2(?:\s|$)", self.info.get("userAgent", "")): + raise Refused("unsupported_server_version") + self.write({"method": "initialized"}) + return self + except BaseException: + self.__exit__(None, None, None) + raise + + def __exit__(self, *args): + self.connection.__exit__(*args) + + def write(self, value): + frame = json.dumps(value, ensure_ascii=False) + if len(frame.encode("utf-8")) > self.LIMIT: + raise Refused("frame_too_large") + try: + self.socket.send(frame) + except Exception as exc: + raise ConnectionError("transport_closed") from exc + + def read(self, deadline): + remaining = deadline - time.monotonic() + if remaining <= 0: + raise TimeoutError() + try: + frame = self.socket.recv(timeout=remaining) + except TimeoutError: + raise + except Exception as exc: + raise ConnectionError("transport_closed") from exc + if not isinstance(frame, str): + raise Refused("invalid_frame") + value = json.loads(frame) + if not isinstance(value, dict): + raise Refused("invalid_rpc_response") + return value + + def call(self, method, params): + self.sequence += 1 + request_id = self.sequence + self.write({"id": request_id, "method": method, "params": params}) + deadline = time.monotonic() + self.timeout + while True: + value = self.read(deadline) + if value.get("id") == request_id: + if "error" in value: + raise Rejected(method, value["error"].get("code")) + if not isinstance(value.get("result"), dict): + raise Refused("invalid_rpc_response") + return value["result"] + if "id" in value: + raise Refused("server_interaction_required") + if len(self.events) >= 2000: + raise Refused("event_limit") + self.events.append(value) + + def wait_event(self, predicate, timeout=10): + deadline = time.monotonic() + timeout + while True: + for index, event in enumerate(self.events): + if predicate(event): + return self.events.pop(index) + event = self.read(deadline) + if "id" in event: + raise Refused("server_interaction_required") + if predicate(event): + return event + if len(self.events) >= 2000: + raise Refused("event_limit") + self.events.append(event) + + +def pages(rpc, method, params): + cursor = None + for _ in range(32): + query = dict(params, limit=100) + if cursor is not None: + query["cursor"] = cursor + result = rpc.call(method, query) + if not isinstance(result.get("data"), list): + raise Refused("invalid_inventory") + yield from result["data"] + next_cursor = result.get("nextCursor") + if next_cursor is None: + return + if not isinstance(next_cursor, str) or next_cursor == cursor: + raise Refused("invalid_cursor") + cursor = next_cursor + raise Refused("inventory_limit") + + +def steer(rpc, target, turn_id, text, *, expected_home, opt_in=False, + require_instant=False, dry_run=False): + if not opt_in: + raise Refused("steering_opt_in_required") + if not isinstance(turn_id, str) or not turn_id.strip() or len(turn_id) > 128: + raise Refused("expected_turn_required") + if not text or "\x00" in text or len(text.encode("utf-8")) > 65536: + raise Refused("invalid_message") + try: + thread_id = str(uuid.UUID(target.removeprefix("codex:"))) + except ValueError as exc: + raise Refused("full_thread_uuid_required") from exc + actual_home = rpc.info.get("codexHome") + if not expected_home or not isinstance(actual_home, str) or Path(actual_home).resolve() != Path(expected_home).resolve(): + raise Refused("server_home_mismatch") + if thread_id not in list(pages(rpc, "thread/loaded/list", {})): + raise Refused("target_not_loaded") + thread = rpc.call("thread/read", {"threadId": thread_id, "includeTurns": False}).get("thread") + if not isinstance(thread, dict) or thread.get("id") != thread_id: + raise Refused("thread_mismatch") + status = thread.get("status") + if not isinstance(status, dict) or status.get("type") != "active": + raise Refused("target_not_active") + if status.get("activeFlags"): + raise Refused("active_interaction_pending") + if thread.get("canAcceptDirectInput") is not True: + raise Refused("direct_input_unverified") + flag = next((f.get("enabled") for f in pages(rpc, "experimentalFeature/list", {"threadId": thread_id}) + if isinstance(f, dict) and f.get("name") == "instant_interrupt"), None) + if flag is not True and flag is not False: + flag = None + if require_instant and flag is not True: + raise Refused("instant_interrupt_not_enabled_or_verified") + result = {"schemaVersion": 1, "ok": True, "experimental": True, "mode": "steer", + "submitted": False, "consumptionConfirmed": False, "dryRun": dry_run, + "instantInterruptEnabledSnapshot": flag, + "status": "validated" if dry_run else "accepted"} + if dry_run: + # Native expectedTurnId is checked only on submission; do not pretend + # that this metadata-only dry-run proves it is the current turn. + result["expectedTurnValidated"] = False + return result + try: + accepted = rpc.call("turn/steer", { + "threadId": thread_id, "expectedTurnId": turn_id, + "input": [{"type": "text", "text": text}], + "clientUserMessageId": str(uuid.uuid4()), + }) + except Rejected as exc: + if exc.code in (-32600, -32601): + raise + raise Unknown("submission_outcome_unknown") from exc + except (OSError, ValueError, Refused) as exc: + raise Unknown("submission_outcome_unknown") from exc + if accepted.get("turnId") != turn_id: + raise Unknown("returned_turn_mismatch") + return dict(result, submitted=True, turnId=turn_id) + + +def main(): + p = argparse.ArgumentParser(description=__doc__) + p.add_argument("--experimental-steering", action="store_true", required=True) + p.add_argument("--socket", required=True) + p.add_argument("--expected-home", required=True) + p.add_argument("--to", required=True) + p.add_argument("--expected-turn-id", required=True) + p.add_argument("--message", required=True) + p.add_argument("--require-instant", action="store_true") + p.add_argument("--dry-run", action="store_true") + args = p.parse_args() + try: + text = sys.stdin.read(65537) if args.message == "-" else args.message + with Rpc(args.socket) as rpc: + result = steer(rpc, args.to, args.expected_turn_id, text, + expected_home=args.expected_home, opt_in=True, + require_instant=args.require_instant, dry_run=args.dry_run) + code = 0 + except Unknown as exc: + result = {"ok": False, "submitted": None, "status": "unknown", + "reason": str(exc), "retryAllowed": False} + code = 1 + except (Refused, OSError, ValueError) as exc: + result = {"ok": False, "submitted": False, "status": "refused", + "reason": str(exc) if isinstance(exc, Refused) else "connection_failed"} + code = 1 + print(json.dumps(result)) + return code + + +if __name__ == "__main__": + import sys + raise SystemExit(main()) diff --git a/experiments/codex_steering/test_steer.py b/experiments/codex_steering/test_steer.py new file mode 100644 index 00000000..8f599aa4 --- /dev/null +++ b/experiments/codex_steering/test_steer.py @@ -0,0 +1,162 @@ +import unittest +from unittest import mock +from pathlib import Path +from types import SimpleNamespace +import os +import stat +from steer import Rpc, Refused, Unknown, Rejected, pages, steer + +THREAD = "00000000-0000-4000-8000-000000000001" +HOME = "/test-home" + + +class Fake: + info = {"codexHome": HOME} + + def __init__(self, active=True, loaded=True, flag=True, error=None, direct=True, flags=None): + self.active, self.loaded, self.flag, self.error, self.direct = active, loaded, flag, error, direct + self.flags = flags or [] + self.calls = [] + + def call(self, method, params): + self.calls.append((method, params)) + if method == "thread/loaded/list": + return {"data": [THREAD] if self.loaded else [], "nextCursor": None} + if method == "thread/read": + return {"thread": {"id": THREAD, "status": {"type": "active" if self.active else "idle", "activeFlags": self.flags}, "canAcceptDirectInput": self.direct}} + if method == "experimentalFeature/list": + return {"data": [{"name": "instant_interrupt", "enabled": self.flag}], "nextCursor": None} + if self.error: + raise self.error + return {"turnId": "turn-test"} + + +def send(rpc, **kwargs): + return steer(rpc, THREAD, kwargs.pop("turn", "turn-test"), kwargs.pop("text", "test input"), + expected_home=kwargs.pop("home", HOME), opt_in=kwargs.pop("opt_in", True), **kwargs) + + +class Contract(unittest.TestCase): + def test_opt_in(self): + rpc = Fake() + with self.assertRaises(Refused): + send(rpc, opt_in=False) + self.assertEqual(rpc.calls, []) + + def test_exact_one_shot(self): + rpc = Fake() + result = send(rpc) + self.assertTrue(result["submitted"]) + self.assertFalse(result["consumptionConfirmed"]) + self.assertEqual(sum(m == "turn/steer" for m, _ in rpc.calls), 1) + self.assertEqual(set(rpc.calls[-1][1]), {"threadId", "expectedTurnId", "input", "clientUserMessageId"}) + self.assertNotIn("turn/start", [m for m, _ in rpc.calls]) + self.assertFalse(rpc.calls[1][1]["includeTurns"]) + + def test_dry_run_not_proof_of_turn(self): + rpc = Fake() + result = send(rpc, dry_run=True) + self.assertFalse(result["submitted"]) + self.assertFalse(result["expectedTurnValidated"]) + self.assertNotIn("turn/steer", [m for m, _ in rpc.calls]) + + def test_idle(self): + with self.assertRaisesRegex(Refused, "not_active"): + send(Fake(active=False)) + + def test_unloaded(self): + with self.assertRaisesRegex(Refused, "not_loaded"): + send(Fake(loaded=False)) + + def test_home(self): + with self.assertRaisesRegex(Refused, "home_mismatch"): + send(Fake(), home="/wrong-home") + + def test_capability(self): + with self.assertRaisesRegex(Refused, "unverified"): + send(Fake(direct=None)) + + def test_approval(self): + with self.assertRaisesRegex(Refused, "interaction_pending"): + send(Fake(flags=["waitingOnApproval"])) + + def test_disabled_requires_instant(self): + with self.assertRaisesRegex(Refused, "not_enabled"): + send(Fake(flag=False), require_instant=True) + + def test_unknown_requires_instant(self): + with self.assertRaises(Refused): + send(Fake(flag=None), require_instant=True) + + def test_noninstant_steer_is_distinct(self): + self.assertFalse(send(Fake(flag=False))["instantInterruptEnabledSnapshot"]) + + def test_turn_required(self): + with self.assertRaises(Refused): + send(Fake(), turn="") + + def test_invalid_message(self): + for text in ("", "\x00", "가" * 22000): + with self.assertRaises(Refused): + send(Fake(), text=text) + + def test_wrong_turn_no_fallback(self): + rpc = Fake(error=Rejected("turn/steer", -32600)) + with self.assertRaises(Refused): + send(rpc) + self.assertEqual(sum(m == "turn/steer" for m, _ in rpc.calls), 1) + + def test_timeout_unknown(self): + rpc = Fake(error=TimeoutError()) + with self.assertRaises(Unknown): + send(rpc) + self.assertEqual(sum(m == "turn/steer" for m, _ in rpc.calls), 1) + + def test_invalid_response_unknown(self): + with self.assertRaises(Unknown): + send(Fake(error=Refused("invalid_rpc_response"))) + + def test_internal_error_unknown(self): + with self.assertRaises(Unknown): + send(Fake(error=Rejected("turn/steer", -32603))) + + def test_returned_turn_mismatch_unknown(self): + with self.assertRaises(Unknown): + send(Fake(), turn="different-turn") + + def test_relative_socket_refused(self): + with self.assertRaisesRegex(Refused, "absolute"): + Rpc("relative.sock").__enter__() + + def test_missing_socket_no_connection(self): + with mock.patch.object(Path, "lstat", side_effect=FileNotFoundError()): + with self.assertRaises(FileNotFoundError): + Rpc("/dummy/missing.sock").__enter__() + + def test_socket_owner_refused(self): + value = SimpleNamespace(st_uid=os.getuid() + 1, st_mode=stat.S_IFSOCK | 0o600) + with mock.patch.object(Path, "lstat", return_value=value), \ + mock.patch.object(Path, "resolve", return_value=Path("/dummy/app.sock")): + with self.assertRaisesRegex(Refused, "owned_socket"): + Rpc("/dummy/app.sock").__enter__() + + def test_socket_mode_refused(self): + value = SimpleNamespace(st_uid=os.getuid(), st_mode=stat.S_IFSOCK | 0o666) + with mock.patch.object(Path, "lstat", return_value=value), \ + mock.patch.object(Path, "resolve", return_value=Path("/dummy/app.sock")): + with self.assertRaisesRegex(Refused, "private_socket"): + Rpc("/dummy/app.sock").__enter__() + + def test_repeated_cursor_refused(self): + rpc = SimpleNamespace(call=lambda *args: {"data": [], "nextCursor": "repeat"}) + with self.assertRaisesRegex(Refused, "cursor"): + list(pages(rpc, "thread/loaded/list", {})) + + def test_invalid_inventory_refused(self): + rpc = SimpleNamespace(call=lambda *args: {"data": None}) + with self.assertRaisesRegex(Refused, "inventory"): + list(pages(rpc, "thread/loaded/list", {})) + + +if __name__ == "__main__": + unittest.main() From 6ebedde93d62d67e750fd5c32c18b8f6564be86d Mon Sep 17 00:00:00 2001 From: YEONGUK LEE Date: Sun, 4 Oct 2026 09:27:52 +0900 Subject: [PATCH 2/3] fix: validate experimental steering RPC errors --- experiments/codex_steering/steer.py | 5 +- experiments/codex_steering/test_steer.py | 67 ++++++++++++++++++++++++ 2 files changed, 71 insertions(+), 1 deletion(-) diff --git a/experiments/codex_steering/steer.py b/experiments/codex_steering/steer.py index d92c37a9..0bdc78fd 100644 --- a/experiments/codex_steering/steer.py +++ b/experiments/codex_steering/steer.py @@ -111,7 +111,10 @@ def call(self, method, params): value = self.read(deadline) if value.get("id") == request_id: if "error" in value: - raise Rejected(method, value["error"].get("code")) + error = value["error"] + if not isinstance(error, dict) or type(error.get("code")) is not int: + raise Refused("invalid_rpc_response") + raise Rejected(method, error["code"]) if not isinstance(value.get("result"), dict): raise Refused("invalid_rpc_response") return value["result"] diff --git a/experiments/codex_steering/test_steer.py b/experiments/codex_steering/test_steer.py index 8f599aa4..03bc4020 100644 --- a/experiments/codex_steering/test_steer.py +++ b/experiments/codex_steering/test_steer.py @@ -1,3 +1,4 @@ +import json import unittest from unittest import mock from pathlib import Path @@ -36,6 +37,72 @@ def send(rpc, **kwargs): expected_home=kwargs.pop("home", HOME), opt_in=kwargs.pop("opt_in", True), **kwargs) +class FrameSocket: + """In-memory JSON frames through the real RPC parser; no server or model.""" + + def __init__(self, method, error): + self.method, self.error = method, error + self.calls = [] + + def send(self, frame): + request = json.loads(frame) + self.calls.append(request) + result = Fake().call(request["method"], request["params"]) + self.response = {"id": request["id"], "result": result} + if request["method"] == self.method: + # A plausible success result must not mask a malformed/error reply. + self.response["error"] = self.error + + def recv(self, timeout): + return json.dumps(self.response) + + +class RpcResponseContract(unittest.TestCase): + malformed_errors = (None, [], "private provider detail", {}, {"code": None}, + {"code": True}, {"code": False}, {"code": -32600.0}, + {"code": "-32600"}, {"code": []}, {"code": {}}) + + def rpc(self, method, error): + rpc = Rpc("/unused-fixture.sock") + rpc.info = {"codexHome": HOME} + rpc.socket = FrameSocket(method, error) + return rpc + + def test_malformed_error_before_submission_is_refused(self): + for error in self.malformed_errors: + with self.subTest(error=error): + rpc = self.rpc("thread/loaded/list", error) + with self.assertRaisesRegex(Refused, "^invalid_rpc_response$") as raised: + send(rpc) + self.assertIs(type(raised.exception), Refused) + self.assertEqual([r["method"] for r in rpc.socket.calls], ["thread/loaded/list"]) + + def test_malformed_error_after_attempt_is_unknown_without_resend(self): + for error in self.malformed_errors: + with self.subTest(error=error): + rpc = self.rpc("turn/steer", error) + with self.assertRaisesRegex(Unknown, "^submission_outcome_unknown$"): + send(rpc) + self.assertEqual([r["method"] for r in rpc.socket.calls], + ["thread/loaded/list", "thread/read", "experimentalFeature/list", "turn/steer"]) + + def test_valid_error_codes_keep_existing_classification(self): + for code in (-32600, -32601, -32603, 0, 123): + with self.subTest(code=code): + rpc = self.rpc("turn/steer", {"code": code, "message": "private provider detail"}) + expected = Rejected if code in (-32600, -32601) else Unknown + with self.assertRaises(expected) as raised: + send(rpc) + self.assertNotIn("private provider detail", str(raised.exception)) + self.assertEqual(sum(r["method"] == "turn/steer" for r in rpc.socket.calls), 1) + + def test_absent_error_keeps_successful_result(self): + rpc = self.rpc("unused", None) + result = send(rpc) + self.assertTrue(result["submitted"]) + self.assertFalse(result["consumptionConfirmed"]) + + class Contract(unittest.TestCase): def test_opt_in(self): rpc = Fake() From eb00c15c203fe86a02dff83126b178e0f82e42be Mon Sep 17 00:00:00 2001 From: YEONGUK LEE Date: Thu, 22 Oct 2026 19:20:51 +0900 Subject: [PATCH 3/3] fix(experiment): bound steering fixtures and preserve submission evidence --- experiments/codex_steering/README.md | 13 +- experiments/codex_steering/native_fixture.py | 126 ++++++++++++------ experiments/codex_steering/steer.py | 48 +++++-- .../codex_steering/test_native_fixture.py | 114 ++++++++++++++++ experiments/codex_steering/test_steer.py | 122 ++++++++++++++++- 5 files changed, 367 insertions(+), 56 deletions(-) create mode 100644 experiments/codex_steering/test_native_fixture.py diff --git a/experiments/codex_steering/README.md b/experiments/codex_steering/README.md index 5c3f2c9a..eda24a29 100644 --- a/experiments/codex_steering/README.md +++ b/experiments/codex_steering/README.md @@ -5,6 +5,12 @@ Existing package rules exclude `experiments/`. Canonical queue/send/wake, generated runtime, Relay/SSH, and the Side Session experiment (#245/#246) are unchanged. +Malformed or contradictory RPC replies, including excessive JSON nesting, are +refused before input and become `unknown` after a steering attempt. Ordinary +connection-close errors preserve validated submission evidence or the original +unknown outcome. The offline parser and cleanup tests do not establish native +model consumption or support for newer Codex versions. + ## Technical review Codex 0.159.0 introduced default-off `instant_interrupt`, still under development @@ -47,7 +53,12 @@ gate stays closed; other cases must not produce it during a 0.75-second window. After gate release, all four cases must include the new input in the next request and finish. Steering must retain the same turn; queue must start a new turn. These are deterministic synthetic protocol observations, not model ACKs -or a real-provider latency SLA. Each case has a 45-second deadline, first +or a real-provider latency SLA. Each case shares a 45-second active-work budget +across startup, RPCs and provider waits. A watchdog sends TERM then KILL after +a 0.2-second grace; cleanup has additional bounded waits for connection close, +process reaping and process-group checks. These are application time budgets, +not guarantees against blocked OS operations. The version probe has a separate +five-second timeout. First scenario assertion failure stops the batch. Failed gates are findings, not redefined expectations. Fixture/setup defects may be corrected separately. diff --git a/experiments/codex_steering/native_fixture.py b/experiments/codex_steering/native_fixture.py index 96ddb6ed..875f2a86 100644 --- a/experiments/codex_steering/native_fixture.py +++ b/experiments/codex_steering/native_fixture.py @@ -4,7 +4,6 @@ assertion failure. Fixture-created homes/sockets are temporary and private. """ import argparse -import contextlib import http.server import json import os @@ -20,8 +19,36 @@ from steer import Rpc, Refused, Rejected, pages, steer +def remaining(deadline, maximum): + budget = deadline - time.monotonic() + if budget <= 0: + raise TimeoutError("fixture_case_deadline") + return min(maximum, budget) + + +def wait_with_budget(event, seconds, deadline): + result = event.wait(remaining(deadline, seconds)) + # A shortened negative-control window is not a successful observation. + remaining(deadline, seconds) + return result + + +def native_version(): + executable = shutil.which("codex") + if executable is None: + raise RuntimeError("fixture_executable_missing") + try: + version = subprocess.check_output([executable, "--version"], text=True, timeout=5).strip() + except subprocess.TimeoutExpired as exc: + raise RuntimeError("fixture_version_deadline") from exc + if version != "codex-cli 0.159.2": + raise RuntimeError("unsupported_fixture_version") + return executable, version + + class Provider: - def __init__(self): + def __init__(self, deadline): + self.deadline = deadline self.release = threading.Event() self.second = threading.Event() self.lock = threading.Lock() @@ -79,7 +106,8 @@ def emit(value): "type": "message", "role": "assistant", "id": "initial-msg", "content": [{"type": "output_text", "text": ""}]}}) emit({"type": "response.output_text.delta", "delta": "ORIGINAL-PREFIX"}) - if not provider.release.wait(40): + if not provider.release.wait(max(0, min(40, + provider.deadline - time.monotonic()))): return message = "ORIGINAL-FINISHED" if number == 1 else "SYNTHETIC-UPDATED-RESPONSE" emit({"type": "response.output_item.done", "item": { @@ -105,23 +133,39 @@ def close(self): assert not self.thread.is_alive() -def kill_owned(process): - # Keep the leader PID unreaped through group cleanup, never reuse an old pgid. - for sig, grace in ((signal.SIGTERM, 0.2), (signal.SIGKILL, 0)): +def signal_owned(process, sig): + # Caller retains the unreaped leader until every signaling thread has joined. + try: + os.killpg(process.pid, sig) + except ProcessLookupError: + pass + except PermissionError: + # macOS reports EPERM for groups containing only zombies. + members = subprocess.run(["ps", "-o", "stat=", "-g", str(process.pid)], + capture_output=True, text=True, timeout=3) + if sys.platform != "darwin" or members.returncode not in (0, 1) or any( + line.strip() and not line.strip().startswith("Z") for line in members.stdout.splitlines()): + raise + + +def terminate_owned_group(process): + signal_owned(process, signal.SIGTERM) + time.sleep(0.2) + signal_owned(process, signal.SIGKILL) + + +def deadline_watchdog(process, deadline, done, hit, errors): + if not done.wait(max(0, deadline - time.monotonic())): + hit.set() try: - os.killpg(process.pid, sig) - except ProcessLookupError: - pass - except PermissionError: - # macOS reports EPERM for groups containing only zombies. Do not - # ignore denial while any member is still running. - members = subprocess.run(["ps", "-o", "stat=", "-g", str(process.pid)], - capture_output=True, text=True, timeout=3) - if sys.platform != "darwin" or members.returncode not in (0, 1) or any( - line.strip() and not line.strip().startswith("Z") for line in members.stdout.splitlines()): - raise - if grace: - time.sleep(grace) + terminate_owned_group(process) + except Exception as exc: + errors.append(exc) + + +def kill_owned(process): + # Watchdog must already be joined. Never poll/wait before group signaling. + terminate_owned_group(process) process.wait(timeout=5) check = subprocess.run(["ps", "-o", "stat=", "-g", str(process.pid)], capture_output=True, text=True, timeout=3) @@ -131,13 +175,14 @@ def kill_owned(process): def run_case(executable, root, mode, enabled): t0 = time.monotonic() + deadline = t0 + 45 case = root / (mode + ("-on" if enabled else "-off")) case.mkdir(mode=0o700) home, cwd = case / "home", case / "cwd" home.mkdir(mode=0o700) cwd.mkdir(mode=0o700) sock = case / "app.sock" - provider = Provider() + provider = Provider(deadline) port = provider.server.server_address[1] # This configuration is generated fixture state. It contains no credentials. config = f'''model = "gpt-6.1-sol" @@ -164,27 +209,21 @@ def run_case(executable, root, mode, enabled): physical = None deadline_hit = threading.Event() watchdog_done = threading.Event() - - def watchdog(): - if not watchdog_done.wait(45) and process is not None: - deadline_hit.set() - with contextlib.suppress(ProcessLookupError): - os.killpg(process.pid, signal.SIGTERM) - - watcher = threading.Thread(target=watchdog, daemon=True) - watcher.start() + watchdog_errors = [] + watcher = None try: process = subprocess.Popen([executable, "app-server", "--listen", "unix://" + str(sock)], cwd=cwd, env=env, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, start_new_session=True) - deadline = t0 + 15 + watcher = threading.Thread(target=deadline_watchdog, + args=(process, deadline, watchdog_done, deadline_hit, watchdog_errors), daemon=True) + watcher.start() + startup_deadline = min(deadline, t0 + 15) while not sock.exists(): - if time.monotonic() > deadline: - raise RuntimeError("fixture_native_startup_deadline") - time.sleep(0.05) + time.sleep(remaining(startup_deadline, 0.05)) physical = sock.resolve() - observer = Rpc(sock, timeout=10) + observer = Rpc(sock, timeout=10, deadline=deadline) with observer: target = observer.call("thread/start", { "cwd": str(cwd), "model": "gpt-6.1-sol", "sandbox": "read-only", @@ -194,7 +233,7 @@ def watchdog(): "input": [{"type": "text", "text": "INITIAL-STEER247 synthetic protocol scenario"}]})["turn"]["id"] observer.wait_event(lambda e: e.get("method") == "item/agentMessage/delta" and "ORIGINAL-PREFIX" in e.get("params", {}).get("delta", "")) - with Rpc(sock, timeout=10) as sender: + with Rpc(sock, timeout=10, deadline=deadline) as sender: flag_rows = list(pages(sender, "experimentalFeature/list", {"threadId": target})) flag = next(f["enabled"] for f in flag_rows if f["name"] == "instant_interrupt") assert flag is enabled, "feature_snapshot_mismatch" @@ -233,11 +272,12 @@ def watchdog(): sender.call("thread/queue/add", {"threadId": target, "clientUserMessageId": str(__import__("uuid").uuid4()), "input": [{"type": "text", "text": "UPDATE-STEER247 synthetic queued follow-up"}]}) - before_release = provider.second.wait(5 if mode == "steer" and enabled else 0.75) + before_release = wait_with_budget(provider.second, + 5 if mode == "steer" and enabled else 0.75, deadline) expected = mode == "steer" and enabled assert before_release is expected, "preemption_gate_mismatch" provider.release.set() - assert provider.second.wait(10), "updated_request_missing" + assert wait_with_budget(provider.second, 10, deadline), "updated_request_missing" observer.wait_event(lambda e: e.get("method") == "item/completed" and e.get("params", {}).get("item", {}).get("text") == "SYNTHETIC-UPDATED-RESPONSE", timeout=10) done = observer.wait_event(lambda e: e.get("method") == "turn/completed" @@ -268,6 +308,7 @@ def watchdog(): "syntheticResponseObserved": True, "wrongTurnRejected": True, "idleRejected": True, "homeMismatchRejected": True, "missingTargetRejected": True, "elapsedMs": round((time.monotonic() - t0) * 1000, 2)} + remaining(deadline, 1) return result except Refused as exc: if str(exc) == "unsupported_server_version" and hasattr(observer, "info"): @@ -277,7 +318,10 @@ def watchdog(): finally: provider.release.set() watchdog_done.set() - watcher.join(timeout=2) + if watcher is not None: + # Its only active work is TERM, a 0.2s grace, KILL and bounded ps + # checks. Join before reaping so the PID cannot be reused under it. + watcher.join() try: if process is not None: kill_owned(process) @@ -291,6 +335,7 @@ def watchdog(): if sock.is_symlink() or sock.exists(): sock.unlink() assert not deadline_hit.is_set(), "batch_deadline_hit" + assert not watchdog_errors, "watchdog_cleanup_failed" def main(): @@ -299,10 +344,7 @@ def main(): p.add_argument("--temp-root", required=True) p.add_argument("--repetitions", type=int, default=1, choices=range(1, 6)) args = p.parse_args() - executable = shutil.which("codex") - version = subprocess.check_output([executable, "--version"], text=True).strip() - if version != "codex-cli 0.159.2": - raise RuntimeError("unsupported_fixture_version") + executable, version = native_version() # tempfile is strictly within the operator-selected approved scratch root. with tempfile.TemporaryDirectory(prefix="codex-steer247-", dir=args.temp_root) as tmp: root = Path(tmp) diff --git a/experiments/codex_steering/steer.py b/experiments/codex_steering/steer.py index 0bdc78fd..95475db1 100644 --- a/experiments/codex_steering/steer.py +++ b/experiments/codex_steering/steer.py @@ -31,11 +31,21 @@ def __init__(self, method, code): class Rpc: LIMIT = 1024 * 1024 - def __init__(self, path, timeout=10): + def __init__(self, path, timeout=10, *, deadline=None): self.path, self.timeout = Path(path), timeout + self.deadline = deadline + self.close_failed = False self.sequence = 0 self.events = [] + def remaining(self, maximum=None): + remaining = self.timeout if maximum is None else maximum + if self.deadline is not None: + remaining = min(remaining, self.deadline - time.monotonic()) + if remaining <= 0: + raise TimeoutError() + return remaining + def __enter__(self): if not self.path.is_absolute(): raise Refused("absolute_socket_required") @@ -54,7 +64,7 @@ def __enter__(self): from websockets.sync.client import unix_connect except ImportError as exc: raise Refused("websockets_dependency_missing") from exc - self.connection = unix_connect(str(resolved), open_timeout=self.timeout, + self.connection = unix_connect(str(resolved), open_timeout=self.remaining(), close_timeout=1, max_size=self.LIMIT) try: self.socket = self.connection.__enter__() @@ -65,7 +75,8 @@ def __enter__(self): "clientInfo": {"name": "session_peer_steer_experiment", "version": "0.0.0"}, "capabilities": {"experimentalApi": True}, }) - if not re.match(r"(?:codex-tui|codex_cli_rs|codex-cli|session_peer_steer_experiment)/0\.159\.2(?:\s|$)", self.info.get("userAgent", "")): + agent = self.info.get("userAgent") + if not isinstance(agent, str) or not re.match(r"(?:codex-tui|codex_cli_rs|codex-cli|session_peer_steer_experiment)/0\.159\.2(?:\s|$)", agent): raise Refused("unsupported_server_version") self.write({"method": "initialized"}) return self @@ -74,9 +85,16 @@ def __enter__(self): raise def __exit__(self, *args): - self.connection.__exit__(*args) + try: + self.connection.__exit__(*args) + except Exception: + # Closing a transport cannot negate a validated response or turn + # an already attempted, unknown submission into a safe refusal. + self.close_failed = True + return False def write(self, value): + self.remaining() frame = json.dumps(value, ensure_ascii=False) if len(frame.encode("utf-8")) > self.LIMIT: raise Refused("frame_too_large") @@ -86,18 +104,20 @@ def write(self, value): raise ConnectionError("transport_closed") from exc def read(self, deadline): - remaining = deadline - time.monotonic() - if remaining <= 0: - raise TimeoutError() + remaining = self.remaining(deadline - time.monotonic()) try: frame = self.socket.recv(timeout=remaining) except TimeoutError: raise except Exception as exc: raise ConnectionError("transport_closed") from exc + self.remaining(deadline - time.monotonic()) if not isinstance(frame, str): raise Refused("invalid_frame") - value = json.loads(frame) + try: + value = json.loads(frame) + except (ValueError, RecursionError) as exc: + raise Refused("invalid_rpc_response") from exc if not isinstance(value, dict): raise Refused("invalid_rpc_response") return value @@ -105,14 +125,17 @@ def read(self, deadline): def call(self, method, params): self.sequence += 1 request_id = self.sequence + deadline = time.monotonic() + self.remaining() self.write({"id": request_id, "method": method, "params": params}) - deadline = time.monotonic() + self.timeout while True: value = self.read(deadline) - if value.get("id") == request_id: + if type(value.get("id")) is int and value["id"] == request_id: + if ("result" in value) == ("error" in value): + raise Refused("invalid_rpc_response") if "error" in value: error = value["error"] - if not isinstance(error, dict) or type(error.get("code")) is not int: + if (not isinstance(error, dict) or type(error.get("code")) is not int + or not isinstance(error.get("message"), str)): raise Refused("invalid_rpc_response") raise Rejected(method, error["code"]) if not isinstance(value.get("result"), dict): @@ -125,8 +148,9 @@ def call(self, method, params): self.events.append(value) def wait_event(self, predicate, timeout=10): - deadline = time.monotonic() + timeout + deadline = time.monotonic() + self.remaining(timeout) while True: + self.remaining(deadline - time.monotonic()) for index, event in enumerate(self.events): if predicate(event): return self.events.pop(index) diff --git a/experiments/codex_steering/test_native_fixture.py b/experiments/codex_steering/test_native_fixture.py new file mode 100644 index 00000000..1603cb13 --- /dev/null +++ b/experiments/codex_steering/test_native_fixture.py @@ -0,0 +1,114 @@ +"""Offline harness checks: fake APIs and an owned Python sleeper, never Codex.""" +import os +import select +import signal +import subprocess +import sys +import tempfile +import threading +import time +import unittest +from unittest import mock + +import native_fixture as fixture + + +class FixtureBudget(unittest.TestCase): + def test_failed_case_stops_batch_and_removes_its_temporary_home(self): + with tempfile.TemporaryDirectory(prefix="codex-batch-", dir=os.environ.get("TMPDIR")) as scratch: + argv = ['fixture', '--run-native-fixture', '--temp-root', scratch, '--repetitions', '2'] + with mock.patch.object(sys, 'argv', argv), \ + mock.patch.object(fixture, 'native_version', return_value=('/fake/codex', 'codex-cli 0.159.2')), \ + mock.patch.object(fixture, 'run_case', side_effect=TimeoutError('fixture_case_deadline')) as case: + with self.assertRaises(TimeoutError): + fixture.main() + case.assert_called_once() + self.assertEqual(os.listdir(scratch), []) + + def test_version_timeout_is_bounded_and_fixed(self): + with mock.patch.object(fixture.shutil, "which", return_value="/fake/codex"), \ + mock.patch.object(fixture.subprocess, "check_output", + side_effect=subprocess.TimeoutExpired("private-command", 5)) as probe: + with self.assertRaisesRegex(RuntimeError, "^fixture_version_deadline$"): + fixture.native_version() + probe.assert_called_once_with(["/fake/codex", "--version"], text=True, timeout=5) + + def test_missing_executable_never_launches(self): + with mock.patch.object(fixture.shutil, "which", return_value=None), \ + mock.patch.object(fixture.subprocess, "check_output") as probe: + with self.assertRaisesRegex(RuntimeError, "^fixture_executable_missing$"): + fixture.native_version() + probe.assert_not_called() + + def test_delayed_waits_share_one_budget(self): + clock = [10.0] + event = mock.Mock() + + def delayed_api(timeout): + clock[0] += timeout + return False + + event.wait.side_effect = delayed_api + with mock.patch.object(fixture.time, "monotonic", side_effect=lambda: clock[0]): + self.assertFalse(fixture.wait_with_budget(event, 0.75, 11.0)) + with self.assertRaisesRegex(TimeoutError, "fixture_case_deadline"): + fixture.wait_with_budget(event, 10, 11.0) + with self.assertRaises(TimeoutError): + fixture.wait_with_budget(event, 10, 11.0) + self.assertEqual(event.wait.call_args_list, [mock.call(0.75), mock.call(0.25)]) + + def test_expired_budget_cannot_pass_even_with_ready_event(self): + event = threading.Event() + event.set() + with mock.patch.object(fixture.time, "monotonic", return_value=12): + with self.assertRaises(TimeoutError): + fixture.wait_with_budget(event, 10, 11) + + def test_stopped_watchdog_never_signals(self): + done, hit, errors = threading.Event(), threading.Event(), [] + done.set() + with mock.patch.object(fixture, "terminate_owned_group") as terminate: + fixture.deadline_watchdog(object(), time.monotonic(), done, hit, errors) + terminate.assert_not_called() + self.assertFalse(hit.is_set()) + self.assertEqual(errors, []) + + @unittest.skipUnless(os.name == "posix", "owned POSIX process group") + def test_deadline_kills_term_resistant_owned_process_before_reaping(self): + # TMPDIR is selected by the invoking review/test workflow. + with tempfile.TemporaryDirectory(prefix="codex-steer-watchdog-", + dir=os.environ.get("TMPDIR")) as scratch: + process = subprocess.Popen([sys.executable, "-I", "-c", + "import signal,time; signal.signal(signal.SIGTERM, signal.SIG_IGN); " + "print('ready', flush=True); time.sleep(60)"], cwd=scratch, + stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, + start_new_session=True) + watcher = None + done, hit, errors = threading.Event(), threading.Event(), [] + try: + self.assertTrue(select.select([process.stdout], [], [], 3)[0], "dummy did not start") + self.assertEqual(process.stdout.readline(), b"ready\n") + started = time.monotonic() + watcher = threading.Thread(target=fixture.deadline_watchdog, + args=(process, started + 0.1, done, hit, errors), daemon=True) + watcher.start() + watcher.join(timeout=8) + self.assertFalse(watcher.is_alive(), "watchdog failed to finish") + self.assertTrue(hit.is_set()) + self.assertEqual(errors, []) + self.assertIsNone(process.returncode, "leader was reaped by watchdog") + # The group has received KILL; normal cleanup owns the sole reap. + fixture.kill_owned(process) + self.assertEqual(process.returncode, -signal.SIGKILL) + self.assertLess(time.monotonic() - started, 8) + finally: + done.set() + if watcher is not None: + watcher.join() + if process.returncode is None: + fixture.kill_owned(process) + process.stdout.close() + + +if __name__ == "__main__": + unittest.main() diff --git a/experiments/codex_steering/test_steer.py b/experiments/codex_steering/test_steer.py index 03bc4020..1d8b2389 100644 --- a/experiments/codex_steering/test_steer.py +++ b/experiments/codex_steering/test_steer.py @@ -1,4 +1,7 @@ import json +import contextlib +import io +import sys import unittest from unittest import mock from pathlib import Path @@ -6,6 +9,7 @@ import os import stat from steer import Rpc, Refused, Unknown, Rejected, pages, steer +import steer as module THREAD = "00000000-0000-4000-8000-000000000001" HOME = "/test-home" @@ -50,7 +54,7 @@ def send(self, frame): result = Fake().call(request["method"], request["params"]) self.response = {"id": request["id"], "result": result} if request["method"] == self.method: - # A plausible success result must not mask a malformed/error reply. + self.response.pop("result") self.response["error"] = self.error def recv(self, timeout): @@ -102,6 +106,122 @@ def test_absent_error_keeps_successful_result(self): self.assertTrue(result["submitted"]) self.assertFalse(result["consumptionConfirmed"]) + def test_contradictory_or_incomplete_error_cannot_prove_non_submission(self): + for error in ({"code": -32600}, {"code": -32600, "message": None}, + {"code": -32600, "message": "provider detail"}): + rpc = self.rpc("turn/steer", error) + receive = rpc.socket.recv + + def recv(timeout): + if rpc.socket.calls[-1]["method"] == "turn/steer" and isinstance(error.get("message"), str): + rpc.socket.response["result"] = {"turnId": "turn-test"} + return receive(timeout) + + rpc.socket.recv = recv + with self.assertRaises(Unknown): + send(rpc) + self.assertEqual(sum(r["method"] == "turn/steer" for r in rpc.socket.calls), 1) + + def test_bad_json_before_or_after_submission_keeps_outcome_boundary(self): + for raw in ('{"id":', '[' * 2000 + '0' + ']' * 2000): + for method, expected in (("thread/loaded/list", Refused), ("turn/steer", Unknown)): + rpc = self.rpc("unused", None) + receive = rpc.socket.recv + rpc.socket.recv = lambda timeout: raw if rpc.socket.calls[-1]["method"] == method else receive(timeout) + with self.assertRaises(expected): + send(rpc) + self.assertEqual(sum(r["method"] == "turn/steer" for r in rpc.socket.calls), int(method == "turn/steer")) + + def test_bool_or_float_id_never_matches_integer_request(self): + for malformed_id in (True, 1.0): + rpc = self.rpc("unused", None) + rpc.socket.recv = lambda timeout: json.dumps({"id": malformed_id, "result": {"data": [THREAD]}}) + with self.assertRaises(Refused): + send(rpc) + self.assertEqual(len(rpc.socket.calls), 1) + + def test_cli_deep_reply_is_unknown_without_diagnostic_reflection(self): + rpc = self.rpc("unused", None) + receive = rpc.socket.recv + rpc.socket.recv = lambda timeout: '[' * 2000 + '0' + ']' * 2000 if rpc.socket.calls[-1]["method"] == "turn/steer" else receive(timeout) + argv = ['steer.py', '--experimental-steering', '--socket', '/unused', '--expected-home', HOME, + '--to', THREAD, '--expected-turn-id', 'turn-test', '--message', 'private-body-sentinel'] + out = io.StringIO() + with mock.patch.object(sys, 'argv', argv), mock.patch.object(Rpc, '__enter__', return_value=rpc), \ + mock.patch.object(Rpc, '__exit__', return_value=None), contextlib.redirect_stdout(out): + code = module.main() + self.assertEqual(code, 1) + result = json.loads(out.getvalue()) + self.assertIsNone(result['submitted']) + self.assertEqual(result['status'], 'unknown') + self.assertIs(result['retryAllowed'], False) + self.assertNotIn('private-body-sentinel', out.getvalue()) + self.assertEqual(sum(r['method'] == 'turn/steer' for r in rpc.socket.calls), 1) + + def test_total_deadline_stops_next_request_and_cached_event(self): + rpc = self.rpc("unused", None) + rpc.deadline = 100 + with mock.patch.object(module.time, 'monotonic', return_value=99): + rpc.call('thread/loaded/list', {}) + count = len(rpc.socket.calls) + rpc.events.append({'method': 'fixture'}) + with mock.patch.object(module.time, 'monotonic', return_value=100): + with self.assertRaises(TimeoutError): + rpc.call('turn/steer', {}) + with self.assertRaises(TimeoutError): + rpc.wait_event(lambda event: True) + self.assertEqual(len(rpc.socket.calls), count) + + def test_cleanup_error_preserves_known_and_unknown_submission(self): + for successful in (True, False): + rpc = self.rpc('unused' if successful else 'turn/steer', + {'code': -32000, 'message': 'private provider detail'}) + rpc.connection = mock.MagicMock() + rpc.connection.__exit__.side_effect = OSError('private cleanup detail') + argv = ['steer.py', '--experimental-steering', '--socket', '/unused', '--expected-home', HOME, + '--to', THREAD, '--expected-turn-id', 'turn-test', '--message', 'fixture'] + out = io.StringIO() + with mock.patch.object(sys, 'argv', argv), mock.patch.object(module, 'Rpc', return_value=rpc), \ + mock.patch.object(Rpc, '__enter__', return_value=rpc), contextlib.redirect_stdout(out): + code = module.main() + result = json.loads(out.getvalue()) + self.assertEqual(code, 0 if successful else 1) + self.assertEqual(result['status'], 'accepted' if successful else 'unknown') + self.assertIs(result['submitted'], True if successful else None) + if not successful: + self.assertIs(result['retryAllowed'], False) + self.assertTrue(rpc.close_failed) + self.assertNotIn('private', out.getvalue()) + self.assertEqual(sum(r['method'] == 'turn/steer' for r in rpc.socket.calls), 1) + + def test_initialize_rejects_non_string_version_without_traceback(self): + client = SimpleNamespace(unix_connect=mock.MagicMock()) + for agent in (None, {}, [], 1): + rpc = Rpc('/unused/socket') + with mock.patch.dict(sys.modules, {'websockets.sync.client': client}), \ + mock.patch.object(Path, 'stat', return_value=SimpleNamespace(st_uid=os.getuid(), st_mode=0o700)), \ + mock.patch.object(Path, 'lstat', return_value=SimpleNamespace(st_uid=os.getuid(), st_mode=stat.S_IFSOCK)), \ + mock.patch.object(Path, 'resolve', return_value=rpc.path), \ + mock.patch.object(rpc, 'call', return_value={'userAgent': agent}): + with self.assertRaisesRegex(Refused, '^unsupported_server_version$'): + rpc.__enter__() + + def test_response_after_absolute_deadline_is_not_accepted(self): + rpc = self.rpc("unused", None) + rpc.deadline = 100 + now = [99] + receive = rpc.socket.recv + + def recv(timeout): + self.assertLessEqual(timeout, 1) + now[0] = 100 + return receive(timeout) + + rpc.socket.recv = recv + with mock.patch.object(module.time, 'monotonic', side_effect=lambda: now[0]): + with self.assertRaises(TimeoutError): + rpc.call('thread/loaded/list', {}) + class Contract(unittest.TestCase): def test_opt_in(self):