From 4fdc92139c8ff89548fed427be29141170e81cb8 Mon Sep 17 00:00:00 2001 From: YEONGUK LEE Date: Tue, 13 Oct 2026 15:03:21 +0900 Subject: [PATCH 1/9] test(codex): record first-input receipt boundaries across TUI topologies --- experiments/codex_first_input/ORDINARY_TUI.md | 111 ++++++++++++++++++ experiments/codex_first_input/RESULTS.md | 79 +++++++++++++ experiments/codex_first_input/probe.py | 81 +++++++++++++ 3 files changed, 271 insertions(+) create mode 100644 experiments/codex_first_input/ORDINARY_TUI.md create mode 100644 experiments/codex_first_input/RESULTS.md create mode 100644 experiments/codex_first_input/probe.py diff --git a/experiments/codex_first_input/ORDINARY_TUI.md b/experiments/codex_first_input/ORDINARY_TUI.md new file mode 100644 index 00000000..5509a5ab --- /dev/null +++ b/experiments/codex_first_input/ORDINARY_TUI.md @@ -0,0 +1,111 @@ +# Already-open ordinary blank TUI investigation + +Date: 2026-10-13. Platform: macOS arm64. Installed Codex: 0.162.1. +Runtime reference: session-peer v1.1.1, b48144bdf2ad1569031562136f5a60e33780020e. + +## Ordinary in-process TUI: negative result + +An operator-owned `codex --no-alt-screen` was launched without a prompt, remote +endpoint, daemon override, or approval/sandbox override in an already approved +test folder. It reached its idle composer. No user message was entered. + +The account's effective `daemon_auto_start` feature was false. Before and after +launch, `codex app-server daemon version` could not connect because the control +socket did not exist. The owned TUI held an exact UUID-named writer lock, but a +read-only lookup of that UUID in `state_5.sqlite` found zero matching thread rows. +The owned process's Unix sockets were unnamed socket pairs, not a discoverable +named external App Server endpoint. + +Exactly one external official `codex queue --thread --message +` was attempted. It returned exit 1 and RPC code +-32603: `failed to read thread: invalid thread-store request: no rollout found +for thread id ...`. The native TUI remained blank; no ACK was observed. There +was no resend, thread resume/start, or database mutation by the harness. The +TUI exited via `/quit` with exit zero and its owned PID was confirmed absent. + +This refutes a general claim that an arbitrary already-open blank TUI can be +reached through the external App Server protocol in this tested configuration. +The UUID exists in memory and the writer lock, so failure is not proof that an +internal conversation identity does not exist. Nor is it a platform-wide +impossibility proof. + +The earlier dedicated server/remote TUI experiment in RESULTS.md is a different +topology. Its positive result must not be relabeled as this ordinary-TUI test. + +## Scope and safety + +Only the owned native TUI was inspected and targeted. No credentials, environment +values, unrelated transcript bodies, global configuration, permissions, product +code, GitHub state, releases, or operating services were modified. Normal native +startup/queue shutdown may create native account metadata. Existing user sessions +were not terminated. Private UUID/PID evidence is not included in this report. + +Official protocol reference: https://learn.chatgpt.com/docs/app-server + +## Already-open ordinary TUI on a shared server: positive result + +For a separate control, an exclusively owned foreground +`codex app-server --listen unix://` was started before +another ordinary `codex --no-alt-screen`. This second TUI had no `--remote` +argument, no initial prompt, and no feature/permission overrides. It attached to +the existing server and reached its idle composer. This does **not** retrofit the +first in-process TUI or move a running conversation between servers. + +An independent external WebSocket client connected after the second TUI was +already open. The server had exactly one loaded thread; `thread/read` with +`includeTurns:false` (no turn-body read) +showed the expected cwd, idle status, and ability to accept direct input. A +read-only lookup immediately before submission still found zero stored rows for +the exact loaded UUID. The queue operation used that same ID, without +`thread/start`, `thread/resume`, or `turn/start`. +Only selected metadata was reported. This is not a claim that the general +`thread/read` API never returns preview text; this target was the owned blank +thread, and no unrelated conversation or transcript was read. + +One `thread/queue/add` was accepted with `queuedSubmission`. The existing native +TUI displayed the external request and then the exact independently generated +reply `ACK CODEX-SHARED-EMPTY-306-8C2F`, reporting four seconds of work. This proves +first-input receipt in that already-open, shared-server-connected TUI. Queue +acceptance itself was not labeled consumption or ACK; receipt was a separate +direct PTY observation. It does not implement this transport in session-peer. + +An initial read-only stdio-proxy probe timed out before initialization, with no +queue attempt. It was discarded; the successful inspection/submission used the +existing Unix-WebSocket harness with websockets 17.0 in a temporary isolated +environment. The harness only gained an optional bounded ASCII test marker. +No uncertain effect was retried. + +Both test TUIs and the foreground server exited zero; their PIDs were checked +absent and the created logical/physical sockets disappeared. Server shutdown +logged a best-effort MCP session-delete HTTP failure; this was not silently +reported as an error-free shutdown. No managed daemon installation or updater +was started, and the pre-existing startup lock was preserved. + +## Implementation evidence and conclusion + +Read-only independent review of the exact upstream rust-v0.162.1 source supports +the observed distinction: + +- [In-process client](https://github.com/openai/codex/blob/rust-v0.162.1/codex-rs/app-server-client/src/lib.rs#L317): async mpsc worker, no external + process boundary; [in-process server](https://github.com/openai/codex/blob/rust-v0.162.1/codex-rs/app-server/src/in_process.rs#L515): InProcess transport and no + remote-control handle. +- [TUI startup](https://github.com/openai/codex/blob/rust-v0.162.1/codex-rs/tui/src/startup_orchestration.rs#L311): chooses a shared socket at startup when one + exists, otherwise uses the applicable embedded/managed startup path. +- [Queue command](https://github.com/openai/codex/blob/rust-v0.162.1/codex-rs/tui/src/session_queue_commands.rs#L95) accepts a UUID without saved name lookup, + but submits through its selected server. [Queue processor](https://github.com/openai/codex/blob/rust-v0.162.1/codex-rs/app-server/src/request_processors/thread_queue_processor.rs#L253) + requires that server's loaded thread or persisted history. + +Thus, **an already shared-server-connected blank TUI can receive; an arbitrary +embedded blank TUI cannot be claimed reachable externally after startup**. A +safe product feature would require explicit shared-endpoint/owner/target binding +and must refuse unavailable embedded targets, not create a substitute session. +No Linux/Windows, remote machine, arbitrary Orca TUI, restart recovery, or general +production support is established by these macOS tests. + +The harness's exclusive-server assumption, UUID/cwd/status checks and assertions +are experiment preconditions, not production PID/start-time/generation binding. +Its generic JSON decoding is not a substitute for session-peer's strict wire +validation. Do not run it with Python optimization or against a shared server +containing unrelated user work. The WebSocket dependency is test-only and is not +added to the Python core runtime. `experiments/` remains excluded from release +artifacts by the existing packaging rules. diff --git a/experiments/codex_first_input/RESULTS.md b/experiments/codex_first_input/RESULTS.md new file mode 100644 index 00000000..6a00be67 --- /dev/null +++ b/experiments/codex_first_input/RESULTS.md @@ -0,0 +1,79 @@ +# Codex first-input App Server experiment — 2026-10-13 + +## Result + +For the subsequent **already-open ordinary TUI** negative/positive controls, +see [ORDINARY_TUI.md](ORDINARY_TUI.md). An embedded blank TUI was unreachable +through the external queue path, while a plain TUI already connected to a shared +server received the external request and replied. Neither establishes arbitrary +in-process TUI attachment or shipped session-peer support. + +**PASS for one dedicated macOS App Server and its fresh TUI.** Before any user +prompt, the thread was loaded in memory but had **zero matching rows** in the +selected Codex home's state database. One `thread/queue/add` request was accepted. +The native TUI then displayed the request and its independently generated reply: + +```text +ACK CODEX-FIRST-INPUT-306-QUEUE +``` + +The TUI reported three seconds of work at 14:17 KST. This establishes a feasible +first-input transport; it does not make the public `session-peer send` support it. + +## Setup and observations + +- macOS arm64; installed Codex CLI/server 0.162.1; TUI-selected GPT-6-Astra medium. +- A dedicated `codex app-server --listen unix://…` process and a fresh + `codex --remote unix://… --no-alt-screen` TUI shared one private endpoint. + The working directory was an already approved temporary test folder. +- The server inherited the normal account home and authentication. Credentials + were not copied; no user configuration, existing daemon, permission setting, + or unrelated session was changed. +- `thread/loaded/list` contained exactly one ID before submission. + `thread/read` with `includeTurns:false` reported that same thread as idle, + non-ephemeral and able to accept direct input. Its cwd matched the owned TUI. +- Read-only SQLite lookup for that exact ID found zero `threads` rows. This was + checked both during inspection and immediately before the single submission. +- `thread/queue/add` returned `queuedSubmission`; the harness kept + `consumptionConfirmed:false`. Actual receipt evidence came separately from the + owned TUI's displayed response, not from queue deletion or turn completion. +- No `turn/start`, `thread/resume`, direct database write, fallback or resend was + used. The message requested only the fixed reply and no tools or file changes. + +## Reproduction harness + +`probe.py` requires Python with websockets 17.x and an **exclusively owned** fresh +server/TUI. It does not launch or discover an arbitrary server. Operators supply +the private socket, the same account home used to launch that server, and the +exact approved TUI cwd. `--mode inspect` is read-only; `--mode queue` sends once +only after checking a zero-row precondition. Do not run it against an existing +working conversation. Run without Python optimization because its experimental +preconditions use assertions. + +The endpoint may be a native-created symlink to a shorter physical Unix socket; +the harness checks both the logical and resolved private directories and socket +ownership. The first harness inspection rejected this symlink before any RPC; +the ownership-preserving resolution was then added. No retry of a send occurred. + +The public output excludes IDs, paths and native response bodies. Native error +text is omitted. This harness is excluded from release artifacts under the +existing `experiments/` packaging rule. + +## Limits and cleanup + +- This is dedicated-server topology, not proof that ordinary daemon, Orca-hosted, + `--no-daemon`, SSH, Linux or Windows targets can be discovered safely. +- An optional post-reply inspection stopped at the exclusive single-loaded-ID + assertion. The cause of the changed inventory was not investigated; no + additional submission was attempted and no post-reply row-count claim is made. +- A product adapter still needs explicit endpoint ownership and generation + binding, version/capability checks, bounded body-free discovery and approval + routing. A queue response alone must never become a generic receipt ACK. +- The dedicated server and TUI exited with status zero. The native socket and + logical link disappeared; the empty owned scratch directory was removed. + The test conversation remains in the native account's normal history; no + existing history or credentials were deleted. +- Runtime source, canonical CLI, package metadata and operations were unchanged. + +Related issue: [#306](https://github.com/abruption/session-peer/issues/306). +Protocol reference: [Codex App Server](https://learn.chatgpt.com/docs/app-server). diff --git a/experiments/codex_first_input/probe.py b/experiments/codex_first_input/probe.py new file mode 100644 index 00000000..802812a6 --- /dev/null +++ b/experiments/codex_first_input/probe.py @@ -0,0 +1,81 @@ +"""One-shot first-input probe for an exclusively owned Codex app server. + +Requires websockets 17.x. Never starts a server or retries a submission. +Prints sanitized facts, not conversation IDs, paths, or native text. +""" +import argparse +import json +import os +from pathlib import Path +import re +import sqlite3 +import stat +import time +import uuid + +from websockets.sync.client import unix_connect + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--socket", required=True) + parser.add_argument("--codex-home", required=True) + parser.add_argument("--expected-cwd", required=True) + parser.add_argument("--mode", choices=("inspect", "queue"), default="inspect") + parser.add_argument("--ack-marker", help="Harmless ASCII test receipt marker; no credentials") + args = parser.parse_args() + if args.ack_marker is not None and not re.fullmatch(r"[A-Za-z0-9 _-]{1,100}", args.ack_marker): + parser.error("invalid ACK marker") + socket = Path(args.socket) + assert socket.lstat().st_uid == os.getuid() + assert socket.parent.stat().st_mode & 0o077 == 0 + socket = socket.resolve(strict=True) + st = socket.lstat() + assert stat.S_ISSOCK(st.st_mode) and st.st_uid == os.getuid() + assert socket.parent.stat().st_mode & 0o077 == 0 + sequence = 0 + with unix_connect(str(socket), open_timeout=5, close_timeout=1, max_size=1048576) as ws: + def call(method, params): + nonlocal sequence + sequence += 1 + ws.send(json.dumps({"id": sequence, "method": method, "params": params})) + deadline = time.monotonic() + 15 + for _ in range(200): + message = json.loads(ws.recv(timeout=max(.01, deadline-time.monotonic()))) + if message.get("id") == sequence: + return message + if "id" in message: + raise RuntimeError("unexpected server interaction; no approval supplied") + raise RuntimeError("notification limit") + hello = call("initialize", {"clientInfo": {"name": "session-peer-first-input-probe", "version": "0"}, "capabilities": {"experimentalApi": True}}) + assert hello["result"]["userAgent"].startswith("codex-tui/0.162.1 ") + ws.send(json.dumps({"method": "initialized"})) + loaded = call("thread/loaded/list", {"limit": 100})["result"] + assert loaded.get("nextCursor") is None and len(loaded["data"]) == 1 + thread_id = loaded["data"][0] + thread = call("thread/read", {"threadId": thread_id, "includeTurns": False})["result"]["thread"] + assert Path(thread["cwd"]).resolve() == Path(args.expected_cwd).resolve() + assert thread["status"]["type"] == "idle" and thread["canAcceptDirectInput"] is True + rows = 0 + dbs = list(Path(args.codex_home).glob("state_*.sqlite")) + assert dbs + for db in dbs: + with sqlite3.connect(db.as_uri()+"?mode=ro", uri=True) as connection: + rows += connection.execute("SELECT count(*) FROM threads WHERE id=?", (thread_id,)).fetchone()[0] + facts = {"version": "0.162.1", "loadedThreads": 1, "storedRowsBefore": rows, "idle": True, "directInput": True, "ephemeral": thread.get("ephemeral"), "mode": args.mode} + print(json.dumps(facts), flush=True) + if args.mode == "inspect": + return + assert rows == 0, "requires fresh pre-first-input thread" + marker = args.ack_marker or "ACK CODEX-FIRST-INPUT-306-" + args.mode.upper() + message = "Owner-authorized one-shot transport test. Reply exactly " + marker + ". Do not run tools or modify files." + method = "thread/queue/add" + result = call(method, {"threadId": thread_id, "input": [{"type": "text", "text": message}], "clientUserMessageId": str(uuid.uuid4())}) + if "error" in result: + print(json.dumps({"method": method, "accepted": False, "errorCode": result["error"].get("code"), "retryAllowed": False})) + else: + print(json.dumps({"method": method, "accepted": True, "resultKeys": sorted(result["result"]), "consumptionConfirmed": False}), flush=True) + + +if __name__ == "__main__": + main() From 7526e1bc9520da8cff8f8edae0dac039adae9ab0 Mon Sep 17 00:00:00 2001 From: YEONGUK LEE Date: Tue, 13 Oct 2026 15:56:24 +0900 Subject: [PATCH 2/9] test(codex): guard first-input probe server binding --- experiments/codex_first_input/ORDINARY_TUI.md | 14 +++ experiments/codex_first_input/probe.py | 65 ++++++++-- tests/codex/test_first_input_probe.py | 118 ++++++++++++++++++ 3 files changed, 186 insertions(+), 11 deletions(-) create mode 100644 tests/codex/test_first_input_probe.py diff --git a/experiments/codex_first_input/ORDINARY_TUI.md b/experiments/codex_first_input/ORDINARY_TUI.md index 5509a5ab..9dfd7942 100644 --- a/experiments/codex_first_input/ORDINARY_TUI.md +++ b/experiments/codex_first_input/ORDINARY_TUI.md @@ -109,3 +109,17 @@ validation. Do not run it with Python optimization or against a shared server containing unrelated user work. The WebSocket dependency is test-only and is not added to the Python core runtime. `experiments/` remains excluded from release artifacts by the existing packaging rules. + +## Review-driven harness guards + +The supplied Codex home is now canonicalized before SQLite URI construction and +must match the absolute `codexHome` returned by `initialize` before any thread +lookup or database inspection. Missing or mismatched server-home evidence +refuses the probe without queuing. A queue RPC rejection returns exit 1; malformed +error objects or non-integer codes also fail safely without reflecting native +error text or retrying. The version guard remains exactly 0.162.1. + +Four synthetic WebSocket/temporary-database tests cover these boundaries, +including relative home paths and exactly one queue attempt on rejection. +These are harness regression tests, not new native delivery, receipt, platform, +or production-support evidence. SQLite handles are explicitly closed. diff --git a/experiments/codex_first_input/probe.py b/experiments/codex_first_input/probe.py index 802812a6..8ff9a7aa 100644 --- a/experiments/codex_first_input/probe.py +++ b/experiments/codex_first_input/probe.py @@ -4,6 +4,7 @@ Prints sanitized facts, not conversation IDs, paths, or native text. """ import argparse +from contextlib import closing import json import os from pathlib import Path @@ -16,16 +17,33 @@ from websockets.sync.client import unix_connect -def main(): +class ProbeError(Exception): + """Sanitized protocol/precondition failure, without native response text.""" + + +def rpc_result(response): + if "error" in response: + error = response["error"] + if not isinstance(error, dict) or type(error.get("code")) is not int: + raise ProbeError("invalid_rpc_error") + raise ProbeError("rpc_rejected") + result = response.get("result") + if not isinstance(result, dict): + raise ProbeError("invalid_rpc_result") + return result + + +def main(argv=None): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--socket", required=True) parser.add_argument("--codex-home", required=True) parser.add_argument("--expected-cwd", required=True) parser.add_argument("--mode", choices=("inspect", "queue"), default="inspect") parser.add_argument("--ack-marker", help="Harmless ASCII test receipt marker; no credentials") - args = parser.parse_args() + args = parser.parse_args(argv) if args.ack_marker is not None and not re.fullmatch(r"[A-Za-z0-9 _-]{1,100}", args.ack_marker): parser.error("invalid ACK marker") + codex_home = Path(args.codex_home).resolve(strict=True) socket = Path(args.socket) assert socket.lstat().st_uid == os.getuid() assert socket.parent.stat().st_mode & 0o077 == 0 @@ -42,40 +60,65 @@ def call(method, params): deadline = time.monotonic() + 15 for _ in range(200): message = json.loads(ws.recv(timeout=max(.01, deadline-time.monotonic()))) + if not isinstance(message, dict): + raise ProbeError("invalid_rpc_response") if message.get("id") == sequence: return message if "id" in message: raise RuntimeError("unexpected server interaction; no approval supplied") raise RuntimeError("notification limit") hello = call("initialize", {"clientInfo": {"name": "session-peer-first-input-probe", "version": "0"}, "capabilities": {"experimentalApi": True}}) - assert hello["result"]["userAgent"].startswith("codex-tui/0.162.1 ") + initialized = rpc_result(hello) + assert initialized["userAgent"].startswith("codex-tui/0.162.1 ") + server_home = initialized.get("codexHome") + if not isinstance(server_home, str) or not server_home or not Path(server_home).is_absolute(): + raise ProbeError("missing_or_invalid_server_home") + try: + bound_home = Path(server_home).resolve(strict=True) + except (OSError, ValueError): + raise ProbeError("missing_or_invalid_server_home") from None + if bound_home != codex_home: + raise ProbeError("server_home_mismatch") ws.send(json.dumps({"method": "initialized"})) - loaded = call("thread/loaded/list", {"limit": 100})["result"] + loaded = rpc_result(call("thread/loaded/list", {"limit": 100})) assert loaded.get("nextCursor") is None and len(loaded["data"]) == 1 thread_id = loaded["data"][0] - thread = call("thread/read", {"threadId": thread_id, "includeTurns": False})["result"]["thread"] + thread = rpc_result(call("thread/read", {"threadId": thread_id, "includeTurns": False}))["thread"] assert Path(thread["cwd"]).resolve() == Path(args.expected_cwd).resolve() assert thread["status"]["type"] == "idle" and thread["canAcceptDirectInput"] is True rows = 0 - dbs = list(Path(args.codex_home).glob("state_*.sqlite")) + dbs = list(codex_home.glob("state_*.sqlite")) assert dbs for db in dbs: - with sqlite3.connect(db.as_uri()+"?mode=ro", uri=True) as connection: + with closing(sqlite3.connect(db.as_uri()+"?mode=ro", uri=True)) as connection: rows += connection.execute("SELECT count(*) FROM threads WHERE id=?", (thread_id,)).fetchone()[0] facts = {"version": "0.162.1", "loadedThreads": 1, "storedRowsBefore": rows, "idle": True, "directInput": True, "ephemeral": thread.get("ephemeral"), "mode": args.mode} print(json.dumps(facts), flush=True) if args.mode == "inspect": - return + return 0 assert rows == 0, "requires fresh pre-first-input thread" marker = args.ack_marker or "ACK CODEX-FIRST-INPUT-306-" + args.mode.upper() message = "Owner-authorized one-shot transport test. Reply exactly " + marker + ". Do not run tools or modify files." method = "thread/queue/add" result = call(method, {"threadId": thread_id, "input": [{"type": "text", "text": message}], "clientUserMessageId": str(uuid.uuid4())}) if "error" in result: - print(json.dumps({"method": method, "accepted": False, "errorCode": result["error"].get("code"), "retryAllowed": False})) + error = result["error"] + if not isinstance(error, dict) or type(error.get("code")) is not int: + raise ProbeError("invalid_rpc_error") + print(json.dumps({"method": method, "accepted": False, "errorCode": error["code"], "retryAllowed": False}), flush=True) + return 1 else: - print(json.dumps({"method": method, "accepted": True, "resultKeys": sorted(result["result"]), "consumptionConfirmed": False}), flush=True) + print(json.dumps({"method": method, "accepted": True, "resultKeys": sorted(rpc_result(result)), "consumptionConfirmed": False}), flush=True) + return 0 + + +def run(argv=None): + try: + return main(argv) + except ProbeError as exc: + print(json.dumps({"accepted": False, "error": str(exc), "retryAllowed": False}), flush=True) + return 1 if __name__ == "__main__": - main() + raise SystemExit(run()) diff --git a/tests/codex/test_first_input_probe.py b/tests/codex/test_first_input_probe.py new file mode 100644 index 00000000..89734224 --- /dev/null +++ b/tests/codex/test_first_input_probe.py @@ -0,0 +1,118 @@ +"""Synthetic probe guards; no native server, model turn, or receipt validation.""" +import contextlib +import importlib.util +import io +import json +import os +from pathlib import Path +import socket +import sqlite3 +import sys +import tempfile +import types +import unittest +from unittest import mock + + +class FakeWebSocket: + def __init__(self, home, queue_response): + self.home = home + self.queue_response = queue_response + self.sent = [] + + def __enter__(self): + return self + + def __exit__(self, *unused): + return False + + def send(self, message): + self.sent.append(json.loads(message)) + + def recv(self, timeout): + request = self.sent[-1] + method = request["method"] + if method == "initialize": + result = {"userAgent": "codex-tui/0.162.1 test"} + if self.home is not None: + result["codexHome"] = self.home + elif method == "thread/loaded/list": + result = {"data": ["owned-thread"], "nextCursor": None} + elif method == "thread/read": + result = {"thread": {"cwd": self.home, "status": {"type": "idle"}, + "canAcceptDirectInput": True, "ephemeral": True}} + else: + return json.dumps(dict(self.queue_response, id=request["id"])) + return json.dumps({"id": request["id"], "result": result}) + + +@unittest.skipUnless(hasattr(socket, "AF_UNIX") and hasattr(os, "getuid"), "Unix experiment only") +class FirstInputProbeTests(unittest.TestCase): + @classmethod + def setUpClass(cls): + # Import with only the transport symbol stubbed: tests need no optional dependency. + client = types.ModuleType("websockets.sync.client") + client.unix_connect = lambda *args, **kwargs: None + spec = importlib.util.spec_from_file_location( + "first_input_probe", Path(__file__).resolve().parents[2] / + "experiments/codex_first_input/probe.py") + cls.probe = importlib.util.module_from_spec(spec) + with mock.patch.dict(sys.modules, {"websockets.sync.client": client}): + spec.loader.exec_module(cls.probe) + + def setUp(self): + self.scratch = tempfile.TemporaryDirectory() + self.addCleanup(self.scratch.cleanup) + self.home = Path(self.scratch.name).resolve() + self.endpoint = self.home / "probe.sock" + self.listener = socket.socket(socket.AF_UNIX) + self.listener.bind(str(self.endpoint)) + self.addCleanup(self.listener.close) + with contextlib.closing(sqlite3.connect(self.home / "state_5.sqlite")) as db: + db.execute("CREATE TABLE threads (id TEXT)") + db.commit() + + def invoke(self, server_home, queue_response=None, supplied_home=None): + ws = FakeWebSocket(server_home, queue_response or {"result": {"queuedSubmission": {}}}) + args = ["--socket", str(self.endpoint), "--codex-home", supplied_home or str(self.home), + "--expected-cwd", str(self.home), "--mode", "queue", "--ack-marker", "ACK TEST"] + output = io.StringIO() + with mock.patch.object(self.probe, "unix_connect", return_value=ws), contextlib.redirect_stdout(output): + exit_code = self.probe.run(args) + return exit_code, ws.sent, output.getvalue() + + def test_missing_or_mismatched_home_refuses_before_thread_or_database(self): + for server_home in (None, str(self.home.parent), "relative", ""): + with self.subTest(server_home=server_home), mock.patch.object( + self.probe.sqlite3, "connect", side_effect=AssertionError("DB must not open")): + exit_code, sent, output = self.invoke(server_home) + self.assertEqual(exit_code, 1) + self.assertEqual([item["method"] for item in sent], ["initialize"]) + self.assertNotIn(str(self.home), output) + + def test_relative_home_is_canonical_before_sqlite_uri(self): + relative = os.path.relpath(str(self.home), os.getcwd()) + exit_code, sent, output = self.invoke(str(self.home), supplied_home=relative) + self.assertEqual(exit_code, 0) + self.assertEqual(sum(item["method"] == "thread/queue/add" for item in sent), 1) + self.assertFalse(json.loads(output.splitlines()[-1])["consumptionConfirmed"]) + + def test_queue_rejection_returns_failure_without_retry(self): + exit_code, sent, output = self.invoke(str(self.home), {"error": {"code": -32603, "message": "PRIVATE"}}) + self.assertEqual(exit_code, 1) + self.assertEqual(sum(item["method"] == "thread/queue/add" for item in sent), 1) + self.assertEqual(json.loads(output.splitlines()[-1])["errorCode"], -32603) + self.assertNotIn("PRIVATE", output) + + def test_malformed_errors_fail_without_traceback_or_retry(self): + for error in (None, [], "PRIVATE", {"code": True}, {"code": "-32603"}, {}): + with self.subTest(error=error): + exit_code, sent, output = self.invoke(str(self.home), {"error": error}) + self.assertEqual(exit_code, 1) + self.assertEqual(sum(item["method"] == "thread/queue/add" for item in sent), 1) + self.assertEqual(json.loads(output.splitlines()[-1])["error"], "invalid_rpc_error") + self.assertNotIn("PRIVATE", output) + + +if __name__ == "__main__": + unittest.main() From dc807057dbf78d6e0bc719923272506f20eaa720 Mon Sep 17 00:00:00 2001 From: YEONGUK LEE Date: Tue, 13 Oct 2026 16:09:29 +0900 Subject: [PATCH 3/9] test(codex): preserve uncertain probe outcomes --- experiments/codex_first_input/ORDINARY_TUI.md | 24 ++++++-- experiments/codex_first_input/probe.py | 45 +++++++++++--- tests/codex/test_first_input_probe.py | 59 ++++++++++++++++++- 3 files changed, 113 insertions(+), 15 deletions(-) diff --git a/experiments/codex_first_input/ORDINARY_TUI.md b/experiments/codex_first_input/ORDINARY_TUI.md index 9dfd7942..70b7c256 100644 --- a/experiments/codex_first_input/ORDINARY_TUI.md +++ b/experiments/codex_first_input/ORDINARY_TUI.md @@ -72,7 +72,8 @@ direct PTY observation. It does not implement this transport in session-peer. An initial read-only stdio-proxy probe timed out before initialization, with no queue attempt. It was discarded; the successful inspection/submission used the existing Unix-WebSocket harness with websockets 17.0 in a temporary isolated -environment. The harness only gained an optional bounded ASCII test marker. +environment. At that live-test stage the harness only gained an optional bounded +ASCII test marker; subsequent review-driven guards are described below. No uncertain effect was retried. Both test TUIs and the foreground server exited zero; their PIDs were checked @@ -116,10 +117,21 @@ The supplied Codex home is now canonicalized before SQLite URI construction and must match the absolute `codexHome` returned by `initialize` before any thread lookup or database inspection. Missing or mismatched server-home evidence refuses the probe without queuing. A queue RPC rejection returns exit 1; malformed -error objects or non-integer codes also fail safely without reflecting native -error text or retrying. The version guard remains exactly 0.162.1. - -Four synthetic WebSocket/temporary-database tests cover these boundaries, +error objects or non-integer codes yield an unknown outcome once queue submission +has been attempted, without reflecting native error text or retrying. Timeout, +invalid JSON, invalid results, and transport failures after that boundary also +report `accepted:null`, `outcome:unknown`, and `retryAllowed:false`; after a queue +attempt, a validated RPC rejection alone supplies `accepted:false`. Runtime +preflight failures report `accepted:false`, `outcome:not_attempted`; argparse +usage errors retain exit 2 and usage output. An interrupted probe returns exit 130 while preserving the +same effect boundary. `ephemeral` is restricted to boolean/null; arbitrary result +keys and native error details are not reflected. The version guard remains +exactly 0.162.1, and Python optimization is refused up front. + +Seven synthetic WebSocket/temporary-database tests cover these boundaries, including relative home paths and exactly one queue attempt on rejection. These are harness regression tests, not new native delivery, receipt, platform, -or production-support evidence. SQLite handles are explicitly closed. +or production-support evidence. None of these review-driven guards were exercised +by another live native run. SQLite handles are explicitly closed. A failure during +connection cleanup may return a nonzero exit while preserving an already observed +acceptance or rejection; it never authorizes retry. diff --git a/experiments/codex_first_input/probe.py b/experiments/codex_first_input/probe.py index 8ff9a7aa..d16c1393 100644 --- a/experiments/codex_first_input/probe.py +++ b/experiments/codex_first_input/probe.py @@ -33,7 +33,11 @@ def rpc_result(response): return result -def main(argv=None): +def main(argv=None, state=None): + if state is None: + state = {"attempted": False, "accepted": None} + if not __debug__: + raise ProbeError("optimization_not_supported") parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--socket", required=True) parser.add_argument("--codex-home", required=True) @@ -56,13 +60,19 @@ def main(argv=None): def call(method, params): nonlocal sequence sequence += 1 - ws.send(json.dumps({"id": sequence, "method": method, "params": params})) + request = json.dumps({"id": sequence, "method": method, "params": params}) + if method == "thread/queue/add": + # An exception from send cannot prove that no bytes reached the server. + state["attempted"] = True + ws.send(request) deadline = time.monotonic() + 15 for _ in range(200): message = json.loads(ws.recv(timeout=max(.01, deadline-time.monotonic()))) if not isinstance(message, dict): raise ProbeError("invalid_rpc_response") if message.get("id") == sequence: + if ("error" in message) == ("result" in message): + raise ProbeError("invalid_rpc_response") return message if "id" in message: raise RuntimeError("unexpected server interaction; no approval supplied") @@ -86,13 +96,16 @@ def call(method, params): thread = rpc_result(call("thread/read", {"threadId": thread_id, "includeTurns": False}))["thread"] assert Path(thread["cwd"]).resolve() == Path(args.expected_cwd).resolve() assert thread["status"]["type"] == "idle" and thread["canAcceptDirectInput"] is True + ephemeral = thread.get("ephemeral") + if ephemeral is not None and type(ephemeral) is not bool: + raise ProbeError("invalid_thread_metadata") rows = 0 dbs = list(codex_home.glob("state_*.sqlite")) assert dbs for db in dbs: with closing(sqlite3.connect(db.as_uri()+"?mode=ro", uri=True)) as connection: rows += connection.execute("SELECT count(*) FROM threads WHERE id=?", (thread_id,)).fetchone()[0] - facts = {"version": "0.162.1", "loadedThreads": 1, "storedRowsBefore": rows, "idle": True, "directInput": True, "ephemeral": thread.get("ephemeral"), "mode": args.mode} + facts = {"version": "0.162.1", "loadedThreads": 1, "storedRowsBefore": rows, "idle": True, "directInput": True, "ephemeral": ephemeral, "mode": args.mode} print(json.dumps(facts), flush=True) if args.mode == "inspect": return 0 @@ -105,20 +118,38 @@ def call(method, params): error = result["error"] if not isinstance(error, dict) or type(error.get("code")) is not int: raise ProbeError("invalid_rpc_error") + state["accepted"] = False print(json.dumps({"method": method, "accepted": False, "errorCode": error["code"], "retryAllowed": False}), flush=True) return 1 else: - print(json.dumps({"method": method, "accepted": True, "resultKeys": sorted(rpc_result(result)), "consumptionConfirmed": False}), flush=True) + queued = rpc_result(result) + if "queuedSubmission" not in queued: + raise ProbeError("invalid_queue_result") + state["accepted"] = True + print(json.dumps({"method": method, "accepted": True, "consumptionConfirmed": False}), flush=True) return 0 def run(argv=None): + state = {"attempted": False, "accepted": None} try: - return main(argv) - except ProbeError as exc: - print(json.dumps({"accepted": False, "error": str(exc), "retryAllowed": False}), flush=True) + return main(argv, state) + except KeyboardInterrupt: + report_failure(state, "interrupted") + return 130 + except Exception as exc: + # Do not reflect paths, server messages, JSON fragments, or exception text. + report_failure(state, str(exc) if isinstance(exc, ProbeError) else "probe_failed") return 1 +def report_failure(state, error): + accepted = state["accepted"] if state["attempted"] else False + outcome = ("unknown" if accepted is None else + "accepted" if accepted else "rejected" if state["attempted"] else "not_attempted") + print(json.dumps({"accepted": accepted, "outcome": outcome, "error": error, + "retryAllowed": False}), flush=True) + + if __name__ == "__main__": raise SystemExit(run()) diff --git a/tests/codex/test_first_input_probe.py b/tests/codex/test_first_input_probe.py index 89734224..cb0af8a4 100644 --- a/tests/codex/test_first_input_probe.py +++ b/tests/codex/test_first_input_probe.py @@ -28,6 +28,8 @@ def __exit__(self, *unused): def send(self, message): self.sent.append(json.loads(message)) + if self.sent[-1]["method"] == "thread/queue/add" and self.queue_response == "send-error": + raise OSError("PRIVATE send details") def recv(self, timeout): request = self.sent[-1] @@ -42,6 +44,10 @@ def recv(self, timeout): result = {"thread": {"cwd": self.home, "status": {"type": "idle"}, "canAcceptDirectInput": True, "ephemeral": True}} else: + if isinstance(self.queue_response, BaseException): + raise self.queue_response + if isinstance(self.queue_response, str): + return self.queue_response return json.dumps(dict(self.queue_response, id=request["id"])) return json.dumps({"id": request["id"], "result": result}) @@ -73,7 +79,7 @@ def setUp(self): db.commit() def invoke(self, server_home, queue_response=None, supplied_home=None): - ws = FakeWebSocket(server_home, queue_response or {"result": {"queuedSubmission": {}}}) + ws = FakeWebSocket(server_home, {"result": {"queuedSubmission": {}}} if queue_response is None else queue_response) args = ["--socket", str(self.endpoint), "--codex-home", supplied_home or str(self.home), "--expected-cwd", str(self.home), "--mode", "queue", "--ack-marker", "ACK TEST"] output = io.StringIO() @@ -84,8 +90,9 @@ def invoke(self, server_home, queue_response=None, supplied_home=None): def test_missing_or_mismatched_home_refuses_before_thread_or_database(self): for server_home in (None, str(self.home.parent), "relative", ""): with self.subTest(server_home=server_home), mock.patch.object( - self.probe.sqlite3, "connect", side_effect=AssertionError("DB must not open")): + self.probe.sqlite3, "connect", side_effect=AssertionError("DB must not open")) as database: exit_code, sent, output = self.invoke(server_home) + database.assert_not_called() self.assertEqual(exit_code, 1) self.assertEqual([item["method"] for item in sent], ["initialize"]) self.assertNotIn(str(self.home), output) @@ -111,8 +118,56 @@ def test_malformed_errors_fail_without_traceback_or_retry(self): self.assertEqual(exit_code, 1) self.assertEqual(sum(item["method"] == "thread/queue/add" for item in sent), 1) self.assertEqual(json.loads(output.splitlines()[-1])["error"], "invalid_rpc_error") + self.assertIsNone(json.loads(output.splitlines()[-1])["accepted"]) + self.assertEqual(json.loads(output.splitlines()[-1])["outcome"], "unknown") self.assertNotIn("PRIVATE", output) + def test_uncertain_queue_failures_preserve_unknown_and_no_retry(self): + for response in (TimeoutError("PRIVATE timeout"), OSError("PRIVATE transport"), + "send-error", "PRIVATE invalid JSON", {"result": []}, + {"result": {}}, {"error": {"code": -1}, "result": {}}, + KeyboardInterrupt()): + with self.subTest(response=type(response).__name__): + exit_code, sent, output = self.invoke(str(self.home), response) + self.assertEqual(exit_code, 130 if isinstance(response, KeyboardInterrupt) else 1) + self.assertEqual(sum(item["method"] == "thread/queue/add" for item in sent), 1) + failure = json.loads(output.splitlines()[-1]) + self.assertIsNone(failure["accepted"]) + self.assertEqual(failure["outcome"], "unknown") + self.assertFalse(failure["retryAllowed"]) + self.assertNotIn("PRIVATE", output) + self.assertNotIn(str(self.home), output) + + def test_preflight_exception_is_sanitized_without_queue(self): + for failure in (FileNotFoundError("PRIVATE path"), sqlite3.DatabaseError("PRIVATE DB"), + AssertionError("PRIVATE assertion"), ValueError("PRIVATE value")): + with self.subTest(failure=type(failure).__name__), mock.patch.object( + self.probe.sqlite3, "connect", side_effect=failure): + exit_code, sent, output = self.invoke(str(self.home)) + self.assertEqual(exit_code, 1) + self.assertFalse(any(item["method"] == "thread/queue/add" for item in sent)) + result = json.loads(output.splitlines()[-1]) + self.assertIs(result["accepted"], False) + self.assertEqual(result["outcome"], "not_attempted") + self.assertNotIn("PRIVATE", output) + + def test_invalid_ephemeral_is_not_reflected_or_submitted(self): + original = FakeWebSocket.recv + + def malicious_metadata(ws, timeout): + raw = original(ws, timeout) + if ws.sent[-1]["method"] == "thread/read": + value = json.loads(raw) + value["result"]["thread"]["ephemeral"] = "PRIVATE" + return json.dumps(value) + return raw + + with mock.patch.object(FakeWebSocket, "recv", malicious_metadata): + exit_code, sent, output = self.invoke(str(self.home)) + self.assertEqual(exit_code, 1) + self.assertFalse(any(item["method"] == "thread/queue/add" for item in sent)) + self.assertNotIn("PRIVATE", output) + if __name__ == "__main__": unittest.main() From 08c26eea620ccbc856b22e62764d6c3346a24de8 Mon Sep 17 00:00:00 2001 From: YEONGUK LEE Date: Thu, 15 Oct 2026 13:39:17 +0900 Subject: [PATCH 4/9] fix(experiment): validate RPC rejections and require persistent Codex threads --- experiments/codex_first_input/ORDINARY_TUI.md | 15 +++-- experiments/codex_first_input/probe.py | 21 ++++--- tests/codex/test_first_input_probe.py | 62 +++++++++++++++++-- 3 files changed, 82 insertions(+), 16 deletions(-) diff --git a/experiments/codex_first_input/ORDINARY_TUI.md b/experiments/codex_first_input/ORDINARY_TUI.md index 70b7c256..f5937535 100644 --- a/experiments/codex_first_input/ORDINARY_TUI.md +++ b/experiments/codex_first_input/ORDINARY_TUI.md @@ -117,19 +117,24 @@ The supplied Codex home is now canonicalized before SQLite URI construction and must match the absolute `codexHome` returned by `initialize` before any thread lookup or database inspection. Missing or mismatched server-home evidence refuses the probe without queuing. A queue RPC rejection returns exit 1; malformed -error objects or non-integer codes yield an unknown outcome once queue submission +error objects, non-integer codes or missing/non-string messages yield an unknown outcome once queue submission has been attempted, without reflecting native error text or retrying. Timeout, invalid JSON, invalid results, and transport failures after that boundary also report `accepted:null`, `outcome:unknown`, and `retryAllowed:false`; after a queue attempt, a validated RPC rejection alone supplies `accepted:false`. Runtime preflight failures report `accepted:false`, `outcome:not_attempted`; argparse usage errors retain exit 2 and usage output. An interrupted probe returns exit 130 while preserving the -same effect boundary. `ephemeral` is restricted to boolean/null; arbitrary result -keys and native error details are not reflected. The version guard remains +same effect boundary. Queue mode requires explicit `ephemeral:false` before +database inspection or submission: a previously used ephemeral conversation may +still have no database row. True, null and missing values cannot establish this +precondition. Inspection mode may report boolean/null metadata without queuing; +that does not prove the conversation is unused. Arbitrary result keys and native +error details are not reflected. The version guard remains exactly 0.162.1, and Python optimization is refused up front. -Seven synthetic WebSocket/temporary-database tests cover these boundaries, -including relative home paths and exactly one queue attempt on rejection. +Ten synthetic WebSocket/temporary-database tests cover these boundaries, +including relative home paths, complete error validation on both RPC paths, +ephemeral/missing/null refusal and exactly one queue attempt on rejection. These are harness regression tests, not new native delivery, receipt, platform, or production-support evidence. None of these review-driven guards were exercised by another live native run. SQLite handles are explicitly closed. A failure during diff --git a/experiments/codex_first_input/probe.py b/experiments/codex_first_input/probe.py index d16c1393..75c87dd6 100644 --- a/experiments/codex_first_input/probe.py +++ b/experiments/codex_first_input/probe.py @@ -21,11 +21,16 @@ class ProbeError(Exception): """Sanitized protocol/precondition failure, without native response text.""" +def rpc_error_code(error): + if (not isinstance(error, dict) or type(error.get("code")) is not int + or not isinstance(error.get("message"), str)): + raise ProbeError("invalid_rpc_error") + return error["code"] + + def rpc_result(response): if "error" in response: - error = response["error"] - if not isinstance(error, dict) or type(error.get("code")) is not int: - raise ProbeError("invalid_rpc_error") + rpc_error_code(response["error"]) raise ProbeError("rpc_rejected") result = response.get("result") if not isinstance(result, dict): @@ -99,6 +104,10 @@ def call(method, params): ephemeral = thread.get("ephemeral") if ephemeral is not None and type(ephemeral) is not bool: raise ProbeError("invalid_thread_metadata") + if args.mode == "queue" and ephemeral is not False: + # Ephemeral conversations can have prior turns without any stored + # row. Missing/null metadata cannot establish persistence either. + raise ProbeError("persistent_thread_required") rows = 0 dbs = list(codex_home.glob("state_*.sqlite")) assert dbs @@ -115,11 +124,9 @@ def call(method, params): method = "thread/queue/add" result = call(method, {"threadId": thread_id, "input": [{"type": "text", "text": message}], "clientUserMessageId": str(uuid.uuid4())}) if "error" in result: - error = result["error"] - if not isinstance(error, dict) or type(error.get("code")) is not int: - raise ProbeError("invalid_rpc_error") + code = rpc_error_code(result["error"]) state["accepted"] = False - print(json.dumps({"method": method, "accepted": False, "errorCode": error["code"], "retryAllowed": False}), flush=True) + print(json.dumps({"method": method, "accepted": False, "errorCode": code, "retryAllowed": False}), flush=True) return 1 else: queued = rpc_result(result) diff --git a/tests/codex/test_first_input_probe.py b/tests/codex/test_first_input_probe.py index cb0af8a4..c0810c5c 100644 --- a/tests/codex/test_first_input_probe.py +++ b/tests/codex/test_first_input_probe.py @@ -42,7 +42,7 @@ def recv(self, timeout): result = {"data": ["owned-thread"], "nextCursor": None} elif method == "thread/read": result = {"thread": {"cwd": self.home, "status": {"type": "idle"}, - "canAcceptDirectInput": True, "ephemeral": True}} + "canAcceptDirectInput": True, "ephemeral": False}} else: if isinstance(self.queue_response, BaseException): raise self.queue_response @@ -78,10 +78,10 @@ def setUp(self): db.execute("CREATE TABLE threads (id TEXT)") db.commit() - def invoke(self, server_home, queue_response=None, supplied_home=None): + def invoke(self, server_home, queue_response=None, supplied_home=None, mode="queue"): ws = FakeWebSocket(server_home, {"result": {"queuedSubmission": {}}} if queue_response is None else queue_response) args = ["--socket", str(self.endpoint), "--codex-home", supplied_home or str(self.home), - "--expected-cwd", str(self.home), "--mode", "queue", "--ack-marker", "ACK TEST"] + "--expected-cwd", str(self.home), "--mode", mode, "--ack-marker", "ACK TEST"] output = io.StringIO() with mock.patch.object(self.probe, "unix_connect", return_value=ws), contextlib.redirect_stdout(output): exit_code = self.probe.run(args) @@ -112,7 +112,10 @@ def test_queue_rejection_returns_failure_without_retry(self): self.assertNotIn("PRIVATE", output) def test_malformed_errors_fail_without_traceback_or_retry(self): - for error in (None, [], "PRIVATE", {"code": True}, {"code": "-32603"}, {}): + for error in (None, [], "PRIVATE", {"code": True, "message": "PRIVATE"}, + {"code": "-32603", "message": "PRIVATE"}, {}, {"code": -32603}, + {"code": -32603, "message": None}, {"code": -32603, "message": 1}, + {"code": -32603, "message": []}): with self.subTest(error=error): exit_code, sent, output = self.invoke(str(self.home), {"error": error}) self.assertEqual(exit_code, 1) @@ -120,8 +123,59 @@ def test_malformed_errors_fail_without_traceback_or_retry(self): self.assertEqual(json.loads(output.splitlines()[-1])["error"], "invalid_rpc_error") self.assertIsNone(json.loads(output.splitlines()[-1])["accepted"]) self.assertEqual(json.loads(output.splitlines()[-1])["outcome"], "unknown") + self.assertFalse(json.loads(output.splitlines()[-1])["retryAllowed"]) self.assertNotIn("PRIVATE", output) + def test_rpc_result_uses_the_same_complete_error_validation(self): + for error in ({"code": -1}, {"code": -1, "message": None}, + {"code": -1, "message": 1}, {"code": True, "message": "PRIVATE"}): + with self.subTest(error=error), self.assertRaises(self.probe.ProbeError) as caught: + self.probe.rpc_result({"error": error}) + self.assertEqual(str(caught.exception), "invalid_rpc_error") + for message in ("", "PRIVATE native text"): + with self.subTest(empty=not message), self.assertRaises(self.probe.ProbeError) as caught: + self.probe.rpc_result({"error": {"code": -1, "message": message}}) + self.assertEqual(str(caught.exception), "rpc_rejected") + + def test_ephemeral_or_unproven_threads_refuse_before_database_or_queue(self): + original = FakeWebSocket.recv + for label, fields in (("ephemeral", {"ephemeral": True}), ("null", {"ephemeral": None}), + ("missing", {})): + def metadata(ws, timeout): + raw = original(ws, timeout) + if ws.sent[-1]["method"] == "thread/read": + value = json.loads(raw) + value["result"]["thread"].pop("ephemeral", None) + value["result"]["thread"].update(fields) + return json.dumps(value) + return raw + with self.subTest(case=label), mock.patch.object(FakeWebSocket, "recv", metadata), \ + mock.patch.object(self.probe.sqlite3, "connect", side_effect=AssertionError("DB must not open")) as db: + exit_code, sent, output = self.invoke(str(self.home)) + self.assertEqual(exit_code, 1) + self.assertEqual(db.call_count, 0) + self.assertFalse(any(item["method"] == "thread/queue/add" for item in sent)) + result = json.loads(output.splitlines()[-1]) + self.assertEqual(result["error"], "persistent_thread_required") + self.assertIs(result["accepted"], False) + self.assertEqual(result["outcome"], "not_attempted") + self.assertFalse(result["retryAllowed"]) + + def test_inspection_does_not_claim_row_absence_proves_ephemeral_freshness(self): + original = FakeWebSocket.recv + def metadata(ws, timeout): + raw = original(ws, timeout) + if ws.sent[-1]["method"] == "thread/read": + value = json.loads(raw) + value["result"]["thread"]["ephemeral"] = True + return json.dumps(value) + return raw + with mock.patch.object(FakeWebSocket, "recv", metadata): + exit_code, sent, output = self.invoke(str(self.home), mode="inspect") + self.assertEqual(exit_code, 0) + self.assertTrue(json.loads(output.splitlines()[-1])["ephemeral"]) + self.assertFalse(any(item["method"] == "thread/queue/add" for item in sent)) + def test_uncertain_queue_failures_preserve_unknown_and_no_retry(self): for response in (TimeoutError("PRIVATE timeout"), OSError("PRIVATE transport"), "send-error", "PRIVATE invalid JSON", {"result": []}, From 5c80632b28c4e43f2afe68591c75d6a8d76b5c06 Mon Sep 17 00:00:00 2001 From: YEONGUK LEE Date: Thu, 15 Oct 2026 13:41:22 +0900 Subject: [PATCH 5/9] test(experiment): cover inspect metadata and document pinned schema --- experiments/codex_first_input/ORDINARY_TUI.md | 4 ++ tests/codex/test_first_input_probe.py | 51 +++++++++++-------- 2 files changed, 35 insertions(+), 20 deletions(-) diff --git a/experiments/codex_first_input/ORDINARY_TUI.md b/experiments/codex_first_input/ORDINARY_TUI.md index f5937535..2a7571c6 100644 --- a/experiments/codex_first_input/ORDINARY_TUI.md +++ b/experiments/codex_first_input/ORDINARY_TUI.md @@ -132,6 +132,10 @@ that does not prove the conversation is unused. Arbitrary result keys and native error details are not reflected. The version guard remains exactly 0.162.1, and Python optimization is refused up front. +The pinned [0.162.1 Thread schema](https://github.com/openai/codex/blob/092d3acd6bec3e3a14bdc7e7a2810ab628ab759d/codex-rs/app-server-protocol/src/protocol/v2/thread_data.rs#L224-L225) +declares `ephemeral` as a bool without an omission attribute. This is a static +source basis for requiring explicit false, not a fresh native serialization test. + Ten synthetic WebSocket/temporary-database tests cover these boundaries, including relative home paths, complete error validation on both RPC paths, ephemeral/missing/null refusal and exactly one queue attempt on rejection. diff --git a/tests/codex/test_first_input_probe.py b/tests/codex/test_first_input_probe.py index c0810c5c..823da606 100644 --- a/tests/codex/test_first_input_probe.py +++ b/tests/codex/test_first_input_probe.py @@ -109,6 +109,8 @@ def test_queue_rejection_returns_failure_without_retry(self): self.assertEqual(exit_code, 1) self.assertEqual(sum(item["method"] == "thread/queue/add" for item in sent), 1) self.assertEqual(json.loads(output.splitlines()[-1])["errorCode"], -32603) + self.assertIs(json.loads(output.splitlines()[-1])["accepted"], False) + self.assertFalse(json.loads(output.splitlines()[-1])["retryAllowed"]) self.assertNotIn("PRIVATE", output) def test_malformed_errors_fail_without_traceback_or_retry(self): @@ -152,29 +154,38 @@ def metadata(ws, timeout): with self.subTest(case=label), mock.patch.object(FakeWebSocket, "recv", metadata), \ mock.patch.object(self.probe.sqlite3, "connect", side_effect=AssertionError("DB must not open")) as db: exit_code, sent, output = self.invoke(str(self.home)) - self.assertEqual(exit_code, 1) - self.assertEqual(db.call_count, 0) - self.assertFalse(any(item["method"] == "thread/queue/add" for item in sent)) - result = json.loads(output.splitlines()[-1]) - self.assertEqual(result["error"], "persistent_thread_required") - self.assertIs(result["accepted"], False) - self.assertEqual(result["outcome"], "not_attempted") - self.assertFalse(result["retryAllowed"]) + self.assertEqual(exit_code, 1) + self.assertEqual(db.call_count, 0) + self.assertFalse(any(item["method"] == "thread/queue/add" for item in sent)) + result = json.loads(output.splitlines()[-1]) + self.assertEqual(result["error"], "persistent_thread_required") + self.assertIs(result["accepted"], False) + self.assertEqual(result["outcome"], "not_attempted") + self.assertFalse(result["retryAllowed"]) def test_inspection_does_not_claim_row_absence_proves_ephemeral_freshness(self): original = FakeWebSocket.recv - def metadata(ws, timeout): - raw = original(ws, timeout) - if ws.sent[-1]["method"] == "thread/read": - value = json.loads(raw) - value["result"]["thread"]["ephemeral"] = True - return json.dumps(value) - return raw - with mock.patch.object(FakeWebSocket, "recv", metadata): - exit_code, sent, output = self.invoke(str(self.home), mode="inspect") - self.assertEqual(exit_code, 0) - self.assertTrue(json.loads(output.splitlines()[-1])["ephemeral"]) - self.assertFalse(any(item["method"] == "thread/queue/add" for item in sent)) + for label, fields in (("ephemeral", {"ephemeral": True}), ("null", {"ephemeral": None}), + ("missing", {}), ("invalid", {"ephemeral": "PRIVATE"})): + def metadata(ws, timeout): + raw = original(ws, timeout) + if ws.sent[-1]["method"] == "thread/read": + value = json.loads(raw) + value["result"]["thread"].pop("ephemeral", None) + value["result"]["thread"].update(fields) + return json.dumps(value) + return raw + with self.subTest(case=label), mock.patch.object(FakeWebSocket, "recv", metadata): + exit_code, sent, output = self.invoke(str(self.home), mode="inspect") + result = json.loads(output.splitlines()[-1]) + if label == "invalid": + self.assertEqual(exit_code, 1) + self.assertEqual(result["error"], "invalid_thread_metadata") + self.assertNotIn("PRIVATE", output) + else: + self.assertEqual(exit_code, 0) + self.assertIs(result["ephemeral"], fields.get("ephemeral")) + self.assertFalse(any(item["method"] == "thread/queue/add" for item in sent)) def test_uncertain_queue_failures_preserve_unknown_and_no_retry(self): for response in (TimeoutError("PRIVATE timeout"), OSError("PRIVATE transport"), From ad8fe10fc8a2562430cdb2838b82b351294e73e1 Mon Sep 17 00:00:00 2001 From: YEONGUK LEE Date: Thu, 15 Oct 2026 13:51:24 +0900 Subject: [PATCH 6/9] fix(experiment): verify queue emptiness and correlate accepted submissions --- experiments/codex_first_input/ORDINARY_TUI.md | 27 +++- experiments/codex_first_input/probe.py | 30 +++- tests/codex/test_first_input_probe.py | 128 ++++++++++++++++-- 3 files changed, 170 insertions(+), 15 deletions(-) diff --git a/experiments/codex_first_input/ORDINARY_TUI.md b/experiments/codex_first_input/ORDINARY_TUI.md index 2a7571c6..1a0503fd 100644 --- a/experiments/codex_first_input/ORDINARY_TUI.md +++ b/experiments/codex_first_input/ORDINARY_TUI.md @@ -136,9 +136,32 @@ The pinned [0.162.1 Thread schema](https://github.com/openai/codex/blob/092d3acd declares `ephemeral` as a bool without an omission attribute. This is a static source basis for requiring explicit false, not a fresh native serialization test. -Ten synthetic WebSocket/temporary-database tests cover these boundaries, +Immediately before the one queue attempt, the probe requests the selected +thread's `thread/queue/list` first page with `limit:1`. Only an explicit empty +`data` array and `nextCursor:null` pass. Missing, malformed, nonempty or paginated +responses refuse before any queue submission. This is a point-in-time check, +not an atomic expected-empty precondition: the native add API exposes no such +precondition. The exclusively owned server and no-other-input-producer setup +remain mandatory. No preflight can establish first-input status if another +producer races it. The queue-list API can return queued input bodies; this +harness neither prints nor inspects them beyond refusing a nonempty list, but +the parsed response exists in memory. It is not a body-free observation API and +must not be used on a server containing unrelated work. + +A successful queue reply must contain a structured `queuedSubmission` with a +nonempty string ID and the exact submitted plain-text `input` and +`clientUserMessageId`. The explicit `text_elements:[]` matches the pinned native +text schema. Missing, null, malformed or mismatched fields after an attempt +remain unknown with no retry; merely finding the key is not acceptance. RPC +response IDs must match the exact integer request ID, not a bool or float. +See the pinned [queue request/response schema](https://github.com/openai/codex/blob/092d3acd6bec3e3a14bdc7e7a2810ab628ab759d/codex-rs/app-server-protocol/src/protocol/v2/thread.rs#L913-L959) +and [plain-text input schema](https://github.com/openai/codex/blob/092d3acd6bec3e3a14bdc7e7a2810ab628ab759d/codex-rs/app-server-protocol/src/protocol/v2/turn.rs#L436-L445). +These checks were not retrospectively performed in the historical live runs. + +Fifteen synthetic WebSocket/temporary-database tests cover these boundaries, including relative home paths, complete error validation on both RPC paths, -ephemeral/missing/null refusal and exactly one queue attempt on rejection. +ephemeral/missing/null refusal, empty-queue preflight, submission echo validation, +exact RPC IDs and exactly one queue attempt on rejection or uncertain failure. These are harness regression tests, not new native delivery, receipt, platform, or production-support evidence. None of these review-driven guards were exercised by another live native run. SQLite handles are explicitly closed. A failure during diff --git a/experiments/codex_first_input/probe.py b/experiments/codex_first_input/probe.py index 75c87dd6..fbeb500f 100644 --- a/experiments/codex_first_input/probe.py +++ b/experiments/codex_first_input/probe.py @@ -38,6 +38,23 @@ def rpc_result(response): return result +def require_empty_queue(result): + if (not isinstance(result.get("data"), list) or "nextCursor" not in result + or (result["nextCursor"] is not None and not isinstance(result["nextCursor"], str))): + raise ProbeError("invalid_queue_snapshot") + if result["data"] or result["nextCursor"] is not None: + raise ProbeError("queue_not_empty_or_unproven") + + +def validate_queued_submission(result, expected_input, client_message_id): + queued = result.get("queuedSubmission") + if (not isinstance(queued, dict) or not isinstance(queued.get("id"), str) or not queued["id"] + or not isinstance(queued.get("input"), list) or queued["input"] != expected_input + or not isinstance(queued.get("clientUserMessageId"), str) + or queued["clientUserMessageId"] != client_message_id): + raise ProbeError("invalid_queue_result") + + def main(argv=None, state=None): if state is None: state = {"attempted": False, "accepted": None} @@ -75,7 +92,7 @@ def call(method, params): message = json.loads(ws.recv(timeout=max(.01, deadline-time.monotonic()))) if not isinstance(message, dict): raise ProbeError("invalid_rpc_response") - if message.get("id") == sequence: + if type(message.get("id")) is int and message["id"] == sequence: if ("error" in message) == ("result" in message): raise ProbeError("invalid_rpc_response") return message @@ -121,17 +138,20 @@ def call(method, params): assert rows == 0, "requires fresh pre-first-input thread" marker = args.ack_marker or "ACK CODEX-FIRST-INPUT-306-" + args.mode.upper() message = "Owner-authorized one-shot transport test. Reply exactly " + marker + ". Do not run tools or modify files." + # A point-in-time preflight, not an atomic expected-empty condition. + # Only use the exclusively owned server with no other input producers. + require_empty_queue(rpc_result(call("thread/queue/list", {"threadId": thread_id, "limit": 1}))) + expected_input = [{"type": "text", "text": message, "text_elements": []}] + client_message_id = str(uuid.uuid4()) method = "thread/queue/add" - result = call(method, {"threadId": thread_id, "input": [{"type": "text", "text": message}], "clientUserMessageId": str(uuid.uuid4())}) + result = call(method, {"threadId": thread_id, "input": expected_input, "clientUserMessageId": client_message_id}) if "error" in result: code = rpc_error_code(result["error"]) state["accepted"] = False print(json.dumps({"method": method, "accepted": False, "errorCode": code, "retryAllowed": False}), flush=True) return 1 else: - queued = rpc_result(result) - if "queuedSubmission" not in queued: - raise ProbeError("invalid_queue_result") + validate_queued_submission(rpc_result(result), expected_input, client_message_id) state["accepted"] = True print(json.dumps({"method": method, "accepted": True, "consumptionConfirmed": False}), flush=True) return 0 diff --git a/tests/codex/test_first_input_probe.py b/tests/codex/test_first_input_probe.py index 823da606..7aefb783 100644 --- a/tests/codex/test_first_input_probe.py +++ b/tests/codex/test_first_input_probe.py @@ -14,10 +14,17 @@ from unittest import mock +def successful_queue_reply(request): + return {"result": {"queuedSubmission": { + "id": "fixture-submission", "input": request["params"]["input"], + "clientUserMessageId": request["params"]["clientUserMessageId"]}}} + + class FakeWebSocket: - def __init__(self, home, queue_response): + def __init__(self, home, queue_response, queue_list_response=None): self.home = home self.queue_response = queue_response + self.queue_list_response = queue_list_response self.sent = [] def __enter__(self): @@ -43,14 +50,25 @@ def recv(self, timeout): elif method == "thread/read": result = {"thread": {"cwd": self.home, "status": {"type": "idle"}, "canAcceptDirectInput": True, "ephemeral": False}} + elif method == "thread/queue/list": + response = ({"result": {"data": [], "nextCursor": None}} if self.queue_list_response is None + else self.queue_list_response) + return self.encode_reply(request, response) else: - if isinstance(self.queue_response, BaseException): - raise self.queue_response - if isinstance(self.queue_response, str): - return self.queue_response - return json.dumps(dict(self.queue_response, id=request["id"])) + response = successful_queue_reply(request) if self.queue_response is None else self.queue_response + return self.encode_reply(request, response) return json.dumps({"id": request["id"], "result": result}) + @staticmethod + def encode_reply(request, response): + if callable(response): + response = response(request) + if isinstance(response, BaseException): + raise response + if isinstance(response, str): + return response + return json.dumps(dict(response, id=request["id"])) + @unittest.skipUnless(hasattr(socket, "AF_UNIX") and hasattr(os, "getuid"), "Unix experiment only") class FirstInputProbeTests(unittest.TestCase): @@ -78,8 +96,8 @@ def setUp(self): db.execute("CREATE TABLE threads (id TEXT)") db.commit() - def invoke(self, server_home, queue_response=None, supplied_home=None, mode="queue"): - ws = FakeWebSocket(server_home, {"result": {"queuedSubmission": {}}} if queue_response is None else queue_response) + def invoke(self, server_home, queue_response=None, supplied_home=None, mode="queue", queue_list_response=None): + ws = FakeWebSocket(server_home, queue_response, queue_list_response) args = ["--socket", str(self.endpoint), "--codex-home", supplied_home or str(self.home), "--expected-cwd", str(self.home), "--mode", mode, "--ack-marker", "ACK TEST"] output = io.StringIO() @@ -113,6 +131,100 @@ def test_queue_rejection_returns_failure_without_retry(self): self.assertFalse(json.loads(output.splitlines()[-1])["retryAllowed"]) self.assertNotIn("PRIVATE", output) + def test_queue_preflight_is_bound_and_immediately_precedes_the_only_add(self): + exit_code, sent, output = self.invoke(str(self.home)) + self.assertEqual(exit_code, 0) + self.assertEqual([item["method"] for item in sent][-2:], ["thread/queue/list", "thread/queue/add"]) + listed, added = sent[-2:] + self.assertEqual(listed["params"], {"threadId": "owned-thread", "limit": 1}) + self.assertEqual(added["params"]["threadId"], listed["params"]["threadId"]) + self.assertEqual(added["params"]["input"][0]["text_elements"], []) + self.assertEqual(sum(item["method"] == "thread/queue/add" for item in sent), 1) + self.assertTrue(json.loads(output.splitlines()[-1])["accepted"]) + self.assertFalse(json.loads(output.splitlines()[-1])["consumptionConfirmed"]) + self.assertTrue(added["params"]["clientUserMessageId"] not in output, "client ID reflected") + self.assertTrue("fixture-submission" not in output, "queue ID reflected") + + def test_nonempty_or_unproven_queue_refuses_without_any_add(self): + results = [None, [], {}, {"data": []}, {"data": None, "nextCursor": None}, + {"data": {}, "nextCursor": None}, {"data": [], "nextCursor": True}, + {"data": [], "nextCursor": "PRIVATE cursor"}, + {"data": [{"input": "PRIVATE queued body"}], "nextCursor": None}] + responses = [{"result": value} for value in results] + responses += [{"error": {"code": -1, "message": "PRIVATE error"}}, + TimeoutError("PRIVATE timeout"), "PRIVATE invalid JSON", KeyboardInterrupt()] + for index, response in enumerate(responses): + with self.subTest(case=index): + code, sent, output = self.invoke(str(self.home), queue_list_response=response) + self.assertEqual(code, 130 if isinstance(response, KeyboardInterrupt) else 1) + self.assertFalse(any(item["method"] == "thread/queue/add" for item in sent)) + result = json.loads(output.splitlines()[-1]) + self.assertIs(result["accepted"], False) + self.assertEqual(result["outcome"], "not_attempted") + self.assertFalse(result["retryAllowed"]) + self.assertNotIn("PRIVATE", output) + + def test_invalid_queued_submission_shapes_preserve_unknown_after_one_add(self): + for index, queued in enumerate((None, [], True, "PRIVATE", {}, {"id": "fixture-submission"})): + with self.subTest(case=index): + code, sent, output = self.invoke(str(self.home), {"result": {"queuedSubmission": queued}}) + self.assertEqual(code, 1) + self.assertEqual(sum(item["method"] == "thread/queue/add" for item in sent), 1) + result = json.loads(output.splitlines()[-1]) + self.assertEqual(result["error"], "invalid_queue_result") + self.assertIsNone(result["accepted"]) + self.assertEqual(result["outcome"], "unknown") + self.assertFalse(result["retryAllowed"]) + self.assertNotIn("PRIVATE", output) + + def test_submission_fields_and_echo_must_match_the_request(self): + changes = [("id", value) for value in (None, False, [], "")] + changes += [("input", value) for value in (None, [], "PRIVATE", [{"type": "text", "text": "PRIVATE"}])] + changes += [("clientUserMessageId", value) for value in (None, False, [], "PRIVATE mismatched ID")] + for index, (key, value) in enumerate(changes): + def response(request): + reply = successful_queue_reply(request) + reply["result"]["queuedSubmission"][key] = value + return reply + with self.subTest(case=index): + code, sent, output = self.invoke(str(self.home), response) + self.assertEqual(code, 1) + self.assertEqual(sum(item["method"] == "thread/queue/add" for item in sent), 1) + result = json.loads(output.splitlines()[-1]) + self.assertIsNone(result["accepted"]) + self.assertEqual(result["outcome"], "unknown") + self.assertFalse(result["retryAllowed"]) + self.assertNotIn("PRIVATE", output) + + def test_response_ids_require_the_exact_integer_request_id(self): + original = FakeWebSocket.recv + for phase in ("initialize", "thread/queue/add"): + for kind in ("float", "bool", "missing"): + def metadata(ws, timeout): + raw = original(ws, timeout) + if ws.sent[-1]["method"] == phase: + value = json.loads(raw) + if kind == "missing": + value.pop("id") + # Unknown notifications remain bounded by 200 frames. + else: + value["id"] = float(value["id"]) if kind == "float" else True + return json.dumps(value) + return raw + with self.subTest(phase=phase, kind=kind), mock.patch.object(FakeWebSocket, "recv", metadata): + code, sent, output = self.invoke(str(self.home)) + self.assertEqual(code, 1) + expected_attempts = 1 if phase == "thread/queue/add" else 0 + self.assertEqual(sum(item["method"] == "thread/queue/add" for item in sent), expected_attempts) + result = json.loads(output.splitlines()[-1]) + if expected_attempts: + self.assertIsNone(result["accepted"]) + self.assertEqual(result["outcome"], "unknown") + else: + self.assertIs(result["accepted"], False) + self.assertEqual(result["outcome"], "not_attempted") + self.assertFalse(result["retryAllowed"]) + def test_malformed_errors_fail_without_traceback_or_retry(self): for error in (None, [], "PRIVATE", {"code": True, "message": "PRIVATE"}, {"code": "-32603", "message": "PRIVATE"}, {}, {"code": -32603}, From ec924debf519e2e51e7f21a8380a1cf037e64619 Mon Sep 17 00:00:00 2001 From: YEONGUK LEE Date: Thu, 15 Oct 2026 13:53:38 +0900 Subject: [PATCH 7/9] test(experiment): isolate queue echo and request ID regressions --- tests/codex/test_first_input_probe.py | 27 ++++++++++++++++++++++++--- 1 file changed, 24 insertions(+), 3 deletions(-) diff --git a/tests/codex/test_first_input_probe.py b/tests/codex/test_first_input_probe.py index 7aefb783..51899df0 100644 --- a/tests/codex/test_first_input_probe.py +++ b/tests/codex/test_first_input_probe.py @@ -132,13 +132,21 @@ def test_queue_rejection_returns_failure_without_retry(self): self.assertNotIn("PRIVATE", output) def test_queue_preflight_is_bound_and_immediately_precedes_the_only_add(self): - exit_code, sent, output = self.invoke(str(self.home)) + # Independent pinned-wire fixture, not an echo of the generated request. + client_id = "00000000-0000-4000-8000-000000000001" + expected_input = [{"type": "text", "text": "Owner-authorized one-shot transport test. Reply exactly ACK TEST. Do not run tools or modify files.", + "text_elements": []}] + reply = {"result": {"queuedSubmission": {"id": "fixture-submission", "input": expected_input, + "clientUserMessageId": client_id}}} + with mock.patch.object(self.probe.uuid, "uuid4", return_value=self.probe.uuid.UUID(client_id)): + exit_code, sent, output = self.invoke(str(self.home), reply) self.assertEqual(exit_code, 0) self.assertEqual([item["method"] for item in sent][-2:], ["thread/queue/list", "thread/queue/add"]) listed, added = sent[-2:] self.assertEqual(listed["params"], {"threadId": "owned-thread", "limit": 1}) self.assertEqual(added["params"]["threadId"], listed["params"]["threadId"]) - self.assertEqual(added["params"]["input"][0]["text_elements"], []) + self.assertEqual(added["params"]["input"], expected_input) + self.assertEqual(added["params"]["clientUserMessageId"], client_id) self.assertEqual(sum(item["method"] == "thread/queue/add" for item in sent), 1) self.assertTrue(json.loads(output.splitlines()[-1])["accepted"]) self.assertFalse(json.loads(output.splitlines()[-1])["consumptionConfirmed"]) @@ -196,9 +204,22 @@ def response(request): self.assertFalse(result["retryAllowed"]) self.assertNotIn("PRIVATE", output) + def missing_default(request): + reply = successful_queue_reply(request) + text = request["params"]["input"][0]["text"] + reply["result"]["queuedSubmission"]["input"] = [{"type": "text", "text": text}] + return reply + code, sent, output = self.invoke(str(self.home), missing_default) + self.assertEqual(code, 1) + self.assertEqual(sum(item["method"] == "thread/queue/add" for item in sent), 1) + result = json.loads(output.splitlines()[-1]) + self.assertIsNone(result["accepted"]) + self.assertEqual(result["outcome"], "unknown") + self.assertFalse(result["retryAllowed"]) + def test_response_ids_require_the_exact_integer_request_id(self): original = FakeWebSocket.recv - for phase in ("initialize", "thread/queue/add"): + for phase in ("initialize", "thread/queue/list", "thread/queue/add"): for kind in ("float", "bool", "missing"): def metadata(ws, timeout): raw = original(ws, timeout) From 91db14b10118ae06e0da97dc97239bddeacc0cca Mon Sep 17 00:00:00 2001 From: YEONGUK LEE Date: Thu, 15 Oct 2026 14:11:06 +0900 Subject: [PATCH 8/9] fix(experiment): derive preflight evidence from server turn history --- experiments/codex_first_input/ORDINARY_TUI.md | 72 +++++- experiments/codex_first_input/RESULTS.md | 19 +- experiments/codex_first_input/probe.py | 66 +++-- tests/codex/test_first_input_probe.py | 225 ++++++++++++++++-- 4 files changed, 325 insertions(+), 57 deletions(-) diff --git a/experiments/codex_first_input/ORDINARY_TUI.md b/experiments/codex_first_input/ORDINARY_TUI.md index 1a0503fd..19d87701 100644 --- a/experiments/codex_first_input/ORDINARY_TUI.md +++ b/experiments/codex_first_input/ORDINARY_TUI.md @@ -113,9 +113,11 @@ artifacts by the existing packaging rules. ## Review-driven harness guards -The supplied Codex home is now canonicalized before SQLite URI construction and -must match the absolute `codexHome` returned by `initialize` before any thread -lookup or database inspection. Missing or mismatched server-home evidence +The supplied Codex home is canonicalized and must match the absolute +`codexHome` returned by `initialize` before any thread lookup. This binds the +account home, not the running server's effective SQLite location. The current +harness makes no direct SQLite access and does not use account-home rows for +freshness. Missing or mismatched server-home evidence refuses the probe without queuing. A queue RPC rejection returns exit 1; malformed error objects, non-integer codes or missing/non-string messages yield an unknown outcome once queue submission has been attempted, without reflecting native error text or retrying. Timeout, @@ -124,11 +126,10 @@ report `accepted:null`, `outcome:unknown`, and `retryAllowed:false`; after a que attempt, a validated RPC rejection alone supplies `accepted:false`. Runtime preflight failures report `accepted:false`, `outcome:not_attempted`; argparse usage errors retain exit 2 and usage output. An interrupted probe returns exit 130 while preserving the -same effect boundary. Queue mode requires explicit `ephemeral:false` before -database inspection or submission: a previously used ephemeral conversation may -still have no database row. True, null and missing values cannot establish this -precondition. Inspection mode may report boolean/null metadata without queuing; -that does not prove the conversation is unused. Arbitrary result keys and native +same effect boundary. Both modes require explicit `ephemeral:false` before +turn-history inspection: a previously used ephemeral conversation may still +have no database row. True, null and missing values cannot establish this +precondition. Arbitrary result keys and native error details are not reflected. The version guard remains exactly 0.162.1, and Python optimization is refused up front. @@ -136,7 +137,7 @@ The pinned [0.162.1 Thread schema](https://github.com/openai/codex/blob/092d3acd declares `ephemeral` as a bool without an omission attribute. This is a static source basis for requiring explicit false, not a fresh native serialization test. -Immediately before the one queue attempt, the probe requests the selected +Before the one queue attempt, the probe requests the selected thread's `thread/queue/list` first page with `limit:1`. Only an explicit empty `data` array and `nextCursor:null` pass. Missing, malformed, nonempty or paginated responses refuse before any queue submission. This is a point-in-time check, @@ -148,6 +149,55 @@ harness neither prints nor inspects them beyond refusing a nonempty list, but the parsed response exists in memory. It is not a body-free observation API and must not be used on a server containing unrelated work. +### Server-owned turn evidence instead of local SQLite + +The current harness requires `--allow-server-history-db-init` before connecting +in either mode. It checks the selected thread's metadata via +`thread/read(includeTurns:false)` and requires an explicitly empty `preview`, +the expected cwd, idle/direct-input state and persistent-thread flag. It then +calls `thread/turns/list` on that same server/thread with first-page `limit:1`, +`sortDirection:"desc"` and `itemsView:"notLoaded"`. Only explicit `data:[]`, +`nextCursor:null` and `backwardsCursor:null` pass. It repeats both metadata and +turn checks after the pending-queue check, before the single add attempt. +Nonempty, missing, malformed or failed evidence refuses before add; there is +no local-DB fallback or resubmission. + +This closes the stale-account-home DB inference: the evidence now comes from +the connected server's own history path, independent of where its SQLite +configuration points. The result reports `serverTurnsEmpty` and +`freshnessSource:"thread/turns/list"`, not `storedRowsBefore`. It does not report +or attest the server's startup SQLite identity or resolve its environment, +profiles, overrides, managed/system/project settings or cloud layers. No new +known-override parser or dependency on the unmerged #311 guard is introduced. + +The pinned [turn-list handler](https://github.com/openai/codex/blob/092d3acd6bec3e3a14bdc7e7a2810ab628ab759d/codex-rs/app-server/src/request_processors/thread_processor.rs#L3073-L3286) +does not request `persist_thread` in this path. This is **not a globally +write-free operation**: the paginated history path can lazily open/create and +migrate a history database ([lazy open](https://github.com/openai/codex/blob/092d3acd6bec3e3a14bdc7e7a2810ab628ab759d/codex-rs/thread-store/src/local/mod.rs#L293-L306), +[read-write initialization and migration](https://github.com/openai/codex/blob/092d3acd6bec3e3a14bdc7e7a2810ab628ab759d/codex-rs/state/src/sqlite.rs#L296-L338)). +`inspect` is non-submitting, not a promise of zero server writes. No new native +history request was performed to test these guards. `includeTurns:true` is +never requested; that separate read path can explicitly persist a thread. + +`itemsView:"notLoaded"` suppresses turn items rather than requesting the default +summary with user/agent messages ([projection](https://github.com/openai/codex/blob/092d3acd6bec3e3a14bdc7e7a2810ab628ab759d/codex-rs/app-server-protocol/src/protocol/turn_items_view.rs#L6-L31)). +This is still not a universal body-free API: preview and turn error metadata +can contain text in memory. They are never printed, and a nonempty preview or +turn list is refused without item inspection. Use only the dedicated owned +test conversation, never unrelated work or transcript files. + +An empty response means only that the connected server reports no visible +turns at that instant. Rollback/compaction, server bugs or paginated live-event +projection can limit that evidence; the paginated path does not merge the same +active-turn overlay as the nonpaginated path. It is not proof of no historical +input ever, an atomic expected-empty condition or a general first-input API. +Idle/exclusive ownership and no competing producers remain mandatory, and the +check-to-add race remains. The historical native observations above were not +re-run with these revised guards. +For rollout-backed history the server may replay the full rollout before +pagination; `limit:1` is not a bound on server-side history work. The client has +bounded receive frames/timeouts, not a general server resource guarantee. + A successful queue reply must contain a structured `queuedSubmission` with a nonempty string ID and the exact submitted plain-text `input` and `clientUserMessageId`. The explicit `text_elements:[]` matches the pinned native @@ -158,12 +208,12 @@ See the pinned [queue request/response schema](https://github.com/openai/codex/b and [plain-text input schema](https://github.com/openai/codex/blob/092d3acd6bec3e3a14bdc7e7a2810ab628ab759d/codex-rs/app-server-protocol/src/protocol/v2/turn.rs#L436-L445). These checks were not retrospectively performed in the historical live runs. -Fifteen synthetic WebSocket/temporary-database tests cover these boundaries, +Twenty-five synthetic WebSocket/adversarial account-home fixtures cover these boundaries, including relative home paths, complete error validation on both RPC paths, ephemeral/missing/null refusal, empty-queue preflight, submission echo validation, exact RPC IDs and exactly one queue attempt on rejection or uncertain failure. These are harness regression tests, not new native delivery, receipt, platform, or production-support evidence. None of these review-driven guards were exercised -by another live native run. SQLite handles are explicitly closed. A failure during +by another live native run. A failure during connection cleanup may return a nonzero exit while preserving an already observed acceptance or rejection; it never authorizes retry. diff --git a/experiments/codex_first_input/RESULTS.md b/experiments/codex_first_input/RESULTS.md index 6a00be67..faccbdf1 100644 --- a/experiments/codex_first_input/RESULTS.md +++ b/experiments/codex_first_input/RESULTS.md @@ -2,6 +2,12 @@ ## Result +This is a historical report, not a validation of the current harness. The +account-home row count below was an observed diagnostic, not proof of the +running server's effective SQLite location. Later review found that relocated +storage can invalidate that inference. The harness now uses server-reported +turn snapshots instead; those new guards have not had a fresh native run. + For the subsequent **already-open ordinary TUI** negative/positive controls, see [ORDINARY_TUI.md](ORDINARY_TUI.md). An embedded blank TUI was unreachable through the external queue path, while a plain TUI already connected to a shared @@ -45,10 +51,15 @@ first-input transport; it does not make the public `session-peer send` support i `probe.py` requires Python with websockets 17.x and an **exclusively owned** fresh server/TUI. It does not launch or discover an arbitrary server. Operators supply the private socket, the same account home used to launch that server, and the -exact approved TUI cwd. `--mode inspect` is read-only; `--mode queue` sends once -only after checking a zero-row precondition. Do not run it against an existing -working conversation. Run without Python optimization because its experimental -preconditions use assertions. +exact approved TUI cwd. The current harness no longer opens local SQLite or +uses account-home row counts. Both modes require +`--allow-server-history-db-init`: even a non-submitting server history lookup +can initialize or migrate the server's history database. `--mode inspect` +reports an empty server-visible-turn snapshot without queuing; `--mode queue` +rechecks the server metadata/turns after checking its pending queue, then sends +once. These are non-atomic snapshots, not storage identity or never-used-thread +attestations. Do not run it against an existing working conversation. Run +without Python optimization because its experimental preconditions use assertions. The endpoint may be a native-created symlink to a shorter physical Unix socket; the harness checks both the logical and resolved private directories and socket diff --git a/experiments/codex_first_input/probe.py b/experiments/codex_first_input/probe.py index fbeb500f..6ddcdb8e 100644 --- a/experiments/codex_first_input/probe.py +++ b/experiments/codex_first_input/probe.py @@ -2,14 +2,13 @@ Requires websockets 17.x. Never starts a server or retries a submission. Prints sanitized facts, not conversation IDs, paths, or native text. +Both modes require explicit opt-in to possible server history-DB initialization. """ import argparse -from contextlib import closing import json import os from pathlib import Path import re -import sqlite3 import stat import time import uuid @@ -46,6 +45,16 @@ def require_empty_queue(result): raise ProbeError("queue_not_empty_or_unproven") +def require_empty_turns(result): + if not isinstance(result.get("data"), list): + raise ProbeError("invalid_turns_snapshot") + for key in ("nextCursor", "backwardsCursor"): + if key not in result or (result[key] is not None and not isinstance(result[key], str)): + raise ProbeError("invalid_turns_snapshot") + if result["data"] or result["nextCursor"] is not None or result["backwardsCursor"] is not None: + raise ProbeError("server_turns_present_or_unproven") + + def validate_queued_submission(result, expected_input, client_message_id): queued = result.get("queuedSubmission") if (not isinstance(queued, dict) or not isinstance(queued.get("id"), str) or not queued["id"] @@ -66,7 +75,11 @@ def main(argv=None, state=None): parser.add_argument("--expected-cwd", required=True) parser.add_argument("--mode", choices=("inspect", "queue"), default="inspect") parser.add_argument("--ack-marker", help="Harmless ASCII test receipt marker; no credentials") + parser.add_argument("--allow-server-history-db-init", action="store_true", + help="Acknowledge that server history reads may initialize/migrate its history DB; owned test server only") args = parser.parse_args(argv) + if not args.allow_server_history_db_init: + raise ProbeError("server_history_read_requires_opt_in") if args.ack_marker is not None and not re.fullmatch(r"[A-Za-z0-9 _-]{1,100}", args.ack_marker): parser.error("invalid ACK marker") codex_home = Path(args.codex_home).resolve(strict=True) @@ -113,34 +126,45 @@ def call(method, params): raise ProbeError("server_home_mismatch") ws.send(json.dumps({"method": "initialized"})) loaded = rpc_result(call("thread/loaded/list", {"limit": 100})) - assert loaded.get("nextCursor") is None and len(loaded["data"]) == 1 + if (not isinstance(loaded.get("data"), list) or len(loaded["data"]) != 1 + or not isinstance(loaded["data"][0], str) or not loaded["data"][0] + or "nextCursor" not in loaded or loaded["nextCursor"] is not None): + raise ProbeError("invalid_loaded_snapshot") thread_id = loaded["data"][0] - thread = rpc_result(call("thread/read", {"threadId": thread_id, "includeTurns": False}))["thread"] - assert Path(thread["cwd"]).resolve() == Path(args.expected_cwd).resolve() - assert thread["status"]["type"] == "idle" and thread["canAcceptDirectInput"] is True - ephemeral = thread.get("ephemeral") - if ephemeral is not None and type(ephemeral) is not bool: - raise ProbeError("invalid_thread_metadata") - if args.mode == "queue" and ephemeral is not False: - # Ephemeral conversations can have prior turns without any stored - # row. Missing/null metadata cannot establish persistence either. - raise ProbeError("persistent_thread_required") - rows = 0 - dbs = list(codex_home.glob("state_*.sqlite")) - assert dbs - for db in dbs: - with closing(sqlite3.connect(db.as_uri()+"?mode=ro", uri=True)) as connection: - rows += connection.execute("SELECT count(*) FROM threads WHERE id=?", (thread_id,)).fetchone()[0] - facts = {"version": "0.162.1", "loadedThreads": 1, "storedRowsBefore": rows, "idle": True, "directInput": True, "ephemeral": ephemeral, "mode": args.mode} + def check_server_snapshot(): + # Use the connected server's store, never an account-home DB whose + # SQLite location may differ from the server's startup selection. + thread = rpc_result(call("thread/read", {"threadId": thread_id, "includeTurns": False}))["thread"] + if not isinstance(thread, dict) or thread.get("id") != thread_id: + raise ProbeError("invalid_thread_metadata") + assert Path(thread["cwd"]).resolve() == Path(args.expected_cwd).resolve() + assert thread["status"]["type"] == "idle" and thread["canAcceptDirectInput"] is True + ephemeral = thread.get("ephemeral") + if ephemeral is not None and type(ephemeral) is not bool: + raise ProbeError("invalid_thread_metadata") + if ephemeral is not False: + raise ProbeError("persistent_thread_required") + preview = thread.get("preview") + if not isinstance(preview, str): + raise ProbeError("invalid_thread_metadata") + if preview: + raise ProbeError("server_preview_present") + require_empty_turns(rpc_result(call("thread/turns/list", { + "threadId": thread_id, "limit": 1, "sortDirection": "desc", "itemsView": "notLoaded"}))) + + check_server_snapshot() + facts = {"version": "0.162.1", "loadedThreads": 1, "serverTurnsEmpty": True, + "freshnessSource": "thread/turns/list", "idle": True, "directInput": True, + "ephemeral": False, "mode": args.mode} print(json.dumps(facts), flush=True) if args.mode == "inspect": return 0 - assert rows == 0, "requires fresh pre-first-input thread" marker = args.ack_marker or "ACK CODEX-FIRST-INPUT-306-" + args.mode.upper() message = "Owner-authorized one-shot transport test. Reply exactly " + marker + ". Do not run tools or modify files." # A point-in-time preflight, not an atomic expected-empty condition. # Only use the exclusively owned server with no other input producers. require_empty_queue(rpc_result(call("thread/queue/list", {"threadId": thread_id, "limit": 1}))) + check_server_snapshot() expected_input = [{"type": "text", "text": message, "text_elements": []}] client_message_id = str(uuid.uuid4()) method = "thread/queue/add" diff --git a/tests/codex/test_first_input_probe.py b/tests/codex/test_first_input_probe.py index 51899df0..b32210b7 100644 --- a/tests/codex/test_first_input_probe.py +++ b/tests/codex/test_first_input_probe.py @@ -21,11 +21,16 @@ def successful_queue_reply(request): class FakeWebSocket: - def __init__(self, home, queue_response, queue_list_response=None): + def __init__(self, home, queue_response, queue_list_response=None, turns_responses=None, + thread_snapshots=None): self.home = home self.queue_response = queue_response self.queue_list_response = queue_list_response self.sent = [] + self.turns_responses = turns_responses + self.thread_snapshots = thread_snapshots + self.turns_count = 0 + self.thread_count = 0 def __enter__(self): return self @@ -48,8 +53,17 @@ def recv(self, timeout): elif method == "thread/loaded/list": result = {"data": ["owned-thread"], "nextCursor": None} elif method == "thread/read": - result = {"thread": {"cwd": self.home, "status": {"type": "idle"}, - "canAcceptDirectInput": True, "ephemeral": False}} + result = {"thread": {"id": "owned-thread", "cwd": self.home, "status": {"type": "idle"}, + "canAcceptDirectInput": True, "ephemeral": False, "preview": ""}} + if self.thread_snapshots is not None: + result["thread"].update(self.thread_snapshots[min(self.thread_count, len(self.thread_snapshots)-1)]) + self.thread_count += 1 + elif method == "thread/turns/list": + response = {"result": {"data": [], "nextCursor": None, "backwardsCursor": None}} + if self.turns_responses is not None: + response = self.turns_responses[min(self.turns_count, len(self.turns_responses)-1)] + self.turns_count += 1 + return self.encode_reply(request, response) elif method == "thread/queue/list": response = ({"result": {"data": [], "nextCursor": None}} if self.queue_list_response is None else self.queue_list_response) @@ -96,26 +110,39 @@ def setUp(self): db.execute("CREATE TABLE threads (id TEXT)") db.commit() - def invoke(self, server_home, queue_response=None, supplied_home=None, mode="queue", queue_list_response=None): - ws = FakeWebSocket(server_home, queue_response, queue_list_response) + def invoke(self, server_home, queue_response=None, supplied_home=None, mode="queue", queue_list_response=None, + turns_responses=None, thread_snapshots=None, allow_history=True): + ws = FakeWebSocket(server_home, queue_response, queue_list_response, turns_responses, thread_snapshots) args = ["--socket", str(self.endpoint), "--codex-home", supplied_home or str(self.home), "--expected-cwd", str(self.home), "--mode", mode, "--ack-marker", "ACK TEST"] + if allow_history: + args.append("--allow-server-history-db-init") output = io.StringIO() - with mock.patch.object(self.probe, "unix_connect", return_value=ws), contextlib.redirect_stdout(output): + with mock.patch.object(self.probe, "unix_connect", return_value=ws) as connect, \ + mock.patch.object(sqlite3, "connect", side_effect=AssertionError("No local DB reads")) as database, \ + contextlib.redirect_stdout(output): exit_code = self.probe.run(args) + database.assert_not_called() + if not allow_history: + connect.assert_not_called() + for item in ws.sent: + self.assertNotIn(item["method"], ("thread/start", "thread/resume", "turn/start")) + if item["method"] == "thread/read": + self.assertIs(item["params"]["includeTurns"], False) + if item["method"] == "thread/turns/list": + self.assertEqual(item["params"], {"threadId": "owned-thread", "limit": 1, + "sortDirection": "desc", "itemsView": "notLoaded"}) return exit_code, ws.sent, output.getvalue() def test_missing_or_mismatched_home_refuses_before_thread_or_database(self): for server_home in (None, str(self.home.parent), "relative", ""): - with self.subTest(server_home=server_home), mock.patch.object( - self.probe.sqlite3, "connect", side_effect=AssertionError("DB must not open")) as database: + with self.subTest(server_home=server_home): exit_code, sent, output = self.invoke(server_home) - database.assert_not_called() self.assertEqual(exit_code, 1) self.assertEqual([item["method"] for item in sent], ["initialize"]) self.assertNotIn(str(self.home), output) - def test_relative_home_is_canonical_before_sqlite_uri(self): + def test_relative_home_is_canonical_before_server_binding(self): relative = os.path.relpath(str(self.home), os.getcwd()) exit_code, sent, output = self.invoke(str(self.home), supplied_home=relative) self.assertEqual(exit_code, 0) @@ -131,7 +158,7 @@ def test_queue_rejection_returns_failure_without_retry(self): self.assertFalse(json.loads(output.splitlines()[-1])["retryAllowed"]) self.assertNotIn("PRIVATE", output) - def test_queue_preflight_is_bound_and_immediately_precedes_the_only_add(self): + def test_queue_and_turn_preflights_are_bound_before_the_only_add(self): # Independent pinned-wire fixture, not an echo of the generated request. client_id = "00000000-0000-4000-8000-000000000001" expected_input = [{"type": "text", "text": "Owner-authorized one-shot transport test. Reply exactly ACK TEST. Do not run tools or modify files.", @@ -141,8 +168,9 @@ def test_queue_preflight_is_bound_and_immediately_precedes_the_only_add(self): with mock.patch.object(self.probe.uuid, "uuid4", return_value=self.probe.uuid.UUID(client_id)): exit_code, sent, output = self.invoke(str(self.home), reply) self.assertEqual(exit_code, 0) - self.assertEqual([item["method"] for item in sent][-2:], ["thread/queue/list", "thread/queue/add"]) - listed, added = sent[-2:] + self.assertEqual([item["method"] for item in sent][-4:], + ["thread/queue/list", "thread/read", "thread/turns/list", "thread/queue/add"]) + listed, added = sent[-4], sent[-1] self.assertEqual(listed["params"], {"threadId": "owned-thread", "limit": 1}) self.assertEqual(added["params"]["threadId"], listed["params"]["threadId"]) self.assertEqual(added["params"]["input"], expected_input) @@ -219,7 +247,7 @@ def missing_default(request): def test_response_ids_require_the_exact_integer_request_id(self): original = FakeWebSocket.recv - for phase in ("initialize", "thread/queue/list", "thread/queue/add"): + for phase in ("initialize", "thread/read", "thread/turns/list", "thread/queue/list", "thread/queue/add"): for kind in ("float", "bool", "missing"): def metadata(ws, timeout): raw = original(ws, timeout) @@ -284,11 +312,9 @@ def metadata(ws, timeout): value["result"]["thread"].update(fields) return json.dumps(value) return raw - with self.subTest(case=label), mock.patch.object(FakeWebSocket, "recv", metadata), \ - mock.patch.object(self.probe.sqlite3, "connect", side_effect=AssertionError("DB must not open")) as db: + with self.subTest(case=label), mock.patch.object(FakeWebSocket, "recv", metadata): exit_code, sent, output = self.invoke(str(self.home)) self.assertEqual(exit_code, 1) - self.assertEqual(db.call_count, 0) self.assertFalse(any(item["method"] == "thread/queue/add" for item in sent)) result = json.loads(output.splitlines()[-1]) self.assertEqual(result["error"], "persistent_thread_required") @@ -296,7 +322,7 @@ def metadata(ws, timeout): self.assertEqual(result["outcome"], "not_attempted") self.assertFalse(result["retryAllowed"]) - def test_inspection_does_not_claim_row_absence_proves_ephemeral_freshness(self): + def test_inspection_requires_proven_persistent_metadata_too(self): original = FakeWebSocket.recv for label, fields in (("ephemeral", {"ephemeral": True}), ("null", {"ephemeral": None}), ("missing", {}), ("invalid", {"ephemeral": "PRIVATE"})): @@ -316,8 +342,8 @@ def metadata(ws, timeout): self.assertEqual(result["error"], "invalid_thread_metadata") self.assertNotIn("PRIVATE", output) else: - self.assertEqual(exit_code, 0) - self.assertIs(result["ephemeral"], fields.get("ephemeral")) + self.assertEqual(exit_code, 1) + self.assertEqual(result["error"], "persistent_thread_required") self.assertFalse(any(item["method"] == "thread/queue/add" for item in sent)) def test_uncertain_queue_failures_preserve_unknown_and_no_retry(self): @@ -340,7 +366,7 @@ def test_preflight_exception_is_sanitized_without_queue(self): for failure in (FileNotFoundError("PRIVATE path"), sqlite3.DatabaseError("PRIVATE DB"), AssertionError("PRIVATE assertion"), ValueError("PRIVATE value")): with self.subTest(failure=type(failure).__name__), mock.patch.object( - self.probe.sqlite3, "connect", side_effect=failure): + FakeWebSocket, "__enter__", side_effect=failure): exit_code, sent, output = self.invoke(str(self.home)) self.assertEqual(exit_code, 1) self.assertFalse(any(item["method"] == "thread/queue/add" for item in sent)) @@ -366,6 +392,163 @@ def malicious_metadata(ws, timeout): self.assertFalse(any(item["method"] == "thread/queue/add" for item in sent)) self.assertNotIn("PRIVATE", output) + def assert_pre_effect_failure(self, code, sent, output): + self.assertEqual(code, 1) + self.assertFalse(any(item["method"] == "thread/queue/add" for item in sent)) + result = json.loads(output.splitlines()[-1]) + self.assertIs(result["accepted"], False) + self.assertEqual(result["outcome"], "not_attempted") + self.assertFalse(result["retryAllowed"]) + self.assertNotIn("PRIVATE", output) + + def test_history_read_opt_in_required_before_any_connection(self): + for mode in ("inspect", "queue"): + with self.subTest(mode=mode): + code, sent, output = self.invoke(str(self.home), mode=mode, allow_history=False) + self.assert_pre_effect_failure(code, sent, output) + self.assertEqual(sent, []) + self.assertEqual(json.loads(output)["error"], "server_history_read_requires_opt_in") + + def test_stale_empty_account_database_cannot_override_server_turns(self): + (self.home / "config.toml").write_text('sqlite_home = "PRIVATE configured storage"\n') + with mock.patch.dict(os.environ, {"CODEX_SQLITE_HOME": "PRIVATE inherited storage"}): + code, sent, output = self.invoke(str(self.home), turns_responses=[{ + "result": {"data": [{"id": "PRIVATE old turn"}], "nextCursor": None, "backwardsCursor": None}}]) + self.assert_pre_effect_failure(code, sent, output) + self.assertEqual(json.loads(output.splitlines()[-1])["error"], "server_turns_present_or_unproven") + self.assertEqual(sum(item["method"] == "thread/turns/list" for item in sent), 1) + self.assertFalse(any(item["method"] == "thread/queue/list" for item in sent)) + + def test_empty_server_turns_do_not_consult_nonempty_or_missing_account_database(self): + with contextlib.closing(sqlite3.connect(self.home / "state_5.sqlite")) as db: + db.execute("INSERT INTO threads VALUES ('owned-thread')") + db.commit() + for missing in (False, True): + if missing: + (self.home / "state_5.sqlite").unlink() + with self.subTest(missing=missing): + code, sent, output = self.invoke(str(self.home)) + self.assertEqual(code, 0) + self.assertEqual(sum(item["method"] == "thread/queue/add" for item in sent), 1) + facts = json.loads(output.splitlines()[0]) + self.assertIs(facts["serverTurnsEmpty"], True) + self.assertEqual(facts["freshnessSource"], "thread/turns/list") + self.assertNotIn("storedRowsBefore", facts) + + def test_probe_environment_and_root_configuration_are_not_server_storage_authority(self): + (self.home / "config.toml").write_text('sqlite_home = "PRIVATE fixture relocation"\n') + with mock.patch.dict(os.environ, {"CODEX_SQLITE_HOME": "PRIVATE fixture environment"}): + code, sent, output = self.invoke(str(self.home)) + self.assertEqual(code, 0) + self.assertEqual(sum(item["method"] == "thread/queue/add" for item in sent), 1) + self.assertNotIn("PRIVATE", output) + + def test_invalid_or_nonempty_server_turns_refuse_at_both_snapshots(self): + empty = {"result": {"data": [], "nextCursor": None, "backwardsCursor": None}} + values = [None, [], {}, {"data": []}, + {"data": None, "nextCursor": None, "backwardsCursor": None}, + {"data": {}, "nextCursor": None, "backwardsCursor": None}, + {"data": [], "nextCursor": "PRIVATE cursor", "backwardsCursor": None}, + {"data": [], "nextCursor": None, "backwardsCursor": "PRIVATE cursor"}, + {"data": [], "nextCursor": True, "backwardsCursor": None}, + {"data": [], "nextCursor": None, "backwardsCursor": False}, + {"data": [], "nextCursor": None}, + {"data": [], "backwardsCursor": None}, + {"data": [{"id": "PRIVATE turn"}], "nextCursor": None, "backwardsCursor": None}] + responses = [{"result": value} for value in values] + responses += [{"error": {"code": -1, "message": "PRIVATE"}}, + {"error": {"code": -1}}, TimeoutError("PRIVATE"), "PRIVATE invalid JSON"] + for phase in (0, 1): + for index, response in enumerate(responses): + with self.subTest(phase=phase, case=index): + snapshots = [response] if phase == 0 else [empty, response] + code, sent, output = self.invoke(str(self.home), turns_responses=snapshots) + self.assert_pre_effect_failure(code, sent, output) + self.assertEqual(sum(item["method"] == "thread/queue/list" for item in sent), phase) + + def test_preview_must_be_explicit_empty_string_at_both_snapshots(self): + original = FakeWebSocket.recv + for phase in (0, 1): + for kind, preview in (("nonempty", "PRIVATE text"), ("null", None), + ("nonstring", 123), ("missing", None)): + def metadata(ws, timeout): + raw = original(ws, timeout) + if ws.sent[-1]["method"] == "thread/read" and ws.thread_count == phase + 1: + value = json.loads(raw) + thread = value["result"]["thread"] + if kind == "missing": + thread.pop("preview") + else: + thread["preview"] = preview + return json.dumps(value) + return raw + with self.subTest(phase=phase, case=kind), mock.patch.object(FakeWebSocket, "recv", metadata): + code, sent, output = self.invoke(str(self.home)) + self.assert_pre_effect_failure(code, sent, output) + self.assertEqual(sum(item["method"] == "thread/queue/list" for item in sent), phase) + + def test_inspect_reports_only_server_snapshot_and_never_submits(self): + code, sent, output = self.invoke(str(self.home), mode="inspect") + self.assertEqual(code, 0) + self.assertEqual([item["method"] for item in sent], + ["initialize", "initialized", "thread/loaded/list", "thread/read", "thread/turns/list"]) + facts = json.loads(output) + self.assertEqual(facts, {"version": "0.162.1", "loadedThreads": 1, "serverTurnsEmpty": True, + "freshnessSource": "thread/turns/list", "idle": True, + "directInput": True, "ephemeral": False, "mode": "inspect"}) + + def test_thread_identity_and_idle_direct_input_constraints_hold_at_both_snapshots(self): + invalid = [{"id": "PRIVATE other thread"}, {"id": None}, + {"cwd": str(self.home.parent)}, {"status": {"type": "active"}}, + {"status": None}, {"canAcceptDirectInput": False}, + {"canAcceptDirectInput": 1}, {"ephemeral": True}] + for phase in (0, 1): + for index, fields in enumerate(invalid): + with self.subTest(phase=phase, case=index): + snapshots = [fields] if phase == 0 else [{}, fields] + code, sent, output = self.invoke(str(self.home), thread_snapshots=snapshots) + self.assert_pre_effect_failure(code, sent, output) + self.assertEqual(sum(item["method"] == "thread/queue/list" for item in sent), phase) + + def test_exact_native_version_required_before_thread_queries(self): + original = FakeWebSocket.recv + for version in ("codex-tui/0.162.0 PRIVATE", "codex-tui/0.162.10 PRIVATE", None, True): + def metadata(ws, timeout): + raw = original(ws, timeout) + if ws.sent[-1]["method"] == "initialize": + value = json.loads(raw) + value["result"]["userAgent"] = version + return json.dumps(value) + return raw + with self.subTest(kind=type(version).__name__), mock.patch.object(FakeWebSocket, "recv", metadata): + code, sent, output = self.invoke(str(self.home)) + self.assert_pre_effect_failure(code, sent, output) + self.assertEqual([item["method"] for item in sent], ["initialize"]) + + def test_loaded_thread_snapshot_is_complete_and_unambiguous(self): + original = FakeWebSocket.recv + for data in ([], "x", [None], [True], [""], ["one", "two"]): + def metadata(ws, timeout): + raw = original(ws, timeout) + if ws.sent[-1]["method"] == "thread/loaded/list": + value = json.loads(raw) + value["result"]["data"] = data + return json.dumps(value) + return raw + with self.subTest(kind=type(data).__name__), mock.patch.object(FakeWebSocket, "recv", metadata): + code, sent, output = self.invoke(str(self.home)) + self.assert_pre_effect_failure(code, sent, output) + self.assertEqual(sent[-1]["method"], "thread/loaded/list") + for result in ({"data": ["owned-thread"]}, {"data": ["owned-thread"], "nextCursor": "PRIVATE cursor"}): + def metadata(ws, timeout): + if ws.sent[-1]["method"] == "thread/loaded/list": + return json.dumps({"id": ws.sent[-1]["id"], "result": result}) + return original(ws, timeout) + with self.subTest(keys=tuple(result)), mock.patch.object(FakeWebSocket, "recv", metadata): + code, sent, output = self.invoke(str(self.home)) + self.assert_pre_effect_failure(code, sent, output) + self.assertEqual(sent[-1]["method"], "thread/loaded/list") + if __name__ == "__main__": unittest.main() From 97f7df702235efa1849aec54f59c68dd9ce3cd46 Mon Sep 17 00:00:00 2001 From: YEONGUK LEE Date: Thu, 15 Oct 2026 14:16:39 +0900 Subject: [PATCH 9/9] test(experiment): pin unmaterialized history refusal and evidence limits --- experiments/codex_first_input/ORDINARY_TUI.md | 22 +++++++++++++++-- experiments/codex_first_input/RESULTS.md | 3 +++ tests/codex/test_first_input_probe.py | 24 ++++++++++++++++--- 3 files changed, 44 insertions(+), 5 deletions(-) diff --git a/experiments/codex_first_input/ORDINARY_TUI.md b/experiments/codex_first_input/ORDINARY_TUI.md index 19d87701..37c41476 100644 --- a/experiments/codex_first_input/ORDINARY_TUI.md +++ b/experiments/codex_first_input/ORDINARY_TUI.md @@ -162,7 +162,7 @@ turn checks after the pending-queue check, before the single add attempt. Nonempty, missing, malformed or failed evidence refuses before add; there is no local-DB fallback or resubmission. -This closes the stale-account-home DB inference: the evidence now comes from +This replaces the stale-account-home DB inference: the evidence now comes from the connected server's own history path, independent of where its SQLite configuration points. The result reports `serverTurnsEmpty` and `freshnessSource:"thread/turns/list"`, not `storedRowsBefore`. It does not report @@ -198,6 +198,24 @@ For rollout-backed history the server may replay the full rollout before pagination; `limit:1` is not a bound on server-side history work. The client has bounded receive frames/timeouts, not a general server resource guarantee. +**Unmaterialized first-input targets may be unavailable.** The pinned +[history fallback](https://github.com/openai/codex/blob/092d3acd6bec3e3a14bdc7e7a2810ab628ab759d/codex-rs/app-server/src/request_processors/thread_processor.rs#L3514-L3571) +loads the live thread's history if no stored rollout is found, but missing +rollout resolution can produce an explicit +["unavailable before first user message" error](https://github.com/openai/codex/blob/092d3acd6bec3e3a14bdc7e7a2810ab628ab759d/codex-rs/app-server/src/request_processors/thread_processor.rs#L5942-L5952). +The harness refuses that response before add in either mode. It never converts +unsupported history into an empty snapshot or falls back to an account-home +DB. This safety correction may prevent reproduction of the original unsaved +first-input positive; a synthetic empty-server reply is not native availability +evidence. A supported empty-TUI proof route remains an open #306 gate. The +historical successful sends do not validate this revised path. + +In queue mode the first JSON object describes only the initial snapshot. +Consumers must check the final object and process exit, because the second +snapshot can refuse while the earlier `serverTurnsEmpty:true` observation +remains in stdout. Neither an initial snapshot nor inspect success authorizes +a later submission without its own preflight. + A successful queue reply must contain a structured `queuedSubmission` with a nonempty string ID and the exact submitted plain-text `input` and `clientUserMessageId`. The explicit `text_elements:[]` matches the pinned native @@ -208,7 +226,7 @@ See the pinned [queue request/response schema](https://github.com/openai/codex/b and [plain-text input schema](https://github.com/openai/codex/blob/092d3acd6bec3e3a14bdc7e7a2810ab628ab759d/codex-rs/app-server-protocol/src/protocol/v2/turn.rs#L436-L445). These checks were not retrospectively performed in the historical live runs. -Twenty-five synthetic WebSocket/adversarial account-home fixtures cover these boundaries, +Twenty-six synthetic WebSocket/adversarial account-home fixtures cover these boundaries, including relative home paths, complete error validation on both RPC paths, ephemeral/missing/null refusal, empty-queue preflight, submission echo validation, exact RPC IDs and exactly one queue attempt on rejection or uncertain failure. diff --git a/experiments/codex_first_input/RESULTS.md b/experiments/codex_first_input/RESULTS.md index faccbdf1..2f9ef028 100644 --- a/experiments/codex_first_input/RESULTS.md +++ b/experiments/codex_first_input/RESULTS.md @@ -7,6 +7,9 @@ account-home row count below was an observed diagnostic, not proof of the running server's effective SQLite location. Later review found that relocated storage can invalidate that inference. The harness now uses server-reported turn snapshots instead; those new guards have not had a fresh native run. +Codex 0.162.1 may refuse that history RPC before the first user message. The +current harness fails closed on such targets, so the historical positive must +not be taken as evidence that the revised reproduction path is available. For the subsequent **already-open ordinary TUI** negative/positive controls, see [ORDINARY_TUI.md](ORDINARY_TUI.md). An embedded blank TUI was unreachable diff --git a/tests/codex/test_first_input_probe.py b/tests/codex/test_first_input_probe.py index b32210b7..044c3c36 100644 --- a/tests/codex/test_first_input_probe.py +++ b/tests/codex/test_first_input_probe.py @@ -134,7 +134,7 @@ def invoke(self, server_home, queue_response=None, supplied_home=None, mode="que "sortDirection": "desc", "itemsView": "notLoaded"}) return exit_code, ws.sent, output.getvalue() - def test_missing_or_mismatched_home_refuses_before_thread_or_database(self): + def test_missing_or_mismatched_home_refuses_before_thread_queries(self): for server_home in (None, str(self.home.parent), "relative", ""): with self.subTest(server_home=server_home): exit_code, sent, output = self.invoke(server_home) @@ -300,7 +300,7 @@ def test_rpc_result_uses_the_same_complete_error_validation(self): self.probe.rpc_result({"error": {"code": -1, "message": message}}) self.assertEqual(str(caught.exception), "rpc_rejected") - def test_ephemeral_or_unproven_threads_refuse_before_database_or_queue(self): + def test_ephemeral_or_unproven_threads_refuse_before_history_or_queue(self): original = FakeWebSocket.recv for label, fields in (("ephemeral", {"ephemeral": True}), ("null", {"ephemeral": None}), ("missing", {})): @@ -363,7 +363,7 @@ def test_uncertain_queue_failures_preserve_unknown_and_no_retry(self): self.assertNotIn(str(self.home), output) def test_preflight_exception_is_sanitized_without_queue(self): - for failure in (FileNotFoundError("PRIVATE path"), sqlite3.DatabaseError("PRIVATE DB"), + for failure in (FileNotFoundError("PRIVATE path"), RuntimeError("PRIVATE protocol state"), AssertionError("PRIVATE assertion"), ValueError("PRIVATE value")): with self.subTest(failure=type(failure).__name__), mock.patch.object( FakeWebSocket, "__enter__", side_effect=failure): @@ -419,6 +419,18 @@ def test_stale_empty_account_database_cannot_override_server_turns(self): self.assertEqual(sum(item["method"] == "thread/turns/list" for item in sent), 1) self.assertFalse(any(item["method"] == "thread/queue/list" for item in sent)) + def test_unmaterialized_history_error_is_refused_without_fallback(self): + unavailable = {"error": {"code": -32602, "message": + "thread PRIVATE is not materialized yet; thread/turns/list is unavailable before first user message"}} + empty = {"result": {"data": [], "nextCursor": None, "backwardsCursor": None}} + for mode, snapshots in (("inspect", [unavailable]), ("queue", [unavailable]), + ("queue", [empty, unavailable])): + with self.subTest(mode=mode, phase=len(snapshots)): + code, sent, output = self.invoke(str(self.home), mode=mode, turns_responses=snapshots) + self.assert_pre_effect_failure(code, sent, output) + self.assertEqual(json.loads(output.splitlines()[-1])["error"], "rpc_rejected") + self.assertEqual(sum(item["method"] == "thread/turns/list" for item in sent), len(snapshots)) + def test_empty_server_turns_do_not_consult_nonempty_or_missing_account_database(self): with contextlib.closing(sqlite3.connect(self.home / "state_5.sqlite")) as db: db.execute("INSERT INTO threads VALUES ('owned-thread')") @@ -496,6 +508,12 @@ def test_inspect_reports_only_server_snapshot_and_never_submits(self): self.assertEqual(facts, {"version": "0.162.1", "loadedThreads": 1, "serverTurnsEmpty": True, "freshnessSource": "thread/turns/list", "idle": True, "directInput": True, "ephemeral": False, "mode": "inspect"}) + for fields in ({"preview": "PRIVATE prior message"},): + code, sent, output = self.invoke(str(self.home), mode="inspect", thread_snapshots=[fields]) + self.assert_pre_effect_failure(code, sent, output) + code, sent, output = self.invoke(str(self.home), mode="inspect", turns_responses=[{ + "result": {"data": [{"id": "PRIVATE prior turn"}], "nextCursor": None, "backwardsCursor": None}}]) + self.assert_pre_effect_failure(code, sent, output) def test_thread_identity_and_idle_direct_input_constraints_hold_at_both_snapshots(self): invalid = [{"id": "PRIVATE other thread"}, {"id": None},