From a0dea84396181426082e7467f113db111ab49b68 Mon Sep 17 00:00:00 2001 From: YEONGUK LEE Date: Wed, 30 Sep 2026 19:45:53 +0900 Subject: [PATCH 1/4] feat(experiment): validate opt-in Codex side-session transport --- experiments/codex_side_session/README.md | 54 ++++ experiments/codex_side_session/RESULTS.md | 94 +++++++ .../codex_side_session/side_session.py | 253 ++++++++++++++++++ .../codex_side_session/test_side_session.py | 100 +++++++ 4 files changed, 501 insertions(+) create mode 100644 experiments/codex_side_session/README.md create mode 100644 experiments/codex_side_session/RESULTS.md create mode 100644 experiments/codex_side_session/side_session.py create mode 100644 experiments/codex_side_session/test_side_session.py diff --git a/experiments/codex_side_session/README.md b/experiments/codex_side_session/README.md new file mode 100644 index 00000000..06e517a0 --- /dev/null +++ b/experiments/codex_side_session/README.md @@ -0,0 +1,54 @@ +# Codex live side-session experiment (#245) + +Not installed or exposed by the canonical `session-peer` CLI. Existing package +rules exclude `experiments/`. No relay, SSH, native queue, generated runtime, +installation or policy changes are made by this prototype. + +This local Unix-domain-socket adapter connects only to an existing owning +Codex app-server. It never starts/resumes/forks a target. A private socket +directory and explicit experimental flag are required. The adapter uses +WebSocket-over-Unix sockets (including Codex's owned rendezvous symlink), +requires `websockets` (tested with 17.1), and gates server version to 0.159.2. +`list` reports loaded +ephemeral threads as **unclassified**, not necessarily `/btw` conversations. +For a real `/btw` test, establish its origin separately using the dedicated TUI. + +Example (dummy paths/UUID): + +```sh +python3 experiments/codex_side_session/side_session.py \ + --experimental-side-session --socket /private-test/app.sock list +python3 experiments/codex_side_session/side_session.py \ + --experimental-side-session --socket /private-test/app.sock send \ + --to codex:00000000-0000-4000-8000-000000000001 \ + --exclusive-test-session --dry-run --message 'test input' +``` + +## Critical limitation + +**Use only a dedicated test session with a single input producer.** Codex +0.159.2 `turn/start` calls `start_or_steer_turn`; an idle preflight is not an +atomic idle-only guard. Concurrent input can turn a intended new turn into +steering. `--exclusive-test-session` acknowledges this restriction; it does +not enforce exclusivity against other native clients. General use remains +blocked until safe native semantics or a cooperating owner can enforce it. + +**Loaded is not UI-open.** After the native TUI dismisses `/btw`, app-server +may retain the ephemeral thread as loaded and idle during its unload grace +period. The live test reproduced this: a dry-run remained validated after the +side UI closed. No further input was sent. A future integration needs an +owner-provided active registration/generation, not merely a loaded-ID check. + +No automatic steering, retries, parent fallback, approval responses, permission +overrides, or persistence of a side transcript. Acceptance is not ACK. The +dedicated live test may read its own TUI output to verify an exact response, +but this does not implement general wait/completion detection. + +Source review: session-peer base 90489fc; Codex CLI 0.159.2/upstream ff6aec9. +See [RESULTS.md](RESULTS.md) for actual roundtrip evidence and unmet gates. + +Unit tests: + +```sh +python3 -m unittest discover -s experiments/codex_side_session -v +``` diff --git a/experiments/codex_side_session/RESULTS.md b/experiments/codex_side_session/RESULTS.md new file mode 100644 index 00000000..c596a7da --- /dev/null +++ b/experiments/codex_side_session/RESULTS.md @@ -0,0 +1,94 @@ +# Side Session experiment results — 2026-09-30 + +## Verdict + +**Transport feasibility: PASS (one real request / one exact side-session ACK).** +**General-use safety and complete issue #245 acceptance: NOT PASS.** + +This is a local opt-in proof of concept in an isolated worktree based on +session-peer 90489fccc87fe2588a2546aaaf7f285429d2bcfd. Canonical CLI, generated +runtime, packaging configuration, existing agent sessions and hosted services +were not changed. No PR, push, package publication or rollout was performed. + +## Actual test, not a simulated fork + +- macOS arm64; Codex CLI/server 0.159.2; Python 3.14 environment with websockets + 17.1. TUI-selected default model GPT-6.1-Sol/high; no model override. +- A dedicated local app-server exposed a private Unix-domain endpoint. A fresh + native Codex TUI connected to that exact server. The new server/test thread + was configured read-only/on-request; no existing permission settings changed. +- A short parent setup request returned `ACK SIDE245-PARENT-READY-20260930`. +- Native `/btw` was entered once through the dedicated terminal. Its rendered + screen showed `Side from main thread`; the server inventory gained one + ephemeral, idle, direct-input-capable thread. No artificial API fork was + substituted for the real slash command. +- The experimental adapter's exact-target dry-run was validated with + `submitted:false`. It then submitted one new input via `turn/start`, with + no resume, fork, sandbox/model overrides or retry. Result: + `status:accepted`, `submitted:true`, `consumptionConfirmed:false`. +- The rendered native **side** screen displayed the request separately from + the response: `ACK SIDE245-ROUNDTRIP-A-20260930` (about 3 seconds of work, + displayed completion 19:36 KST). This response was actually generated by the + target, not inferred from submission. +- Parent history before/after: one turn, idle, identical serialized-turn hash: + `e08cdef71e4c1c171bee1c793ef1958954ad070020ef001ce02960023c443d82`. + The side request/ACK did not enter the parent transcript. + +ACK observation used the dedicated terminal's rendered screen. The adapter did +**not** implement generic ACK waiting, structured Reply-To return routing, or +delivery back into the original working Codex conversation. Acceptance output +correctly remained `consumptionConfirmed:false`; the exact ACK is separate test +evidence. Side history was never exported/persisted by the adapter. + +## Controls and limitations + +| Check | Outcome | +| --- | --- | +| Actual side discovery, dry-run and one-shot ACK | PASS | +| Parent transcript unchanged | PASS | +| Persistent parent selected by side adapter | Refused before input: `ephemeral_target_required` | +| Native `thread/queue/add` on real side ID | Rejected: JSON-RPC -32600 | +| Target absent from loaded inventory | Unit-test refusal; no fallback | +| Busy target at preflight | Unit-test refusal; actual concurrent race not tested | +| Timeout, malformed response, internal error after attempt | Unit-test unknown, no retry | +| Side dismissed from TUI but retained in server | **Unmet safety gate:** dry-run still validated | + +Two production blockers remain: + +1. `turn/start` is backed by `start_or_steer_turn`. A preflight idle check cannot + atomically prevent steering if another producer submits in between. The + exclusive-test flag acknowledges a single-producer requirement, but does not + enforce it against a native client. No adversarial concurrency test was run. +2. UI dismissal is not immediate server unload. Ctrl+C returned the TUI to its + parent, but the old side ID remained ephemeral/idle in memory, and a dry-run + still validated it. **No further message was sent after this finding.** An + owner-cooperating active-registration/generation mechanism is needed. + +`ephemeral` alone does not identify a `/btw` conversation. The experiment knew +its provenance because it controlled the dedicated TUI and observed creation. +Existing Orca-hosted private app servers, SSH, Linux, Windows, busy steering, +multi-client approval handling and normal automatic Reply-To are unverified. + +## Verification and cleanup + +- 13 scoped unit tests pass with `-X dev -W error::ResourceWarning`; no warnings. + One explicitly models the loaded-but-UI-closed limitation, without sending. +- Native WebSocket-over-Unix initialize/read/start was exercised with the real + server. Native stdio `app-server proxy` was not usable for the initial probe + (no initialize response); direct WebSocket transport was used instead. +- Existing package rules exclude `experiments/`; no artifact is being released. +- The two dedicated Orca terminals were closed, with PTY shutdown confirmed. +- The one newly created persisted parent test thread was deleted through its + owning server only after confirming exact cwd, one turn and unchanged hash. + Existing session records were not deleted. This test record is not retained; + its sanitized outcome is preserved here. +- Final checks: no Orca terminals in this worktree, no process whose command + references the unique test endpoint, no Codex process with this test cwd, and + no open physical test socket. Stale socket/rendezvous/startup-lock files created + by this test were removed by exact path after verifying no open handles. The + empty private scratch directory was removed. Unrelated daemon files and + existing worktrees/processes were preserved; the implementation worktree remains. + +References: [App Server](https://learn.chatgpt.com/docs/app-server), +[slash commands](https://learn.chatgpt.com/docs/developer-commands?surface=cli). +Source review is pinned to upstream ff6aec96948b70d94983af2641a6b67c94faeff5. diff --git a/experiments/codex_side_session/side_session.py b/experiments/codex_side_session/side_session.py new file mode 100644 index 00000000..1f178d6e --- /dev/null +++ b/experiments/codex_side_session/side_session.py @@ -0,0 +1,253 @@ +"""Local-only opt-in proof of concept; not part of the installed CLI. + +Requires exclusive control of a dedicated test session: turn/start can steer +if another caller wins the idle check race. Never use against a working session. +""" +import argparse +import json +import os +import stat +import sys +import time +import uuid +from pathlib import Path + + +class Refused(Exception): + pass + + +class Unknown(Exception): + pass + + +class RpcRejected(Refused): + def __init__(self, method, code): + super().__init__("rpc_rejected:" + method) + self.code = code + + +class Rpc: + MAX_FRAME = 1024 * 1024 + + def __init__(self, path, timeout=10): + self.path = Path(path) + self.timeout = timeout + self.socket = None + self.buffer = b"" + self.sequence = 0 + self.notifications = [] + + def __enter__(self): + if not self.path.is_absolute(): + raise Refused("absolute_socket_required") + parent = self.path.parent.stat() + link = self.path.lstat() + if link.st_uid != os.getuid(): + raise Refused("socket_not_owned") + resolved = self.path.resolve(strict=True) + value = resolved.lstat() + if not stat.S_ISSOCK(value.st_mode) or value.st_uid != os.getuid(): + raise Refused("socket_not_owned") + if parent.st_uid != os.getuid() or parent.st_mode & 0o077: + raise Refused("private_socket_directory_required") + resolved_parent = resolved.parent.stat() + if resolved_parent.st_uid != os.getuid() or resolved_parent.st_mode & 0o077: + raise Refused("private_socket_directory_required") + try: + from websockets.sync.client import unix_connect + except ImportError as exc: + raise Refused("websockets_dependency_missing") from exc + # Codex's local control socket carries WebSocket frames, not JSONL. + # Connect to the named server; never bootstrap a daemon. + self.connection = unix_connect(str(resolved), open_timeout=self.timeout, + close_timeout=1, max_size=self.MAX_FRAME) + try: + self.socket = self.connection.__enter__() + except Exception as exc: + raise Refused("connection_or_protocol_failed") from exc + try: + self.info = self.call("initialize", { + "clientInfo": {"name": "session_peer_side_experiment", + "title": "session-peer side experiment", "version": "0.0.0"}, + "capabilities": {"experimentalApi": True}, + }) + if not self.info.get("userAgent", "").startswith("codex-tui/0.159.2 "): + raise Refused("unsupported_server_version") + self.write({"method": "initialized"}) + return self + except BaseException: + self.__exit__() + raise + + def __exit__(self, *args): + self.connection.__exit__(*args if args else (None, None, None)) + + def write(self, message): + frame = json.dumps(message, ensure_ascii=False) + if len(frame.encode("utf-8")) > self.MAX_FRAME: + raise Refused("frame_too_large") + try: + self.socket.send(frame) + except Exception as exc: + raise ConnectionError("transport_closed") from exc + + def read(self, deadline): + remaining = deadline - time.monotonic() + if remaining <= 0: + raise TimeoutError() + try: + frame = self.socket.recv(timeout=remaining) + except TimeoutError: + raise + except Exception as exc: + raise ConnectionError("transport_closed") from exc + if not isinstance(frame, str): + raise Refused("invalid_frame") + message = json.loads(frame) + if not isinstance(message, dict): + raise Refused("invalid_rpc_response") + return message + + def call(self, method, params): + self.sequence += 1 + request_id = self.sequence + self.write({"id": request_id, "method": method, "params": params}) + deadline = time.monotonic() + self.timeout + while True: + message = self.read(deadline) + if message.get("id") == request_id: + if "error" in message: + # Never expose provider errors, transcript content or paths. + raise RpcRejected(method, message["error"].get("code")) + result = message.get("result") + if not isinstance(result, dict): + raise Refused("invalid_rpc_response") + return result + if "id" in message: + raise Refused("server_interaction_required") + if len(self.notifications) >= 100: + raise Refused("notification_limit") + self.notifications.append(message) + + +def loaded_ids(rpc): + ids = [] + cursor = None + for _ in range(32): + params = {"limit": 100} + if cursor is not None: + params["cursor"] = cursor + result = rpc.call("thread/loaded/list", params) + data = result.get("data") + if not isinstance(data, list) or any(not isinstance(x, str) for x in data): + raise Refused("invalid_loaded_list") + ids.extend(data) + next_cursor = result.get("nextCursor") + if next_cursor is None: + return ids + if not isinstance(next_cursor, str) or next_cursor == cursor: + raise Refused("invalid_cursor") + cursor = next_cursor + raise Refused("inventory_limit") + + +def target_info(rpc, target): + try: + thread_id = str(uuid.UUID(target.removeprefix("codex:"))) + except ValueError as exc: + raise Refused("full_thread_uuid_required") from exc + if thread_id not in loaded_ids(rpc): + raise Refused("target_not_loaded") + thread = rpc.call("thread/read", {"threadId": thread_id, "includeTurns": False}).get("thread") + if not isinstance(thread, dict) or thread.get("id") != thread_id: + raise Refused("target_mismatch") + if thread.get("ephemeral") is not True: + raise Refused("ephemeral_target_required") + status = thread.get("status") + if not isinstance(status, dict) or status.get("type") != "idle": + raise Refused("target_not_idle") + if thread.get("canAcceptDirectInput") is not True: + raise Refused("direct_input_unverified") + return thread_id + + +def send(rpc, target, text, *, dry_run=False, exclusive=False): + if not exclusive: + raise Refused("exclusive_test_session_required") + if not text or "\x00" in text or len(text.encode("utf-8")) > 65536: + raise Refused("invalid_message") + thread_id = target_info(rpc, target) + result = {"schemaVersion": 1, "ok": True, "experimental": True, + "transport": "codex-live-app-server", "submitted": False, + "consumptionConfirmed": False, "dryRun": dry_run} + if dry_run: + result["status"] = "validated" + return result + # No resume, fork, policy override, fallback, queue write, or retry. + # This is NOT atomic idle-only submission; see the module's scope warning. + try: + started = rpc.call("turn/start", { + "threadId": thread_id, + "input": [{"type": "text", "text": text}], + "clientUserMessageId": str(uuid.uuid4()), + }) + except RpcRejected as exc: + if exc.code in (-32600, -32601): + raise + raise Unknown("submission_outcome_unknown") from exc + except (OSError, ValueError, TimeoutError, Refused) as exc: + raise Unknown("submission_outcome_unknown") from exc + turn = started.get("turn") + if not isinstance(turn, dict) or not isinstance(turn.get("id"), str): + raise Unknown("submission_response_invalid") + result.update(status="accepted", submitted=True, turnId=turn["id"]) + return result + + +def main(argv=None): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--experimental-side-session", action="store_true", required=True) + parser.add_argument("--socket", required=True) + parser.add_argument("--timeout", type=float, default=10) + commands = parser.add_subparsers(dest="command", required=True) + commands.add_parser("list") + sending = commands.add_parser("send") + sending.add_argument("--to", required=True) + sending.add_argument("--message", required=True) + sending.add_argument("--dry-run", action="store_true") + sending.add_argument("--exclusive-test-session", action="store_true") + args = parser.parse_args(argv) + if not 0 < args.timeout <= 30: + parser.error("timeout must be greater than 0 and at most 30") + try: + with Rpc(args.socket, args.timeout) as rpc: + if args.command == "list": + rows = [] + for thread_id in loaded_ids(rpc): + t = rpc.call("thread/read", {"threadId": thread_id, "includeTurns": False})["thread"] + if t.get("ephemeral") is True: + rows.append({"id": thread_id, "ephemeral": True, "kind": "ephemeral_unclassified", + "status": t.get("status", {}).get("type"), + "canAcceptDirectInput": t.get("canAcceptDirectInput")}) + result = {"schemaVersion": 1, "ok": True, "experimental": True, "sessions": rows} + else: + text = sys.stdin.read(65537) if args.message == "-" else args.message + result = send(rpc, args.to, text, dry_run=args.dry_run, + exclusive=args.exclusive_test_session) + code = 0 + except Unknown as exc: + result = {"ok": False, "status": "unknown", "submitted": None, + "consumptionConfirmed": False, "reason": str(exc), "retryAllowed": False} + code = 1 + except (Refused, OSError, ValueError) as exc: + result = {"ok": False, "status": "refused", "submitted": False, + "consumptionConfirmed": False, + "reason": str(exc) if isinstance(exc, Refused) else "connection_or_protocol_failed"} + code = 1 + print(json.dumps(result, ensure_ascii=False)) + return code + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/experiments/codex_side_session/test_side_session.py b/experiments/codex_side_session/test_side_session.py new file mode 100644 index 00000000..85950980 --- /dev/null +++ b/experiments/codex_side_session/test_side_session.py @@ -0,0 +1,100 @@ +import unittest +from side_session import Refused, RpcRejected, Unknown, loaded_ids, send + + +TARGET = "00000000-0000-4000-8000-000000000001" + + +class Fake: + def __init__(self, ephemeral=True, status="idle", direct=True, loaded=True, fail=None): + self.calls = [] + self.ephemeral, self.status, self.direct, self.loaded, self.fail = ephemeral, status, direct, loaded, fail + + def call(self, method, params): + self.calls.append((method, params)) + if method == "thread/loaded/list": + return {"data": [TARGET] if self.loaded else [], "nextCursor": None} + if method == "thread/read": + return {"thread": {"id": TARGET, "ephemeral": self.ephemeral, + "status": {"type": self.status}, "canAcceptDirectInput": self.direct}} + if self.fail: + raise self.fail + return {"turn": {"id": "test-turn"}} + + +class Contract(unittest.TestCase): + def test_opt_in(self): + rpc = Fake() + with self.assertRaisesRegex(Refused, "exclusive"): + send(rpc, TARGET, "hello") + self.assertEqual(rpc.calls, []) + + def test_dry_run(self): + rpc = Fake() + self.assertFalse(send(rpc, TARGET, "hello", exclusive=True, dry_run=True)["submitted"]) + self.assertNotIn("turn/start", [m for m, _ in rpc.calls]) + + def test_accept_not_ack(self): + rpc = Fake() + result = send(rpc, "codex:" + TARGET, "hello", exclusive=True) + self.assertTrue(result["submitted"]) + self.assertFalse(result["consumptionConfirmed"]) + self.assertEqual([m for m, _ in rpc.calls], ["thread/loaded/list", "thread/read", "turn/start"]) + params = rpc.calls[-1][1] + self.assertEqual(set(params), {"threadId", "input", "clientUserMessageId"}) + self.assertFalse(rpc.calls[1][1]["includeTurns"]) + + def test_persistent_refused(self): + with self.assertRaisesRegex(Refused, "ephemeral"): + send(Fake(ephemeral=False), TARGET, "hello", exclusive=True) + + def test_busy_refused(self): + with self.assertRaisesRegex(Refused, "idle"): + send(Fake(status="active"), TARGET, "hello", exclusive=True) + + def test_unloaded_refused(self): + with self.assertRaisesRegex(Refused, "not_loaded"): + send(Fake(loaded=False), TARGET, "hello", exclusive=True) + + def test_loaded_idle_has_no_ui_lifetime_proof(self): + # Models the live closure finding. This is a documented limitation, + # not a test claiming that UI-closed destinations are safely refused. + rpc = Fake(loaded=True, ephemeral=True, status="idle") + result = send(rpc, TARGET, "hello", exclusive=True, dry_run=True) + self.assertEqual(result["status"], "validated") + self.assertNotIn("turn/start", [m for m, _ in rpc.calls]) + + def test_capability_refused(self): + with self.assertRaisesRegex(Refused, "unverified"): + send(Fake(direct=None), TARGET, "hello", exclusive=True) + + def test_timeout_unknown_no_retry(self): + rpc = Fake(fail=TimeoutError()) + with self.assertRaises(Unknown): + send(rpc, TARGET, "hello", exclusive=True) + self.assertEqual(sum(m == "turn/start" for m, _ in rpc.calls), 1) + + def test_rejection_no_retry(self): + rpc = Fake(fail=RpcRejected("turn/start", -32600)) + with self.assertRaises(Refused): + send(rpc, TARGET, "hello", exclusive=True) + self.assertEqual(sum(m == "turn/start" for m, _ in rpc.calls), 1) + + def test_internal_error_unknown(self): + with self.assertRaises(Unknown): + send(Fake(fail=RpcRejected("turn/start", -32603)), TARGET, "hello", exclusive=True) + + def test_invalid_response_unknown(self): + with self.assertRaises(Unknown): + send(Fake(fail=Refused("invalid_rpc_response")), TARGET, "hello", exclusive=True) + + def test_invalid_inputs(self): + for text in ("", "\x00", "가" * 22000): + with self.subTest(text_length=len(text)), self.assertRaises(Refused): + send(Fake(), TARGET, text, exclusive=True) + with self.assertRaises(Refused): + send(Fake(), "not-uuid", "hello", exclusive=True) + + +if __name__ == "__main__": + unittest.main() From 6baf066c8646db7f41a9acde0dc6ba376312c639 Mon Sep 17 00:00:00 2001 From: YEONGUK LEE Date: Sun, 4 Oct 2026 09:27:52 +0900 Subject: [PATCH 2/4] fix: validate experimental side-session RPC errors --- .../codex_side_session/side_session.py | 5 +- .../codex_side_session/test_side_session.py | 68 ++++++++++++++++++- 2 files changed, 71 insertions(+), 2 deletions(-) diff --git a/experiments/codex_side_session/side_session.py b/experiments/codex_side_session/side_session.py index 1f178d6e..631d15b4 100644 --- a/experiments/codex_side_session/side_session.py +++ b/experiments/codex_side_session/side_session.py @@ -119,7 +119,10 @@ def call(self, method, params): if message.get("id") == request_id: if "error" in message: # Never expose provider errors, transcript content or paths. - raise RpcRejected(method, message["error"].get("code")) + error = message["error"] + if not isinstance(error, dict) or type(error.get("code")) is not int: + raise Refused("invalid_rpc_response") + raise RpcRejected(method, error["code"]) result = message.get("result") if not isinstance(result, dict): raise Refused("invalid_rpc_response") diff --git a/experiments/codex_side_session/test_side_session.py b/experiments/codex_side_session/test_side_session.py index 85950980..dcea47d8 100644 --- a/experiments/codex_side_session/test_side_session.py +++ b/experiments/codex_side_session/test_side_session.py @@ -1,5 +1,6 @@ +import json import unittest -from side_session import Refused, RpcRejected, Unknown, loaded_ids, send +from side_session import Refused, Rpc, RpcRejected, Unknown, loaded_ids, send TARGET = "00000000-0000-4000-8000-000000000001" @@ -22,6 +23,71 @@ def call(self, method, params): return {"turn": {"id": "test-turn"}} +class FrameSocket: + """In-memory JSON frames through the real RPC parser; no server or model.""" + + def __init__(self, method, error): + self.method, self.error = method, error + self.calls = [] + + def send(self, frame): + request = json.loads(frame) + self.calls.append(request) + result = Fake().call(request["method"], request["params"]) + self.response = {"id": request["id"], "result": result} + if request["method"] == self.method: + # A plausible success result must not mask a malformed/error reply. + self.response["error"] = self.error + + def recv(self, timeout): + return json.dumps(self.response) + + +class RpcResponseContract(unittest.TestCase): + malformed_errors = (None, [], "private provider detail", {}, {"code": None}, + {"code": True}, {"code": False}, {"code": -32600.0}, + {"code": "-32600"}, {"code": []}, {"code": {}}) + + def rpc(self, method, error): + rpc = Rpc("/unused-fixture.sock") + rpc.socket = FrameSocket(method, error) + return rpc + + def test_malformed_error_before_submission_is_refused(self): + for error in self.malformed_errors: + with self.subTest(error=error): + rpc = self.rpc("thread/loaded/list", error) + with self.assertRaisesRegex(Refused, "^invalid_rpc_response$") as raised: + send(rpc, TARGET, "hello", exclusive=True) + self.assertIs(type(raised.exception), Refused) + self.assertEqual([r["method"] for r in rpc.socket.calls], ["thread/loaded/list"]) + + def test_malformed_error_after_attempt_is_unknown_without_resend(self): + for error in self.malformed_errors: + with self.subTest(error=error): + rpc = self.rpc("turn/start", error) + with self.assertRaisesRegex(Unknown, "^submission_outcome_unknown$"): + send(rpc, TARGET, "hello", exclusive=True) + self.assertEqual([r["method"] for r in rpc.socket.calls], + ["thread/loaded/list", "thread/read", "turn/start"]) + + def test_valid_error_codes_keep_existing_classification(self): + for code in (-32600, -32601, -32603, 0, 123): + with self.subTest(code=code): + rpc = self.rpc("turn/start", {"code": code, "message": "private provider detail"}) + expected = RpcRejected if code in (-32600, -32601) else Unknown + with self.assertRaises(expected) as raised: + send(rpc, TARGET, "hello", exclusive=True) + self.assertNotIn("private provider detail", str(raised.exception)) + self.assertEqual(sum(r["method"] == "turn/start" for r in rpc.socket.calls), 1) + + def test_absent_error_keeps_successful_result(self): + rpc = self.rpc("unused", None) + result = send(rpc, TARGET, "hello", exclusive=True) + self.assertTrue(result["submitted"]) + self.assertFalse(result["consumptionConfirmed"]) + + class Contract(unittest.TestCase): def test_opt_in(self): rpc = Fake() From 32d71b281eaf61a9e09667d7955594d54c2c0065 Mon Sep 17 00:00:00 2001 From: YEONGUK LEE Date: Thu, 22 Oct 2026 19:20:35 +0900 Subject: [PATCH 3/4] fix(experiment): preserve side-session outcomes on malformed replies --- experiments/codex_side_session/README.md | 7 + .../codex_side_session/side_session.py | 37 +++-- .../codex_side_session/test_side_session.py | 136 +++++++++++++++++- 3 files changed, 168 insertions(+), 12 deletions(-) diff --git a/experiments/codex_side_session/README.md b/experiments/codex_side_session/README.md index 06e517a0..86026481 100644 --- a/experiments/codex_side_session/README.md +++ b/experiments/codex_side_session/README.md @@ -13,6 +13,13 @@ requires `websockets` (tested with 17.1), and gates server version to 0.159.2. ephemeral threads as **unclassified**, not necessarily `/btw` conversations. For a real `/btw` test, establish its origin separately using the dedicated TUI. +Malformed RPC envelopes, error fields, deeply nested JSON and invalid discovery +metadata are rejected with fixed diagnostics. Once `turn/start` has been +attempted, an invalid response is `unknown`, with no automatic retry. An ordinary +connection-close error preserves an already validated submission or the original +unknown outcome; it does not establish non-submission. These boundaries are +covered by offline fake-transport tests, not a new native-version acceptance run. + Example (dummy paths/UUID): ```sh diff --git a/experiments/codex_side_session/side_session.py b/experiments/codex_side_session/side_session.py index 631d15b4..08629652 100644 --- a/experiments/codex_side_session/side_session.py +++ b/experiments/codex_side_session/side_session.py @@ -37,6 +37,7 @@ def __init__(self, path, timeout=10): self.buffer = b"" self.sequence = 0 self.notifications = [] + self.close_failed = False def __enter__(self): if not self.path.is_absolute(): @@ -72,7 +73,8 @@ def __enter__(self): "title": "session-peer side experiment", "version": "0.0.0"}, "capabilities": {"experimentalApi": True}, }) - if not self.info.get("userAgent", "").startswith("codex-tui/0.159.2 "): + agent = self.info.get("userAgent") + if not isinstance(agent, str) or not agent.startswith("codex-tui/0.159.2 "): raise Refused("unsupported_server_version") self.write({"method": "initialized"}) return self @@ -81,7 +83,12 @@ def __enter__(self): raise def __exit__(self, *args): - self.connection.__exit__(*args if args else (None, None, None)) + try: + self.connection.__exit__(*args if args else (None, None, None)) + except Exception: + # Cleanup cannot invalidate submission evidence or replace its error. + self.close_failed = True + return False def write(self, message): frame = json.dumps(message, ensure_ascii=False) @@ -104,7 +111,10 @@ def read(self, deadline): raise ConnectionError("transport_closed") from exc if not isinstance(frame, str): raise Refused("invalid_frame") - message = json.loads(frame) + try: + message = json.loads(frame) + except (ValueError, RecursionError) as exc: + raise Refused("invalid_rpc_response") from exc if not isinstance(message, dict): raise Refused("invalid_rpc_response") return message @@ -116,11 +126,14 @@ def call(self, method, params): deadline = time.monotonic() + self.timeout while True: message = self.read(deadline) - if message.get("id") == request_id: + if type(message.get("id")) is int and message["id"] == request_id: + if ("result" in message) == ("error" in message): + raise Refused("invalid_rpc_response") if "error" in message: # Never expose provider errors, transcript content or paths. error = message["error"] - if not isinstance(error, dict) or type(error.get("code")) is not int: + if (not isinstance(error, dict) or type(error.get("code")) is not int + or not isinstance(error.get("message"), str)): raise Refused("invalid_rpc_response") raise RpcRejected(method, error["code"]) result = message.get("result") @@ -228,11 +241,19 @@ def main(argv=None): if args.command == "list": rows = [] for thread_id in loaded_ids(rpc): - t = rpc.call("thread/read", {"threadId": thread_id, "includeTurns": False})["thread"] + t = rpc.call("thread/read", {"threadId": thread_id, "includeTurns": False}).get("thread") + if not isinstance(t, dict) or t.get("id") != thread_id: + raise Refused("invalid_thread_metadata") + status = t.get("status") + if not isinstance(status, dict) or not isinstance(status.get("type"), str): + raise Refused("invalid_thread_metadata") if t.get("ephemeral") is True: + direct = t.get("canAcceptDirectInput") + if direct is not None and type(direct) is not bool: + raise Refused("invalid_thread_metadata") rows.append({"id": thread_id, "ephemeral": True, "kind": "ephemeral_unclassified", - "status": t.get("status", {}).get("type"), - "canAcceptDirectInput": t.get("canAcceptDirectInput")}) + "status": status["type"], + "canAcceptDirectInput": direct}) result = {"schemaVersion": 1, "ok": True, "experimental": True, "sessions": rows} else: text = sys.stdin.read(65537) if args.message == "-" else args.message diff --git a/experiments/codex_side_session/test_side_session.py b/experiments/codex_side_session/test_side_session.py index dcea47d8..2b091b8f 100644 --- a/experiments/codex_side_session/test_side_session.py +++ b/experiments/codex_side_session/test_side_session.py @@ -1,6 +1,12 @@ import json +import io +import os +import stat import unittest -from side_session import Refused, Rpc, RpcRejected, Unknown, loaded_ids, send +from contextlib import redirect_stdout +from types import SimpleNamespace +from unittest.mock import patch, MagicMock +from side_session import Refused, Rpc, RpcRejected, Unknown, loaded_ids, send, main TARGET = "00000000-0000-4000-8000-000000000001" @@ -26,8 +32,9 @@ def call(self, method, params): class FrameSocket: """In-memory JSON frames through the real RPC parser; no server or model.""" - def __init__(self, method, error): + def __init__(self, method, error, both=False): self.method, self.error = method, error + self.both = both self.calls = [] def send(self, frame): @@ -36,7 +43,8 @@ def send(self, frame): result = Fake().call(request["method"], request["params"]) self.response = {"id": request["id"], "result": result} if request["method"] == self.method: - # A plausible success result must not mask a malformed/error reply. + if not self.both: + del self.response["result"] self.response["error"] = self.error def recv(self, timeout): @@ -46,7 +54,9 @@ def recv(self, timeout): class RpcResponseContract(unittest.TestCase): malformed_errors = (None, [], "private provider detail", {}, {"code": None}, {"code": True}, {"code": False}, {"code": -32600.0}, - {"code": "-32600"}, {"code": []}, {"code": {}}) + {"code": "-32600"}, {"code": []}, {"code": {}}, + {"code": -32600}, {"code": -32600, "message": None}, + {"code": -32600, "message": {"private": "detail"}}) def rpc(self, method, error): rpc = Rpc("/unused-fixture.sock") @@ -87,6 +97,124 @@ def test_absent_error_keeps_successful_result(self): self.assertTrue(result["submitted"]) self.assertFalse(result["consumptionConfirmed"]) + def test_ambiguous_envelope_never_proves_rejection(self): + for method in ("thread/loaded/list", "turn/start"): + rpc = self.rpc(method, {"code": -32600, "message": "private detail"}) + rpc.socket.both = True + expected = Unknown if method == "turn/start" else Refused + with self.subTest(method=method), self.assertRaises(expected) as raised: + send(rpc, TARGET, "hello", exclusive=True) + self.assertIs(type(raised.exception), expected) + self.assertNotIn("private detail", str(raised.exception)) + self.assertEqual(sum(r["method"] == "turn/start" for r in rpc.socket.calls), + int(method == "turn/start")) + + def test_response_ids_require_exact_integers(self): + for response_id in (True, 1.0, "1", None): + rpc = self.rpc("unused", None) + with patch.object(rpc.socket, "recv", return_value=json.dumps( + {"id": response_id, "result": {}})): + with self.subTest(response_id=response_id), self.assertRaises(Refused): + rpc.call("thread/loaded/list", {}) + + def test_bad_json_is_refused_before_attempt_and_unknown_after(self): + for frame in ('{"private detail":', '[' * 2000 + '0' + ']' * 2000): + for method in ("thread/loaded/list", "turn/start"): + rpc = self.rpc("unused", None) + original_recv = rpc.socket.recv + def recv(timeout): + return frame if rpc.socket.calls[-1]["method"] == method else original_recv(timeout) + with patch.object(rpc.socket, "recv", side_effect=recv): + expected = Unknown if method == "turn/start" else Refused + with self.subTest(method=method, size=len(frame)), self.assertRaises(expected) as raised: + send(rpc, TARGET, "hello", exclusive=True) + self.assertNotIn("private detail", str(raised.exception)) + + +class CliContract(unittest.TestCase): + def run_cli(self, rpc, command): + output = io.StringIO() + with patch("side_session.Rpc", return_value=rpc), redirect_stdout(output): + code = main(["--experimental-side-session", "--socket", "/unused", *command]) + self.assertEqual(code, 1) + return json.loads(output.getvalue()) + + def test_attempted_malformed_response_is_unknown_in_cli(self): + rpc = Rpc("/unused") + rpc.socket = FrameSocket("turn/start", {"code": -32600}) + context = MagicMock() + context.__enter__.return_value = rpc + result = self.run_cli(context, ["send", "--to", TARGET, "--message", "hello", + "--exclusive-test-session"]) + self.assertEqual(result["status"], "unknown") + self.assertIsNone(result["submitted"]) + self.assertFalse(result["retryAllowed"]) + self.assertEqual(sum(r["method"] == "turn/start" for r in rpc.socket.calls), 1) + + def test_malformed_discovery_metadata_is_private_refusal(self): + base = {"id": TARGET, "ephemeral": True, "status": {"type": "idle"}} + responses = [{}, {"thread": None}, {"thread": "private detail"}, + *({"thread": {**base, "status": status}} for status in + (None, [], "private detail", {}, {"type": {"private": "detail"}})), + {"thread": {**base, "canAcceptDirectInput": "private detail"}}] + for response in responses: + context = MagicMock() + context.__enter__.return_value.call.side_effect = [{"data": [TARGET]}, response] + with self.subTest(response=response): + result = self.run_cli(context, ["list"]) + self.assertEqual(result["status"], "refused") + self.assertFalse(result["submitted"]) + self.assertEqual(result["reason"], "invalid_thread_metadata") + self.assertNotIn("private detail", json.dumps(result)) + + def test_malformed_initialize_agent_is_refused(self): + client = SimpleNamespace(unix_connect=MagicMock()) + for agent in (None, [], {}, 1, True): + rpc = Rpc("/unused/socket") + with patch.dict("sys.modules", {"websockets.sync.client": client}), \ + patch("side_session.Path.stat", return_value=SimpleNamespace(st_uid=os.getuid(), st_mode=0o700)), \ + patch("side_session.Path.lstat", return_value=SimpleNamespace(st_uid=os.getuid(), st_mode=stat.S_IFSOCK)), \ + patch("side_session.Path.resolve", return_value=rpc.path), \ + patch.object(rpc, "call", return_value={"userAgent": agent}): + with self.subTest(agent=agent), self.assertRaisesRegex(Refused, "^unsupported_server_version$"): + rpc.__enter__() + + def test_cleanup_failure_preserves_submission_evidence(self): + for method, expected_status, expected_submitted in ( + ("unused", "accepted", True), + ("turn/start", "unknown", None), + ("thread/loaded/list", "refused", False)): + rpc = Rpc("/unused") + rpc.socket = FrameSocket(method, {"code": -32600}) + rpc.connection = MagicMock() + rpc.connection.__exit__.side_effect = OSError("private cleanup detail") + output = io.StringIO() + # Keep the real __exit__ while avoiding all native connections. + with self.subTest(method=method), \ + patch.object(Rpc, "__enter__", return_value=rpc), \ + patch("side_session.Rpc", return_value=rpc), redirect_stdout(output): + code = main(["--experimental-side-session", "--socket", "/unused", + "send", "--to", TARGET, "--message", "hello", + "--exclusive-test-session"]) + result = json.loads(output.getvalue()) + self.assertEqual(code, 0 if expected_submitted else 1) + self.assertEqual(result["status"], expected_status) + self.assertIs(result["submitted"], expected_submitted) + self.assertTrue(rpc.close_failed) + self.assertNotIn("private cleanup detail", output.getvalue()) + self.assertEqual(sum(r["method"] == "turn/start" for r in rpc.socket.calls), + 0 if method == "thread/loaded/list" else 1) + if expected_status == "unknown": + self.assertFalse(result["retryAllowed"]) + + def test_cleanup_does_not_swallow_process_interrupts(self): + for interruption in (KeyboardInterrupt, SystemExit): + rpc = Rpc("/unused") + rpc.connection = MagicMock() + rpc.connection.__exit__.side_effect = interruption() + with self.subTest(interruption=interruption), self.assertRaises(interruption): + rpc.__exit__() + class Contract(unittest.TestCase): def test_opt_in(self): From 667cd6b83fae30df1b50e463e1fd142f425e89a7 Mon Sep 17 00:00:00 2001 From: YEONGUK LEE Date: Thu, 22 Oct 2026 19:27:53 +0900 Subject: [PATCH 4/4] feat(experiment): bind side-session home and validate offline in CI --- .github/workflows/ci.yml | 9 ++ experiments/codex_side_session/README.md | 21 +++- .../codex_side_session/side_session.py | 18 ++- .../codex_side_session/test_side_session.py | 117 +++++++++++++++++- 4 files changed, 158 insertions(+), 7 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index b1fc612a..535e6860 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -72,6 +72,15 @@ jobs: run: python -m unittest discover -s tests/codex -t . -v - name: Distribution source tests run: python -m unittest discover -s tests/distribution -t . -v + - name: Offline Codex experiments (no native model calls) + if: matrix.os != 'windows-latest' && matrix.python == '3.13' + shell: bash + run: | + for suite in experiments/codex_side_session experiments/codex_steering; do + if test -d "$suite"; then + python -X dev -W error::ResourceWarning -m unittest discover -s "$suite" -p 'test_*.py' -v + fi + done - name: Complete default discovery contract if: matrix.os == 'ubuntu-latest' && matrix.python == '3.13' run: python -m unittest discover -s tests -v diff --git a/experiments/codex_side_session/README.md b/experiments/codex_side_session/README.md index 86026481..bd284455 100644 --- a/experiments/codex_side_session/README.md +++ b/experiments/codex_side_session/README.md @@ -9,6 +9,9 @@ Codex app-server. It never starts/resumes/forks a target. A private socket directory and explicit experimental flag are required. The adapter uses WebSocket-over-Unix sockets (including Codex's owned rendezvous symlink), requires `websockets` (tested with 17.1), and gates server version to 0.159.2. +The required absolute `--expected-home` must match the canonical `codexHome` +reported by that server before inventory or input. This metadata comparison +does not establish a server generation, atomic storage identity or side-UI lifetime. `list` reports loaded ephemeral threads as **unclassified**, not necessarily `/btw` conversations. For a real `/btw` test, establish its origin separately using the dedicated TUI. @@ -24,9 +27,11 @@ Example (dummy paths/UUID): ```sh python3 experiments/codex_side_session/side_session.py \ - --experimental-side-session --socket /private-test/app.sock list + --experimental-side-session --socket /private-test/app.sock \ + --expected-home /private-test/home list python3 experiments/codex_side_session/side_session.py \ - --experimental-side-session --socket /private-test/app.sock send \ + --experimental-side-session --socket /private-test/app.sock \ + --expected-home /private-test/home send \ --to codex:00000000-0000-4000-8000-000000000001 \ --exclusive-test-session --dry-run --message 'test input' ``` @@ -51,6 +56,18 @@ overrides, or persistence of a side transcript. Acceptance is not ACK. The dedicated live test may read its own TUI output to verify an exact response, but this does not implement general wait/completion detection. +## Review-ready experiment scope + +The offline suite runs explicitly in the Linux/macOS Python 3.13 CI jobs. It +uses fake filesystem and RPC responses, including private-endpoint, server-home, +version, approval-request and idle/start rejection cases. It never launches Codex +or sends a model request. The historical native evidence in `RESULTS.md` predates +the required home comparison and is not a rerun of the current adapter. + +Ready for code review means this bounded experiment can be reviewed with those +limitations visible. General `/btw` support still requires the atomic submission +and owner-lifecycle contracts described above; issue #245 remains open. + Source review: session-peer base 90489fc; Codex CLI 0.159.2/upstream ff6aec9. See [RESULTS.md](RESULTS.md) for actual roundtrip evidence and unmet gates. diff --git a/experiments/codex_side_session/side_session.py b/experiments/codex_side_session/side_session.py index 08629652..b8557754 100644 --- a/experiments/codex_side_session/side_session.py +++ b/experiments/codex_side_session/side_session.py @@ -30,8 +30,9 @@ def __init__(self, method, code): class Rpc: MAX_FRAME = 1024 * 1024 - def __init__(self, path, timeout=10): + def __init__(self, path, timeout=10, *, expected_home=None): self.path = Path(path) + self.expected_home = Path(expected_home) if expected_home is not None else None self.timeout = timeout self.socket = None self.buffer = b"" @@ -40,6 +41,8 @@ def __init__(self, path, timeout=10): self.close_failed = False def __enter__(self): + if self.expected_home is None or not self.expected_home.is_absolute(): + raise Refused("absolute_expected_home_required") if not self.path.is_absolute(): raise Refused("absolute_socket_required") parent = self.path.parent.stat() @@ -76,6 +79,15 @@ def __enter__(self): agent = self.info.get("userAgent") if not isinstance(agent, str) or not agent.startswith("codex-tui/0.159.2 "): raise Refused("unsupported_server_version") + advertised_home = self.info.get("codexHome") + if not isinstance(advertised_home, str) or not Path(advertised_home).is_absolute(): + raise Refused("invalid_server_home") + try: + home_matches = Path(advertised_home).resolve() == self.expected_home.resolve() + except (OSError, ValueError, RuntimeError) as exc: + raise Refused("invalid_server_home") from exc + if not home_matches: + raise Refused("server_home_mismatch") self.write({"method": "initialized"}) return self except BaseException: @@ -225,6 +237,8 @@ def main(argv=None): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--experimental-side-session", action="store_true", required=True) parser.add_argument("--socket", required=True) + parser.add_argument("--expected-home", required=True, + help="absolute Codex home expected in initialize metadata") parser.add_argument("--timeout", type=float, default=10) commands = parser.add_subparsers(dest="command", required=True) commands.add_parser("list") @@ -237,7 +251,7 @@ def main(argv=None): if not 0 < args.timeout <= 30: parser.error("timeout must be greater than 0 and at most 30") try: - with Rpc(args.socket, args.timeout) as rpc: + with Rpc(args.socket, args.timeout, expected_home=args.expected_home) as rpc: if args.command == "list": rows = [] for thread_id in loaded_ids(rpc): diff --git a/experiments/codex_side_session/test_side_session.py b/experiments/codex_side_session/test_side_session.py index 2b091b8f..592b14a9 100644 --- a/experiments/codex_side_session/test_side_session.py +++ b/experiments/codex_side_session/test_side_session.py @@ -3,8 +3,9 @@ import os import stat import unittest -from contextlib import redirect_stdout +from contextlib import redirect_stdout, contextmanager from types import SimpleNamespace +from pathlib import Path from unittest.mock import patch, MagicMock from side_session import Refused, Rpc, RpcRejected, Unknown, loaded_ids, send, main @@ -135,7 +136,8 @@ class CliContract(unittest.TestCase): def run_cli(self, rpc, command): output = io.StringIO() with patch("side_session.Rpc", return_value=rpc), redirect_stdout(output): - code = main(["--experimental-side-session", "--socket", "/unused", *command]) + code = main(["--experimental-side-session", "--socket", "/unused", + "--expected-home", "/expected", *command]) self.assertEqual(code, 1) return json.loads(output.getvalue()) @@ -170,7 +172,7 @@ def test_malformed_discovery_metadata_is_private_refusal(self): def test_malformed_initialize_agent_is_refused(self): client = SimpleNamespace(unix_connect=MagicMock()) for agent in (None, [], {}, 1, True): - rpc = Rpc("/unused/socket") + rpc = Rpc("/unused/socket", expected_home="/expected") with patch.dict("sys.modules", {"websockets.sync.client": client}), \ patch("side_session.Path.stat", return_value=SimpleNamespace(st_uid=os.getuid(), st_mode=0o700)), \ patch("side_session.Path.lstat", return_value=SimpleNamespace(st_uid=os.getuid(), st_mode=stat.S_IFSOCK)), \ @@ -194,6 +196,7 @@ def test_cleanup_failure_preserves_submission_evidence(self): patch.object(Rpc, "__enter__", return_value=rpc), \ patch("side_session.Rpc", return_value=rpc), redirect_stdout(output): code = main(["--experimental-side-session", "--socket", "/unused", + "--expected-home", "/expected", "send", "--to", TARGET, "--message", "hello", "--exclusive-test-session"]) result = json.loads(output.getvalue()) @@ -216,6 +219,114 @@ def test_cleanup_does_not_swallow_process_interrupts(self): rpc.__exit__() +class ConnectionBoundary(unittest.TestCase): + @contextmanager + def fixture(self, *, expected="/expected", advertised="/expected", agent="codex-tui/0.159.2 test", + parent_uid=None, parent_mode=0o700, link_uid=None, target_uid=None, + target_mode=stat.S_IFSOCK, resolved_parent_uid=None, resolved_parent_mode=0o700): + uid = os.getuid() + owner = lambda value: uid if value is None else value + rpc = Rpc("/rendezvous/socket", expected_home=expected) + client = SimpleNamespace(unix_connect=MagicMock()) + def resolve(path, *args, **kwargs): + if str(path) == "/rendezvous/socket": + return Path("/resolved/socket") + return Path(os.path.normpath(str(path))) + with patch.dict("sys.modules", {"websockets.sync.client": client}), \ + patch("side_session.Path.stat", side_effect=[ + SimpleNamespace(st_uid=owner(parent_uid), st_mode=parent_mode), + SimpleNamespace(st_uid=owner(resolved_parent_uid), st_mode=resolved_parent_mode)]), \ + patch("side_session.Path.lstat", side_effect=[ + SimpleNamespace(st_uid=owner(link_uid), st_mode=stat.S_IFLNK), + SimpleNamespace(st_uid=owner(target_uid), st_mode=target_mode)]), \ + patch("side_session.Path.resolve", new=resolve), \ + patch.object(rpc, "call", return_value={"userAgent": agent, "codexHome": advertised}) as call, \ + patch.object(rpc, "write") as write: + yield rpc, client, call, write + + def test_socket_ownership_and_directory_boundaries(self): + other = os.getuid() + 1 + cases = ({"link_uid": other}, {"target_uid": other}, {"target_mode": stat.S_IFREG}, + {"parent_uid": other}, {"parent_mode": 0o750}, + {"resolved_parent_uid": other}, {"resolved_parent_mode": 0o707}) + for case in cases: + with self.subTest(case=case), self.fixture(**case) as (rpc, client, call, write): + with self.assertRaises(Refused): + rpc.__enter__() + client.unix_connect.assert_not_called() + call.assert_not_called() + write.assert_not_called() + + def test_absolute_expected_home_is_required_before_connect(self): + for expected in (None, "", "relative/home"): + with self.subTest(expected=expected), self.fixture(expected=expected) as (rpc, client, call, write): + with self.assertRaisesRegex(Refused, "^absolute_expected_home_required$"): + rpc.__enter__() + client.unix_connect.assert_not_called() + call.assert_not_called() + + def test_advertised_home_is_checked_before_inventory_or_input(self): + for advertised in (None, {}, [], True, "", "private relative detail", "/other-private-home"): + with self.subTest(advertised=advertised), self.fixture(advertised=advertised) as (rpc, client, call, write): + with self.assertRaises(Refused) as raised: + rpc.__enter__() + self.assertIn(str(raised.exception), ("invalid_server_home", "server_home_mismatch")) + self.assertEqual([c.args[0] for c in call.call_args_list], ["initialize"]) + write.assert_not_called() + client.unix_connect.return_value.__exit__.assert_called_once() + + def test_home_comparison_uses_canonical_paths(self): + with self.fixture(expected="/expected/sub/..") as (rpc, client, call, write): + with rpc: + self.assertEqual(rpc.info["codexHome"], "/expected") + write.assert_called_once_with({"method": "initialized"}) + self.assertEqual([c.args[0] for c in call.call_args_list], ["initialize"]) + + def test_other_server_versions_are_refused_before_inventory_or_input(self): + for agent in ("codex-tui/0.159.3 private detail", "codex-tui/0.159.2", "private detail"): + with self.subTest(agent=agent), self.fixture(agent=agent) as (rpc, client, call, write): + with self.assertRaisesRegex(Refused, "^unsupported_server_version$"): + rpc.__enter__() + self.assertEqual([c.args[0] for c in call.call_args_list], ["initialize"]) + write.assert_not_called() + + def test_incoming_server_request_is_never_answered(self): + for method in ("thread/loaded/list", "turn/start"): + rpc = Rpc("/unused") + rpc.socket = FrameSocket("unused", None) + original_recv = rpc.socket.recv + def recv(timeout): + if rpc.socket.calls[-1]["method"] == method: + return json.dumps({"id": 999, "method": "item/commandExecution/requestApproval", + "params": {"private": "detail"}}) + return original_recv(timeout) + with self.subTest(method=method), patch.object(rpc.socket, "recv", side_effect=recv): + expected = Unknown if method == "turn/start" else Refused + with self.assertRaises(expected) as raised: + send(rpc, TARGET, "hello", exclusive=True) + self.assertNotIn("private", str(raised.exception)) + self.assertEqual([r["method"] for r in rpc.socket.calls], + ["thread/loaded/list", "thread/read", "turn/start"] + if method == "turn/start" else ["thread/loaded/list"]) + + def test_idle_preflight_does_not_close_native_start_race(self): + # A competitor can become active after our idle read. Native failure + # classification remains code-dependent, without a retry or UI proof. + for code in (-32600, -32601, -32603, -32000): + rpc = Rpc("/unused") + rpc.socket = FrameSocket("turn/start", {"code": code, "message": "native busy"}) + expected = RpcRejected if code in (-32600, -32601) else Unknown + with self.subTest(code=code), self.assertRaises(expected) as raised: + send(rpc, TARGET, "hello", exclusive=True) + self.assertIs(type(raised.exception), expected) + if isinstance(raised.exception, RpcRejected): + self.assertEqual(raised.exception.code, code) + else: + self.assertEqual(raised.exception.__cause__.code, code) + self.assertEqual([r["method"] for r in rpc.socket.calls], + ["thread/loaded/list", "thread/read", "turn/start"]) + + class Contract(unittest.TestCase): def test_opt_in(self): rpc = Fake()