From 373da984a00cb883e31f41c4743930a59bd4e922 Mon Sep 17 00:00:00 2001 From: dimavrem22 Date: Mon, 3 Aug 2026 06:27:56 +0000 Subject: [PATCH 1/3] Fix app-server reader recovery for large messages --- .env.example | 1 + README.md | 1 + inkbox_codex/__init__.py | 2 +- inkbox_codex/codex_client.py | 172 ++++++++++++++++++++------ inkbox_codex/codex_usage.py | 2 +- inkbox_codex/config.py | 9 ++ inkbox_codex/mcp_stdio.py | 2 +- inkbox_codex/sessions.py | 5 +- pyproject.toml | 2 +- tests/contract/test_host_interface.py | 2 +- tests/test_codex_client_stream.py | 140 +++++++++++++++++++++ tests/test_config.py | 13 +- tests/test_sessions.py | 41 ++++++ 13 files changed, 344 insertions(+), 48 deletions(-) create mode 100644 tests/test_codex_client_stream.py diff --git a/.env.example b/.env.example index 0d41310..5b44eee 100644 --- a/.env.example +++ b/.env.example @@ -38,4 +38,5 @@ CODEX_PROJECT_DIR=/path/to/the/repo/codex/should/work/in # CODEX_BIN=codex # CODEX_SANDBOX=workspace-write # CODEX_APPROVAL_POLICY=on-request +# CODEX_APP_SERVER_STREAM_LIMIT_BYTES=16777216 # INKBOX_PERMISSION_TIMEOUT_S=600 diff --git a/README.md b/README.md index 50e544a..7d038fe 100644 --- a/README.md +++ b/README.md @@ -272,6 +272,7 @@ curl --fail-with-body --request POST 'https://your-agent-host.example/webhook' \ | `CODEX_BIN` | no | `codex` | Codex CLI executable to run. | | `CODEX_SANDBOX` | no | `workspace-write` | App-server thread sandbox (`read-only`, `workspace-write`, `danger-full-access`). | | `CODEX_APPROVAL_POLICY` | no | `on-request` | Codex approval policy for bridged turns. | +| `CODEX_APP_SERVER_STREAM_LIMIT_BYTES` | no | `16777216` | Maximum size of one newline-delimited app-server message. | | `INKBOX_REALTIME_ENABLED` | no | `false` | Use OpenAI Realtime for calls. Needs a key; off → Inkbox STT/TTS. | | `INKBOX_REALTIME_API_KEY` | realtime | `OPENAI_API_KEY` | OpenAI key with `/v1/realtime` access. | | `INKBOX_REALTIME_MODEL` | no | `gpt-realtime-2` | Realtime model id. | diff --git a/inkbox_codex/__init__.py b/inkbox_codex/__init__.py index 669e62e..3ee6fe2 100644 --- a/inkbox_codex/__init__.py +++ b/inkbox_codex/__init__.py @@ -1,3 +1,3 @@ """Inkbox bridge for Codex — email, SMS, iMessage, and voice.""" -__version__ = "0.2.8" +__version__ = "0.2.9" diff --git a/inkbox_codex/codex_client.py b/inkbox_codex/codex_client.py index 53ee280..bb78cfa 100644 --- a/inkbox_codex/codex_client.py +++ b/inkbox_codex/codex_client.py @@ -85,8 +85,20 @@ def __init__( self._pending: Dict[int, "asyncio.Future[Any]"] = {} self._turns: Dict[str, _TurnCapture] = {} self._current_turn_id: Optional[str] = None + self._turn_start_pending = False + self._early_turn_notifications: Dict[str, list[Dict[str, Any]]] = {} self._initialized = False + @property + def is_healthy(self) -> bool: + """Return whether the subprocess and its stdout reader are usable.""" + return ( + self._proc is not None + and self._proc.returncode is None + and self._reader_task is not None + and not self._reader_task.done() + ) + async def connect(self, resume_thread_id: Optional[str] = None) -> str: """Start app-server and create or resume a Codex thread.""" await self._ensure_process() @@ -114,21 +126,36 @@ async def run_detailed(self, text: str) -> CodexTurnResult: """Run one turn and return its final reply and sanitized MCP outcomes.""" if not self.thread_id: await self.connect() + elif not self.is_healthy: + raise CodexAppServerError("Codex app-server reader is not running") assert self.thread_id is not None - result = await self._request( - "turn/start", - { - "threadId": self.thread_id, - "input": [{"type": "text", "text": text}], - "cwd": self.cfg.project_dir or None, - "model": self.cfg.codex_model or None, - "approvalPolicy": self.cfg.codex_approval_policy or "on-request", - }, - ) + self._turn_start_pending = True + self._early_turn_notifications.clear() + try: + result = await self._request( + "turn/start", + { + "threadId": self.thread_id, + "input": [{"type": "text", "text": text}], + "cwd": self.cfg.project_dir or None, + "model": self.cfg.codex_model or None, + "approvalPolicy": self.cfg.codex_approval_policy or "on-request", + }, + ) + except BaseException: + self._turn_start_pending = False + self._early_turn_notifications.clear() + raise + if not self.is_healthy: + self._turn_start_pending = False + self._early_turn_notifications.clear() + raise CodexAppServerError("Codex app-server reader is not running") turn = result.get("turn") or {} turn_id = str(turn.get("id") or "") if not turn_id: + self._turn_start_pending = False + self._early_turn_notifications.clear() raise CodexAppServerError(f"app-server did not return a turn id: {result!r}") loop = asyncio.get_running_loop() @@ -139,6 +166,11 @@ async def run_detailed(self, text: str) -> CodexTurnResult: ) self._turns[turn_id] = capture self._current_turn_id = turn_id + early_notifications = self._early_turn_notifications.pop(turn_id, []) + self._turn_start_pending = False + self._early_turn_notifications.clear() + for notification in early_notifications: + self._handle_notification(notification) try: return await capture.future finally: @@ -166,18 +198,43 @@ async def disconnect(self) -> None: capture.future.set_exception(CodexAppServerError("Codex app-server disconnected")) self._turns.clear() - if self._proc is not None and self._proc.returncode is None: - self._proc.terminate() + proc = self._proc + if proc is not None and proc.stdin is not None: + proc.stdin.close() + try: + await asyncio.wait_for(proc.stdin.wait_closed(), timeout=1) + except (asyncio.TimeoutError, BrokenPipeError, ConnectionResetError): + pass + if proc is not None and proc.returncode is None: + try: + proc.terminate() + except ProcessLookupError: + pass + if proc is not None: try: - await asyncio.wait_for(self._proc.wait(), timeout=5) + await asyncio.wait_for(proc.wait(), timeout=5) except asyncio.TimeoutError: - self._proc.kill() - await self._proc.wait() - if self._reader_task is not None: - self._reader_task.cancel() - if self._stderr_task is not None: - self._stderr_task.cancel() + proc.kill() + await proc.wait() + background_tasks = [ + task + for task in (self._reader_task, self._stderr_task) + if task is not None + ] + for task in background_tasks: + if not task.done(): + task.cancel() + if background_tasks: + await asyncio.gather(*background_tasks, return_exceptions=True) + await asyncio.sleep(0) self._proc = None + self._reader_task = None + self._stderr_task = None + self.thread_id = None + self._current_turn_id = None + self._turn_start_pending = False + self._early_turn_notifications.clear() + self._initialized = False def _thread_params(self) -> Dict[str, Any]: config: Dict[str, Any] = {} @@ -195,8 +252,10 @@ def _thread_params(self) -> Dict[str, Any]: } async def _ensure_process(self) -> None: - if self._proc is not None and self._proc.returncode is None: + if self.is_healthy: return + if self._proc is not None: + await self.disconnect() env = os.environ.copy() self._proc = await asyncio.create_subprocess_exec( self.cfg.codex_bin or "codex", @@ -205,6 +264,7 @@ async def _ensure_process(self) -> None: stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, env=env, + limit=max(1, int(self.cfg.codex_app_server_stream_limit_bytes)), ) self._reader_task = asyncio.create_task(self._reader_loop()) self._stderr_task = asyncio.create_task(self._stderr_loop()) @@ -216,7 +276,7 @@ async def _initialize(self) -> None: "clientInfo": { "name": "inkbox_codex", "title": "Inkbox Codex Bridge", - "version": "0.2.8", + "version": "0.2.9", }, "capabilities": {"experimentalApi": True}, }, @@ -225,7 +285,7 @@ async def _initialize(self) -> None: self._initialized = True async def _request(self, method: str, params: Dict[str, Any]) -> Any: - if self._proc is None or self._proc.stdin is None: + if not self.is_healthy or self._proc is None or self._proc.stdin is None: raise CodexAppServerError("Codex app-server is not running") message_id = self._next_id self._next_id += 1 @@ -245,25 +305,42 @@ def _write(self, message: Dict[str, Any]) -> None: async def _reader_loop(self) -> None: assert self._proc is not None and self._proc.stdout is not None - while True: - line = await self._proc.stdout.readline() - if not line: - self._fail_all(CodexAppServerError("Codex app-server exited")) - return - try: - message = json.loads(line.decode()) - except json.JSONDecodeError: - logger.warning("invalid app-server JSON: %r", line[:500]) - continue - - if "id" in message and ("result" in message or "error" in message) and "method" not in message: - self._handle_response(message) - continue - if "id" in message and "method" in message: - asyncio.create_task(self._handle_server_request(message)) - continue - if "method" in message: - self._handle_notification(message) + try: + while True: + line = await self._proc.stdout.readline() + if not line: + raise CodexAppServerError("Codex app-server stdout closed") + try: + message = json.loads(line.decode()) + except (UnicodeDecodeError, json.JSONDecodeError): + logger.warning("invalid app-server JSON: %r", line[:500]) + continue + + if "id" in message and ("result" in message or "error" in message) and "method" not in message: + self._handle_response(message) + continue + if "id" in message and "method" in message: + asyncio.create_task(self._handle_server_request(message)) + continue + if "method" in message: + self._handle_notification(message) + except asyncio.CancelledError: + raise + except Exception as exc: + logger.exception("Codex app-server reader stopped unexpectedly") + self._fail_all( + CodexAppServerError(f"Codex app-server reader failed: {exc}") + ) + if self._proc is not None and self._proc.returncode is None: + # A line-limit failure pauses the pipe; close it so the child + # cannot keep teardown blocked while its stdout is unreadable. + stdout_transport = getattr(self._proc.stdout, "_transport", None) + if stdout_transport is not None: + stdout_transport.close() + try: + self._proc.terminate() + except ProcessLookupError: + pass async def _stderr_loop(self) -> None: assert self._proc is not None and self._proc.stderr is not None @@ -306,6 +383,19 @@ def _handle_notification(self, message: Dict[str, Any]) -> None: params = message.get("params") or {} turn_id = str(params.get("turnId") or (params.get("turn") or {}).get("id") or "") + if ( + turn_id + and turn_id not in self._turns + and self._turn_start_pending + and method in { + "item/agentMessage/delta", + "item/completed", + "turn/completed", + } + ): + self._early_turn_notifications.setdefault(turn_id, []).append(message) + return + if method == "item/agentMessage/delta": capture = self._turns.get(turn_id) if capture is not None: diff --git a/inkbox_codex/codex_usage.py b/inkbox_codex/codex_usage.py index 8c38931..2db37d0 100644 --- a/inkbox_codex/codex_usage.py +++ b/inkbox_codex/codex_usage.py @@ -33,7 +33,7 @@ def _request_account_usage(codex_bin: str = "codex", timeout: float = 10.0) -> d "clientInfo": { "name": "inkbox_codex_usage", "title": "Inkbox Codex Usage", - "version": "0.2.8", + "version": "0.2.9", }, "capabilities": {"experimentalApi": True}, }, diff --git a/inkbox_codex/config.py b/inkbox_codex/config.py index 314138f..3d7cffb 100644 --- a/inkbox_codex/config.py +++ b/inkbox_codex/config.py @@ -26,6 +26,7 @@ DEFAULT_HOST = "0.0.0.0" DEFAULT_PORT = 8767 DEFAULT_WEBHOOK_PATH = "/webhook" +DEFAULT_CODEX_APP_SERVER_STREAM_LIMIT_BYTES = 16 * 1024 * 1024 class VoiceStack(str, Enum): @@ -125,6 +126,9 @@ class BridgeConfig: permission_timeout_s: float = 600.0 codex_turn_timeout_s: float = 1800.0 codex_interrupt_timeout_s: float = 10.0 + codex_app_server_stream_limit_bytes: int = ( + DEFAULT_CODEX_APP_SERVER_STREAM_LIMIT_BYTES + ) voice_stack: VoiceStack = VoiceStack.INKBOX_TTS_STT voice_stack_invalid_value: str = "" voice_ai_authority_mode: str = "contact_scoped" @@ -244,6 +248,11 @@ def read_config(extra: Dict[str, Any] | None = None) -> BridgeConfig: permission_timeout_s=float(os.getenv("INKBOX_PERMISSION_TIMEOUT_S") or 600.0), codex_turn_timeout_s=float(os.getenv("CODEX_TURN_TIMEOUT_S") or 1800.0), codex_interrupt_timeout_s=float(os.getenv("CODEX_INTERRUPT_TIMEOUT_S") or 10.0), + codex_app_server_stream_limit_bytes=int( + extra.get("codex_app_server_stream_limit_bytes") + or os.getenv("CODEX_APP_SERVER_STREAM_LIMIT_BYTES") + or DEFAULT_CODEX_APP_SERVER_STREAM_LIMIT_BYTES + ), voice_stack=voice_stack, voice_stack_invalid_value=invalid_voice_stack, voice_ai_authority_mode=str( diff --git a/inkbox_codex/mcp_stdio.py b/inkbox_codex/mcp_stdio.py index 9ba697b..500a1c4 100644 --- a/inkbox_codex/mcp_stdio.py +++ b/inkbox_codex/mcp_stdio.py @@ -58,7 +58,7 @@ async def handle(self, message: Dict[str, Any]) -> Dict[str, Any] | None: "capabilities": {"tools": {}}, "serverInfo": { "name": "inkbox-codex", - "version": "0.2.8", + "version": "0.2.9", }, }, ) diff --git a/inkbox_codex/sessions.py b/inkbox_codex/sessions.py index b1c6565..2595438 100644 --- a/inkbox_codex/sessions.py +++ b/inkbox_codex/sessions.py @@ -874,8 +874,11 @@ async def _typing_loop(self) -> None: return async def _ensure_client(self) -> CodexAppServerClient: - if self._client is not None: + if self._client is not None and getattr(self._client, "is_healthy", True): return self._client + if self._client is not None: + await self._client.disconnect() + self._client = None developer_instructions = build_channel_prompt( project_dir=self.cfg.project_dir, identity_handle=self.identity_info.get("handle", ""), diff --git a/pyproject.toml b/pyproject.toml index 2729751..6bafef9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "codex-plugin" -version = "0.2.8" +version = "0.2.9" description = "Inkbox bridge for Codex — talk to your coding agent over email, SMS, iMessage, and voice" requires-python = ">=3.11" dependencies = [ diff --git a/tests/contract/test_host_interface.py b/tests/contract/test_host_interface.py index 89973c3..ba05b96 100644 --- a/tests/contract/test_host_interface.py +++ b/tests/contract/test_host_interface.py @@ -97,7 +97,7 @@ def raw(tmp_path, monkeypatch): import os session = _RawAppServer(env=dict(os.environ)) session.request("initialize", { - "clientInfo": {"name": "inkbox_codex", "title": "Inkbox Codex Bridge", "version": "0.2.8"}, + "clientInfo": {"name": "inkbox_codex", "title": "Inkbox Codex Bridge", "version": "0.2.9"}, "capabilities": {"experimentalApi": True}, }) session.notify("initialized", {}) diff --git a/tests/test_codex_client_stream.py b/tests/test_codex_client_stream.py new file mode 100644 index 0000000..406ed53 --- /dev/null +++ b/tests/test_codex_client_stream.py @@ -0,0 +1,140 @@ +import asyncio +import os +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from inkbox_codex.codex_client import CodexAppServerError, CodexAppServerClient +from inkbox_codex.config import BridgeConfig + + +MOCK_APP_SERVER = r'''#!/usr/bin/env python3 +import json +import sys + + +def send(message): + sys.stdout.write(json.dumps(message, separators=(",", ":")) + "\n") + sys.stdout.flush() + + +for raw in sys.stdin: + message = json.loads(raw) + method = message.get("method") + message_id = message.get("id") + if method == "initialize": + send({"id": message_id, "result": {}}) + elif method == "thread/start": + send({"id": message_id, "result": {"thread": {"id": "thread-1"}}}) + elif method == "turn/start": + send({"id": message_id, "result": {"turn": {"id": "turn-1"}}}) + reply = "reply:" + ("x" * 70000) + send({ + "method": "item/completed", + "params": { + "threadId": "thread-1", + "turnId": "turn-1", + "item": {"type": "agentMessage", "text": reply}, + }, + }) + send({ + "method": "turn/completed", + "params": { + "threadId": "thread-1", + "turn": { + "id": "turn-1", + "status": "completed", + "items": [{"type": "commandExecution", "output": "y" * 70000}], + }, + }, + }) +''' + + +def _mock_app_server(tmp_path: Path) -> Path: + executable = tmp_path / "mock-codex" + executable.write_text(MOCK_APP_SERVER) + executable.chmod(executable.stat().st_mode | 0o111) + return executable + + +def _client(codex_bin: Path, *, stream_limit: int) -> CodexAppServerClient: + return CodexAppServerClient( + BridgeConfig( + codex_bin=os.fspath(codex_bin), + codex_app_server_stream_limit_bytes=stream_limit, + ), + developer_instructions="test", + ) + + +def test_large_item_and_turn_notifications_complete(tmp_path): + async def scenario(): + client = _client(_mock_app_server(tmp_path), stream_limit=16 * 1024 * 1024) + try: + await client.connect() + result = await asyncio.wait_for(client.run_detailed("test"), timeout=2) + assert result.text.startswith("reply:") + assert len(result.text) > 65536 + assert client.is_healthy is True + finally: + await client.disconnect() + + asyncio.run(scenario()) + + +def test_reader_limit_failure_fails_turn_and_marks_client_unhealthy(tmp_path): + async def scenario(): + client = _client(_mock_app_server(tmp_path), stream_limit=1024) + try: + await client.connect() + with pytest.raises(CodexAppServerError, match="reader"): + await asyncio.wait_for(client.run_detailed("test"), timeout=2) + assert client.is_healthy is False + finally: + await client.disconnect() + + asyncio.run(scenario()) + + +def test_notifications_before_turn_start_response_are_replayed(): + async def scenario(): + client = CodexAppServerClient( + BridgeConfig(), + developer_instructions="test", + ) + reader_blocker = asyncio.create_task(asyncio.Event().wait()) + client.thread_id = "thread-1" + client._proc = SimpleNamespace(returncode=None) + client._reader_task = reader_blocker + + async def request(method, _params): + assert method == "turn/start" + client._handle_notification({ + "method": "item/completed", + "params": { + "threadId": "thread-1", + "turnId": "turn-1", + "item": {"type": "agentMessage", "text": "early reply"}, + }, + }) + client._handle_notification({ + "method": "turn/completed", + "params": { + "threadId": "thread-1", + "turn": {"id": "turn-1", "status": "completed"}, + }, + }) + return {"turn": {"id": "turn-1"}} + + client._request = request + try: + result = await asyncio.wait_for(client.run_detailed("test"), timeout=1) + assert result.text == "early reply" + finally: + reader_blocker.cancel() + with pytest.raises(asyncio.CancelledError): + await reader_blocker + + asyncio.run(scenario()) diff --git a/tests/test_config.py b/tests/test_config.py index 57541e9..5981577 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -1,4 +1,8 @@ -from inkbox_codex.config import VoiceStack, read_config +from inkbox_codex.config import ( + DEFAULT_CODEX_APP_SERVER_STREAM_LIMIT_BYTES, + VoiceStack, + read_config, +) def test_read_config_defaults(monkeypatch): @@ -7,6 +11,7 @@ def test_read_config_defaults(monkeypatch): "INKBOX_ALLOWED_USERS", "CODEX_BIN", "CODEX_SANDBOX", "CODEX_APPROVAL_POLICY", "INKBOX_CODEX_AUTO_APPROVE_INKBOX_TOOLS", "INKBOX_BASE_URL", "CODEX_TURN_TIMEOUT_S", "CODEX_INTERRUPT_TIMEOUT_S", + "CODEX_APP_SERVER_STREAM_LIMIT_BYTES", "INKBOX_CONTACT_MEMORIES_ENABLED", ): monkeypatch.delenv(var, raising=False) @@ -19,6 +24,10 @@ def test_read_config_defaults(monkeypatch): assert cfg.auto_approve_inkbox_tools is False assert cfg.codex_turn_timeout_s == 1800.0 assert cfg.codex_interrupt_timeout_s == 10.0 + assert ( + cfg.codex_app_server_stream_limit_bytes + == DEFAULT_CODEX_APP_SERVER_STREAM_LIMIT_BYTES + ) assert cfg.contact_memories_enabled is True @@ -33,6 +42,7 @@ def test_read_config_env(monkeypatch): monkeypatch.setenv("INKBOX_CODEX_AUTO_APPROVE_INKBOX_TOOLS", "true") monkeypatch.setenv("CODEX_TURN_TIMEOUT_S", "42") monkeypatch.setenv("CODEX_INTERRUPT_TIMEOUT_S", "3") + monkeypatch.setenv("CODEX_APP_SERVER_STREAM_LIMIT_BYTES", "8388608") cfg = read_config() assert cfg.api_key == "ApiKey_test" assert cfg.base_url == "https://proxy.example" @@ -43,6 +53,7 @@ def test_read_config_env(monkeypatch): assert cfg.auto_approve_inkbox_tools is True assert cfg.codex_turn_timeout_s == 42.0 assert cfg.codex_interrupt_timeout_s == 3.0 + assert cfg.codex_app_server_stream_limit_bytes == 8388608 def test_contact_memories_can_be_disabled(monkeypatch): diff --git a/tests/test_sessions.py b/tests/test_sessions.py index 1080133..c5ab1da 100644 --- a/tests/test_sessions.py +++ b/tests/test_sessions.py @@ -500,6 +500,47 @@ async def disconnect(self): asyncio.run(scenario()) +def test_ensure_client_restarts_unhealthy_app_server(monkeypatch): + async def scenario(): + session = make_session([]) + session.resume_session_id = "saved-thread" + session.on_session_id = lambda _chat_id, _thread_id: None + + class UnhealthyClient: + is_healthy = False + + def __init__(self): + self.disconnects = 0 + + async def disconnect(self): + self.disconnects += 1 + + class ReplacementClient: + is_healthy = True + + def __init__(self, *_args, **_kwargs): + self.thread_id = None + self.resumed_from = None + + async def connect(self, resume_thread_id=None): + self.resumed_from = resume_thread_id + self.thread_id = "resumed-thread" + return self.thread_id + + old_client = UnhealthyClient() + session._client = old_client + monkeypatch.setattr(sessions_mod, "CodexAppServerClient", ReplacementClient) + + replacement = await session._ensure_client() + + assert old_client.disconnects == 1 + assert isinstance(replacement, ReplacementClient) + assert replacement.resumed_from == "saved-thread" + assert session.resume_session_id == "resumed-thread" + + asyncio.run(scenario()) + + def test_stop_command_interrupts_turn_without_clearing(): async def scenario(): sent = [] From 28717a7a1fdb73eaeda8abe6edffac865387aec5 Mon Sep 17 00:00:00 2001 From: dimavrem22 Date: Mon, 3 Aug 2026 18:20:51 +0000 Subject: [PATCH 2/3] Strengthen app-server reader failure regressions --- tests/test_codex_client_stream.py | 52 ++++++++++++++++++++++++++++--- 1 file changed, 47 insertions(+), 5 deletions(-) diff --git a/tests/test_codex_client_stream.py b/tests/test_codex_client_stream.py index 406ed53..2496f7b 100644 --- a/tests/test_codex_client_stream.py +++ b/tests/test_codex_client_stream.py @@ -1,4 +1,5 @@ import asyncio +import logging import os from pathlib import Path from types import SimpleNamespace @@ -14,8 +15,11 @@ import sys -def send(message): - sys.stdout.write(json.dumps(message, separators=(",", ":")) + "\n") +def send(message, require_oversized=False): + encoded = json.dumps(message, separators=(",", ":")) + "\n" + if require_oversized: + assert len(encoded.encode()) > 65536 + sys.stdout.write(encoded) sys.stdout.flush() @@ -37,7 +41,7 @@ def send(message): "turnId": "turn-1", "item": {"type": "agentMessage", "text": reply}, }, - }) + }, require_oversized=True) send({ "method": "turn/completed", "params": { @@ -48,7 +52,12 @@ def send(message): "items": [{"type": "commandExecution", "output": "y" * 70000}], }, }, - }) + }, require_oversized=True) + elif method == "probe/fail": + send({ + "method": "probe/oversized", + "params": {"padding": "z" * 70000}, + }, require_oversized=True) ''' @@ -84,7 +93,7 @@ async def scenario(): asyncio.run(scenario()) -def test_reader_limit_failure_fails_turn_and_marks_client_unhealthy(tmp_path): +def test_reader_limit_failure_fails_turn_and_terminates_child(tmp_path): async def scenario(): client = _client(_mock_app_server(tmp_path), stream_limit=1024) try: @@ -92,6 +101,39 @@ async def scenario(): with pytest.raises(CodexAppServerError, match="reader"): await asyncio.wait_for(client.run_detailed("test"), timeout=2) assert client.is_healthy is False + assert client._proc is not None + await asyncio.wait_for(client._proc.wait(), timeout=1) + assert client._proc.returncode is not None + finally: + await client.disconnect() + + asyncio.run(scenario()) + + +def test_reader_failure_logs_and_fails_all_pending_requests(tmp_path, caplog): + async def scenario(): + client = _client(_mock_app_server(tmp_path), stream_limit=1024) + caplog.set_level(logging.ERROR, logger="inkbox_codex.codex_client") + try: + await client.connect() + first = asyncio.create_task(client._request("probe/wait", {})) + second = asyncio.create_task(client._request("probe/wait", {})) + await asyncio.sleep(0) + assert len(client._pending) == 2 + trigger = asyncio.create_task(client._request("probe/fail", {})) + + results = await asyncio.wait_for( + asyncio.gather(first, second, trigger, return_exceptions=True), + timeout=2, + ) + + assert all(isinstance(result, CodexAppServerError) for result in results) + assert all("reader" in str(result) for result in results) + assert client._pending == {} + assert client.is_healthy is False + assert "Codex app-server reader stopped unexpectedly" in caplog.text + assert client._proc is not None + await asyncio.wait_for(client._proc.wait(), timeout=1) finally: await client.disconnect() From 71c5554769ffc7f058c3438da861820633e6492f Mon Sep 17 00:00:00 2001 From: dimavrem22 Date: Mon, 3 Aug 2026 18:25:37 +0000 Subject: [PATCH 3/3] Align hosted voice gate with final delivery proof --- tests/live/test_voice.py | 29 ++++++++++------------- tests/test_live_voice_contract_helpers.py | 23 +++++++----------- 2 files changed, 20 insertions(+), 32 deletions(-) diff --git a/tests/live/test_voice.py b/tests/live/test_voice.py index 0a24010..e78ffa0 100644 --- a/tests/live/test_voice.py +++ b/tests/live/test_voice.py @@ -179,60 +179,55 @@ def _has_sms_action_intent(value: str | None) -> bool: return send and sms -def _matching_post_call_action(call, marker): - """Return the open current-marker SMS action persisted for a hosted call.""" - marker_key = _voice_marker_key(marker) +def _open_post_call_sms_action(call): + """Return the open SMS action persisted for this hosted call.""" for item in getattr(call, "post_call_action_items", None) or []: if isinstance(item, dict): status = item.get("status", "") action = item.get("action", "") + description = item.get("description", "") details = item.get("details", "") else: status = getattr(item, "status", "") action = getattr(item, "action", "") + description = getattr(item, "description", "") details = getattr(item, "details", "") - value = f"{action} {details}" + value = f"{action} {description} {details}" if ( str(status).casefold() == "open" - and marker_key in _voice_marker_key(value) and _has_sms_action_intent(value) ): return item return None -def _post_call_action_diagnostic(call, marker) -> dict[str, int | bool]: +def _post_call_action_diagnostic(call) -> dict[str, int | bool]: """Bounded, content-redacted predicates for the hosted action gate.""" items = getattr(call, "post_call_action_items", None) or [] open_count = 0 - marker_count = 0 sms_count = 0 matching_action = False - marker_key = _voice_marker_key(marker) for item in items[:10]: if isinstance(item, dict): status = item.get("status", "") action = item.get("action", "") + description = item.get("description", "") details = item.get("details", "") else: status = getattr(item, "status", "") action = getattr(item, "action", "") + description = getattr(item, "description", "") details = getattr(item, "details", "") - value = f"{action} {details}" + value = f"{action} {description} {details}" is_open = str(status).casefold() == "open" - has_marker = bool(marker_key) and marker_key in _voice_marker_key(value) has_sms_intent = _has_sms_action_intent(value) open_count += int(is_open) - marker_count += int(has_marker) sms_count += int(has_sms_intent) - matching_action = matching_action or ( - is_open and has_marker and has_sms_intent - ) + matching_action = matching_action or (is_open and has_sms_intent) return { "item_count": len(items), "inspected_count": min(len(items), 10), "open_count": open_count, - "marker_count": marker_count, "sms_count": sms_count, "matching_action": matching_action, } @@ -374,8 +369,8 @@ def _wait_for_persisted_hosted_request( transcript_diagnostic = {"error_type": type(exc).__name__} try: aut_call = aut.calls.get(aut_call_id) - action_ready = _matching_post_call_action(aut_call, marker) is not None - action_diagnostic = _post_call_action_diagnostic(aut_call, marker) + action_ready = _open_post_call_sms_action(aut_call) is not None + action_diagnostic = _post_call_action_diagnostic(aut_call) except Exception as exc: action_diagnostic = {"error_type": type(exc).__name__} if transcript_ready and action_ready: diff --git a/tests/test_live_voice_contract_helpers.py b/tests/test_live_voice_contract_helpers.py index 0aa3b0f..b2a9597 100644 --- a/tests/test_live_voice_contract_helpers.py +++ b/tests/test_live_voice_contract_helpers.py @@ -113,24 +113,21 @@ def test_call_pair_duplicate_diagnostic_names_owner_and_ids(): ) -def test_matching_post_call_action_requires_open_current_marker_sms(): - marker = "victor echo juliet" +def test_hosted_action_gate_requires_open_sms_action(): matching = { "status": "open", - "action": "send_sms", - "details": "After the call, send Victor-Echo, Juliet to the caller.", + "description": "Send the requested SMS to the caller after the call.", } - assert voice._matching_post_call_action( - SimpleNamespace(post_call_action_items=[matching]), marker + assert voice._open_post_call_sms_action( + SimpleNamespace(post_call_action_items=[matching]) ) is matching for item in ( {**matching, "status": "canceled"}, - {**matching, "details": "Send a different marker."}, - {**matching, "action": "create_note", "details": marker}, + {**matching, "description": "Record the request in a note."}, ): - assert voice._matching_post_call_action( - SimpleNamespace(post_call_action_items=[item]), marker + assert voice._open_post_call_sms_action( + SimpleNamespace(post_call_action_items=[item]) ) is None @@ -150,16 +147,12 @@ def test_action_gate_diagnostic_is_bounded_and_content_redacted(): ] ) - diagnostic = voice._post_call_action_diagnostic( - call, - "Victor Echo Juliet", - ) + diagnostic = voice._post_call_action_diagnostic(call) assert diagnostic == { "item_count": 16, "inspected_count": 10, "open_count": 1, - "marker_count": 1, "sms_count": 1, "matching_action": True, }