fix(codex): serialize stdin writes to prevent parallel tool-call interleaving (#5229)

When Codex emits parallel tool calls, the harness processes them
sequentially but _send_message's write+drain sequence is not atomic:
a concurrent caller can write() between another caller's write() and
drain(), interleaving bytes on the subprocess stdin pipe.  The Codex
app-server then reads a corrupted JSON-RPC line, drops the response,
and the remaining tool outputs never arrive — causing the turn to stall
for minutes before timing out.

Fix: guard _send_message with an asyncio.Lock (_stdin_lock) so that
the write/drain pair is always atomic.  The lock is initialized in
__init__ alongside the other per-session state.

Signed-off-by: Tomu Hirata <tomu.hirata@gmail.com>

omnigent/inner/codex_executor.py

@@ -2123,6 +2123,9 @@ class _CodexAppServerSession:
         # on the next ``turn/completed`` so each TurnComplete carries the
         # usage for the turn that just finished.
         self._last_turn_usage: dict[str, object] | None = None
+        # Serialize concurrent writes to the subprocess stdin so that parallel
+        # tool-call responses don't interleave bytes on the pipe.
+        self._stdin_lock = asyncio.Lock()
 
     async def start(self) -> None:
         if self._started:
@@ -2947,8 +2950,9 @@ class _CodexAppServerSession:
 
     async def _send_message(self, payload: CodexMessage) -> None:
         assert self._proc is not None and self._proc.stdin is not None
-        self._proc.stdin.write((json.dumps(payload) + "\n").encode("utf-8"))
-        await self._proc.stdin.drain()
+        async with self._stdin_lock:
+            self._proc.stdin.write((json.dumps(payload) + "\n").encode("utf-8"))
+            await self._proc.stdin.drain()
 
     @staticmethod
     async def _iter_stream_chunks(stream: asyncio.StreamReader) -> AsyncIterator[bytes]: