Add Codex WebSocket support - #521
Conversation
thinkter
changed the title
thinkter
changed the title
|
Hey! I was working on this too. I didn't notice your PR before pushing, but if you want to take anything from my branch (I think there are a couple bugs that I'm happy to point out, but if you ask fable/sol to diff your work against mine they should find it) |
05bf5a4 to
18229c8
Compare
|
hey @mjlbach! I already had my agent look into your implementation. I want to harden this implementation as much as I can |
mjlbach
left a comment
mjlbach
left a comment
There was a problem hiding this comment.
Findings from diffing this against #523, inline below. One note with no code anchor: the listed verification is loopback-only — previous_response_id over a reused socket, pings mid-stream, and the close handshake are the parts a fixture can't validate, so worth a live run against chatgpt.com before marking ready.
| // deliberately outside the global pool mutex. | ||
| if (displaced) |connection| websocket_transport.close(connection, pool_alloc); | ||
| if (reusable) |connection| { | ||
| websocket_transport.ping(connection, args.cancel_flag, args.deadline, args.delivery) catch |err| { |
There was a problem hiding this comment.
Retained slots have an age limit but no idle TTL. After the session idles past a NAT/proxy timeout the socket is silently dead: this ping's write succeeds, the pong read blocks, and the turn stalls until the event-idle watcher fires (default 30s) before reconnecting. An idle TTL (~5 min, matching codex-rs and pi) avoids that, and with it this per-reuse ping — a blocking RTT on every reused turn — could probably go too, or at least get its own short deadline (~2s) instead of the 30s idle timeout.
| slots = .empty; | ||
| pool_mutex.unlock(io_mod.getIo()); | ||
| var owned = retired; | ||
| for (owned.items) |*slot| slot.deinit(); |
There was a problem hiding this comment.
shutdown() deinits busy slots too, closing and freeing a Connection a streaming thread may still hold. Related: the two relock sites in acquire (the ping-failure path and the post-connect commit) index slots.items[slot_index] without the length check that release/continuation/rollbackReservation do — after this swap that's an out-of-bounds index. Skipping busy slots here (release already closes the connection when index >= slots.items.len) and adding the bounds checks covers both.
| prior_health = slot.health_failures; | ||
| const expired = age_limit != 0 and | ||
| io_mod.milliTimestamp() - slot.opened_at_ms > age_limit; | ||
| if (slot.connection != null and slot.health_failures < health_budget and !expired) { |
There was a problem hiding this comment.
When can health_failures < health_budget be false here? Both increment paths (release(.failed), the ping failure below) also null slot.connection, and release(.completed) resets to 0 — so an idle slot with a live connection always has 0 failures. If it's future-proofing, fine; if it's meant to poison a flapping lane it doesn't currently do that.
| return error.InvalidOpenAICodexTransport; | ||
| } | ||
|
|
||
| fn allowsSseFallback(err: anyerror, delivery: gateway_client.DeliveryCertainty.State) bool { |
There was a problem hiding this comment.
acquire's entry check returns error.Timeout when the request deadline already expired, with delivery still definitely_unsent — so a turn that just ran out of time latches the whole process to SSE. Same when a short request deadline bounds the connect instead of the transport's own timeout. Worth latching only when the transport's own connect timeout fired.
| ) !void { | ||
| try writeFrame(writer, .close, &.{ 0x03, 0xe8 }); | ||
| try connection.flush(); | ||
| const frame = readFrame(alloc, reader) catch |err| { |
There was a problem hiding this comment.
This reads exactly one frame, so a server ping or trailing data frame between our close and its close reply returns WebSocketProtocolViolation. close() swallows it, but the one-shot stream() propagates it — failing a turn whose response was already fully consumed. Looping until the close frame arrives (answering pings, discarding data) matches RFC 6455 here.
| @@ -0,0 +1,128 @@ | |||
| # Codex WebSocket transport: plan to finish PR #521 | |||
There was a problem hiding this comment.
Seems like AI coding agent leakage :) You should remove this from the PR
2 participants
Summary
openai_codex_websocket.zig, leaving shared Responses serialization in the base providerprevious_response_idonly when serialized history proves an exact safe extensionprevious_response_not_foundresponses_websockets=2026-02-06beta protocol headerFX_CODEX_WEBSOCKET_MAX_SLOTS, in addition to the per-identityFX_CODEX_WEBSOCKET_MAX_LANESlimitFX_CODEX_TRANSPORTvalues as configuration errors rather than silently coercing them to SSELocal verification
zig fmt --check src/zig buildFull CI is required on exact commit
18229c8f720980e836b8a873300c1d7bae733756before readiness. GitHub currently marks the fork-origin workflow runs asaction_required; avercel-labs/fxmaintainer must approve them.