diff --git a/experiments/codex_first_input/ORDINARY_TUI.md b/experiments/codex_first_input/ORDINARY_TUI.md new file mode 100644 index 00000000..37c41476 --- /dev/null +++ b/experiments/codex_first_input/ORDINARY_TUI.md @@ -0,0 +1,237 @@ +# 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. 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 +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. + +## Review-driven harness guards + +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, +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. 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. + +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. + +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. + +### 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 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 +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. + +**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 +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. + +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. +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. 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 new file mode 100644 index 00000000..2f9ef028 --- /dev/null +++ b/experiments/codex_first_input/RESULTS.md @@ -0,0 +1,93 @@ +# Codex first-input App Server experiment — 2026-10-13 + +## 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. +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 +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. 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 +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..6ddcdb8e --- /dev/null +++ b/experiments/codex_first_input/probe.py @@ -0,0 +1,206 @@ +"""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. +Both modes require explicit opt-in to possible server history-DB initialization. +""" +import argparse +import json +import os +from pathlib import Path +import re +import stat +import time +import uuid + +from websockets.sync.client import unix_connect + + +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: + rpc_error_code(response["error"]) + raise ProbeError("rpc_rejected") + result = response.get("result") + if not isinstance(result, dict): + raise ProbeError("invalid_rpc_result") + 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 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"] + 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} + 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) + 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) + 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 + 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 type(message.get("id")) is int and message["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") + raise RuntimeError("notification limit") + hello = call("initialize", {"clientInfo": {"name": "session-peer-first-input-probe", "version": "0"}, "capabilities": {"experimentalApi": True}}) + 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 = rpc_result(call("thread/loaded/list", {"limit": 100})) + 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] + 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 + 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" + 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: + 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 + + +def run(argv=None): + state = {"attempted": False, "accepted": None} + try: + 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 new file mode 100644 index 00000000..044c3c36 --- /dev/null +++ b/tests/codex/test_first_input_probe.py @@ -0,0 +1,572 @@ +"""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 + + +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, 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 + + def __exit__(self, *unused): + return False + + 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] + 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": {"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) + return self.encode_reply(request, response) + else: + 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): + @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, 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) 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_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) + 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_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) + 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.assertIs(json.loads(output.splitlines()[-1])["accepted"], False) + self.assertFalse(json.loads(output.splitlines()[-1])["retryAllowed"]) + self.assertNotIn("PRIVATE", output) + + 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.", + "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][-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) + 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"]) + 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 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/read", "thread/turns/list", "thread/queue/list", "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}, + {"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) + 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.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_history_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): + 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.assertEqual(result["error"], "persistent_thread_required") + self.assertIs(result["accepted"], False) + self.assertEqual(result["outcome"], "not_attempted") + self.assertFalse(result["retryAllowed"]) + + 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"})): + 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, 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): + 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"), 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): + 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) + + 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_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')") + 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"}) + 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}, + {"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()