Fix app-server reader recovery for large messages

.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

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. |

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"

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:

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},
                 },

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(

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",
                     },
                 },
             )

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", ""),

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 = [

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", {})