Fix streaming reliability: BrokenPipeError, tool call visibility, keepalive

Three fixes for streaming reliability issues found in session
019eee1b-bbf5-7992-a133-f3a61c0ed3dc:

1. BrokenPipeError crash: When Codex aborts a turn, it closes the
   connection. The proxy kept trying to write SSE events and crashed.
   Now send_event() catches BrokenPipeError, sets client_alive=False,
   and silently skips further writes. Prevents proxy crashes and
   allows clean turn aborts.

2. Tool calls showing as "thinking": The reasoning item stayed
   status=in_progress the entire time tool calls were being made.
   The UI saw reasoning still active and showed "thinking" instead
   of the tool calls. Now the reasoning item is closed (status=
   completed) before emitting tool call output_item.added events.

3. SSE keepalive: When the upstream model thinks for 30+ seconds
   before responding, Codex timed out waiting for content. Now a
   background thread sends SSE comment lines (": keepalive") every
   15s while waiting for the upstream first byte, keeping the
   connection alive.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

Author: kartikkabadi

src/opencode_go_proxy/app.py

@@ -196,14 +196,24 @@ def handle_streaming_request(payload: Json, config: ProxyConfig, request_id: str
     response_id = new_response_id()
     model = request_model or DEFAULT_MODEL
 
+    client_alive = True
+
     def send_event(event: Json) -> None:
-        wfile.write(b"data: " + json.dumps(event, separators=(",", ":")).encode("utf-8") + b"\n\n")
-        wfile.flush()
+        nonlocal client_alive
+        if not client_alive:
+            return
+        try:
+            wfile.write(b"data: " + json.dumps(event, separators=(",", ":")).encode("utf-8") + b"\n\n")
+            wfile.flush()
+        except BrokenPipeError:
+            client_alive = False
+            trace("client.disconnected", request_id=request_id, message="client closed connection during stream")
 
     def send_error(msg: str) -> None:
         send_event({"type": "response.error", "error": {"message": msg}})
-        wfile.write(b"data: [DONE]\n\n")
-        wfile.flush()
+        if client_alive:
+            wfile.write(b"data: [DONE]\n\n")
+            wfile.flush()
 
     try:
         api_key = resolve_api_key(config, request_id)
@@ -238,8 +248,26 @@ def send_error(msg: str) -> None:
     reasoning_open = False
     got_data = False
 
+    # Keepalive: send SSE comments every 15s while waiting for upstream first byte.
+    # Prevents Codex from timing out when the model thinks for 30+ seconds before responding.
+    keepalive_stop = threading.Event()
+
+    def keepalive() -> None:
+        while not keepalive_stop.wait(15):
+            if not client_alive:
+                return
+            try:
+                wfile.write(b": keepalive\n\n")
+                wfile.flush()
+            except BrokenPipeError:
+                return
+
+    ka_thread = threading.Thread(target=keepalive, daemon=True)
+    ka_thread.start()
+
     try:
         with urllib.request.urlopen(req, timeout=config.timeout_sec) as resp:
+            keepalive_stop.set()  # Stop keepalive once upstream starts responding.
             for line in resp:
                 line = line.decode("utf-8", errors="replace").strip()
                 if not line.startswith("data: "):
@@ -282,6 +310,12 @@ def send_error(msg: str) -> None:
                     text += d
                     send_event({"type": "response.output_text.delta", "item_id": message_id, "output_index": 1 if reasoning_open else 0, "delta": d})
                 tcs = delta.get("tool_calls")
+                if isinstance(tcs, list) and tcs and reasoning_open:
+                    # Close reasoning item before tool calls so UI shows tool calls, not "thinking".
+                    rs_done = {"type": "reasoning", "id": reasoning_id,
+                               "summary": [{"type": "summary_text", "text": reasoning}], "status": "completed"}
+                    send_event({"type": "response.output_item.done", "output_index": 0, "item": rs_done})
+                    reasoning_open = False
                 if isinstance(tcs, list):
                     for tc in tcs:
                         idx = tc.get("index", 0)
@@ -317,6 +351,7 @@ def send_error(msg: str) -> None:
                         if fn.get("arguments"):
                             tool_calls[idx]["function"]["arguments"] += fn["arguments"]
     except urllib.error.HTTPError as exc:
+        keepalive_stop.set()
         body = exc.read().decode("utf-8", errors="replace")
         trace("upstream.error", request_id=request_id, status=exc.code, body=body[:2000])
         if exc.code == 429:
@@ -328,6 +363,7 @@ def send_error(msg: str) -> None:
             send_error(f"upstream HTTP {exc.code}")
         return
     except (urllib.error.URLError, TimeoutError) as exc:
+        keepalive_stop.set()
         trace("upstream.network_error", request_id=request_id, reason=str(getattr(exc, "reason", exc)))
         send_error(f"upstream network error: {getattr(exc, 'reason', exc)}")
         return
@@ -339,6 +375,10 @@ def send_error(msg: str) -> None:
         send_error("upstream returned no SSE data")
         return
 
+    if not client_alive:
+        trace("client.gone", request_id=request_id, message="client disconnected before final events")
+        return
+
     # Build final response from accumulated data.
     fake_msg: Json = {}
     if reasoning: