Add streamable HTTP transport#41
Conversation
|
@codex review |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 6917c3d4c1
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| case "session/cancel", | ||
| "session/close", | ||
| "session/load", | ||
| "session/prompt", | ||
| "session/resume", | ||
| "session/set_config_option", | ||
| "session/set_mode", | ||
| "session/set_model": |
There was a problem hiding this comment.
When clients use the unstable session/fork method, its params include a required sessionId (UnstableForkSessionRequest.SessionId), so it needs the same Acp-Session-Id enforcement/routing as the other session-scoped POSTs. Leaving it out here means the HTTP server accepts a fork POST without the session header and records the response on the connection-scoped stream, which breaks streamable-HTTP clients that rely on the header contract for session-scoped methods and can deliver the fork response on the wrong stream.
Useful? React with 👍 / 👎.
![coder-agents-review[bot]](https://avatars.githubusercontent.com/in/3454270?s=60&v=4){kind=small}
There was a problem hiding this comment.
Solid transport implementation. The pipe-bridge architecture is pragmatic given the SDK's existing io.Reader/io.Writer contract, the routing logic (connection vs session streams, session/load special-casing, pre-subscribe buffering) is well thought through, and the E2E test wiring both sides through real SDK connections is a strong trust signal. The RFD adherence is thorough.
Severity count: 1 P1, 8 P2, 13 P3, 5 Nit.
The P1 is convergent across 10 reviewers: outboundStream.push silently drops messages to slow subscribers, which can permanently hang the client SDK with zero diagnostic. The structural fix (close the subscriber on channel overflow so the client reconnects instead of hanging forever) is surgical and several reviewers independently converged on it.
The P2s cluster around three themes: (1) routing correctness (session header used without checking IsSessionScoped, missing session/fork in the method list), (2) resource lifecycle (zombie connections after agent death, unbounded connection creation, session stream accumulation), and (3) test gaps on error paths and the replay buffer.
"The stdio path applies backpressure through the pipe: a slow reader slows the writer. The HTTP transport's fire-and-forget push removes that backpressure and introduces silent data loss. The trade-off is reasonable for SSE, but the implicit reliability downgrade relative to stdio is undocumented." (Pariston)
🤖 This review was automatically generated with Coder Agents.
| return | ||
| } | ||
| for _, sub := range s.subscribers { | ||
| select { |
There was a problem hiding this comment.
P1 [DEREM-4] Silent message drop to slow subscribers can permanently hang the client SDK.
When a subscriber's 128-entry channel is full, push does a non-blocking send and silently discards the message. If the SSE handler is blocked on Fprintf/Flush (TCP backpressure, slow client, reverse proxy buffering), the channel fills. Once it overflows, JSON-RPC responses to client requests are discarded. The client SDK blocks indefinitely waiting for a response that was dropped. No log, no subscriber teardown, no way for either side to detect the gap. The response is already consumed from pending by takePendingRoute in the router, so there is no replay path.
The drop-over-block tradeoff protects the router goroutine from stalling all streams on one slow client. That's the right instinct, but the failure mode needs to be visible. (Ryosuke)
The structural fix: on overflow, call sub.closeDone() to tear down the stream. The SSE handler already checks sub.done in its select and will exit cleanly. This converts a silent protocol hang into a visible disconnect that the client's reconnect logic already handles. Data is still lost, but a silent hang becomes a detectable reconnect. (Ryosuke P1, Knov P2, Hisoka P2, Killua P2, Chopper P2, Meruem P2, Knuckle P2, Mafuuu P2)
🤖
| // Observe the agent-side connection's lifecycle so if the agent | ||
| // goroutines die (peer closed, unrecoverable error), we tear the | ||
| // connection down. | ||
| go func() { |
There was a problem hiding this comment.
P2 [DEREM-5] Watcher goroutine leaks zombie connections in the server map.
The watcher calls c.shutdown() when the agent-side connection dies, but never calls s.removeConn(c.id). The connection struct has no back-reference to the server. Every client that disconnects without sending DELETE (crash, timeout, network drop) leaves a zombie entry in s.connections. The map grows without bound until the server is restarted or Close() is called.
Fix: capture s in the closure and call s.removeConn(c.id) before c.shutdown(). (Kite)
🤖
| // stream alongside session/new: the client hasn't necessarily opened | ||
| // the session-scoped GET when it issues session/load, so the | ||
| // connection stream is the only place it is guaranteed to land. | ||
| if sessionHeader != "" && envelope.Method != "session/load" { |
There was a problem hiding this comment.
P2 [DEREM-6] Response routing uses raw Acp-Session-Id header regardless of whether the method is session-scoped.
The condition sessionHeader != "" && envelope.Method != "session/load" consults the raw header, not IsSessionScoped. A malformed (or adversarial) session/new POST that includes Acp-Session-Id: some-id passes the "missing header" guard (IsSessionScoped returns false for session/new), but lands in the session-route branch here because sessionHeader is non-empty. The response gets pushed to a session stream the client isn't listening on. Client hangs.
The well-behaved client in this PR never triggers this, but this is a public transport API.
Fix: gate session routing on IsSessionScoped:
if acphttp.IsSessionScoped(envelope.Method) && envelope.Method != "session/load" && sessionHeader != "" {(Knov)
🤖
| func IsSessionScoped(method string) bool { | ||
| switch method { | ||
| case "session/cancel", | ||
| "session/close", | ||
| "session/load", | ||
| "session/prompt", | ||
| "session/resume", | ||
| "session/set_config_option", | ||
| "session/set_mode", | ||
| "session/set_model": |
There was a problem hiding this comment.
P2 [DEREM-7] IsSessionScoped omits session/fork (and other unstable methods with required sessionId).
The function's doc says it lists "every agent-side request type whose params include a required 'sessionId' field." The comment at lines 119-122 justifies including session/set_model (also unstable) so "transports speaking the unstable protocol get correct header behaviour out of the box." session/fork (UnstableForkSessionRequest in types_gen.go) has the same shape: unstable, required sessionId.
Consequence: the server accepts a session/fork POST without Acp-Session-Id, and the response routing records routeConnection instead of routeSession, delivering the fork response on the wrong stream.
Melody also identified siblings: nes/close, nes/suggest, nes/accept, nes/reject, and five document/did* methods all carry sessionId in params and are absent. (Melody P2, Luffy P3, Zoro P3)
🤖
| if s.closed { | ||
| return fmt.Errorf("httpserver: server closed") | ||
| } | ||
| s.connections[id] = c |
There was a problem hiding this comment.
P2 [DEREM-8] No limit on concurrent connections; every initialize POST creates an agent, goroutines, and pipes with no cap.
addConn checks only s.closed. An attacker (or a misfiring reconnect loop) sends initialize in a loop. Each call runs Factory(ctx), allocates two io.Pipe pairs, a router goroutine, a lifecycle-watcher goroutine, plus the connection's outbound streams. Goroutines, file descriptors, and memory scale linearly with request rate. createConnection calls the factory before addConn, so even a hypothetical check in addConn would still execute the factory.
Fix: add a MaxConnections field to Config, enforce in createConnection before calling Factory, return 503 from handleInitialize when the limit is hit. (Kurapika, Knov, Pen Botter)
🤖
|
|
||
| // --- JSON envelope helper used in tests --- | ||
|
|
||
| type jsonRPCMessage struct { |
There was a problem hiding this comment.
P3 [DEREM-2] jsonRPCMessage type is defined but unused. Silenced with var _ = jsonRPCMessage{} and annotated "kept for future use." Dead test infrastructure should not ship; add it when needed. (Netero)
🤖
| func (t *Transport) runStream(ctx context.Context, sessionID, label string) { | ||
| // Backoff for reconnect attempts when the network drops but the | ||
| // transport is still alive (e.g. server restart). | ||
| backoff := 250 * time.Millisecond |
There was a problem hiding this comment.
P3 [DEREM-3] Reconnect backoff never resets after a healthy stream connection. After several disconnect/reconnect cycles, backoff is permanently at 5s regardless of connection health. Reset backoff = 250 * time.Millisecond when a stream successfully opens and delivers at least one event. (Netero)
🤖
| closedCh chan struct{} | ||
| } | ||
|
|
||
| // Dial creates a Transport. It does not perform any network I/O; the |
There was a problem hiding this comment.
Nit [DEREM-33] Dial implies network I/O but the godoc says "it does not perform any network I/O." Standard Go naming: Dial means "connect now" (net.Dial, grpc.Dial, websocket.Dial). Consider NewTransport to match the type it returns and the zero-I/O contract. (Gon)
🤖
| sessionStreams sync.Map // map[string]*outboundStream | ||
|
|
||
| // pending records where server → client responses to client → server | ||
| // POSTs must go, keyed by the canonical JSON-RPC id. |
There was a problem hiding this comment.
Note [DEREM-35] pending map has no eviction for orphaned entries. If the agent drops a response (bug, crash mid-turn), the entry is never removed and accumulates for the connection's lifetime. Per-entry cost is low and the trigger requires an agent bug, so this is low priority. A periodic sweep or size cap would close the gap. (Knuckle)
🤖
| // Handler returns an http.Handler that serves the ACP endpoint. Mount it | ||
| // at the root of an http.Server (routing is done internally so callers can | ||
| // mix with unrelated routes if they use an outer mux). | ||
| func (s *Server) Handler() http.Handler { |
There was a problem hiding this comment.
Note [DEREM-36] Handler() users must call Close() manually to avoid goroutine leaks. Connection contexts are rooted in context.Background(), not any server-level context. ListenAndServe/Serve call Close() on context cancellation, but Handler() returns a raw http.Handler with no lifecycle hook. Worth noting in the doc comment. (Meruem)
🤖
|
@ThomasK33 Hey Thomas, I addressed most of the review comments. |
|
/coder-agents-review |
|
Chat: Review posted | View chat Review history
deep-review v0.5.0 | Round 2 | Last posted: Round 2, 35 findings (1 P1, 9 P2, 17 P3, 6 Nit, 2 Note), COMMENT. Review Finding inventoryFindings
Contested and acknowledged(none) Round logRound 1Panel (23 reviewers: Bisky, Chopper, Ging-Go, Gon, Hisoka, Killua, Kite, Knov, Kurapika, Leorio, Luffy, Mafu-san, Mafuuu, Melody, Meruem, Pariston, Razor, Robin, Ryosuke, Takumi, Zoro + wildcards Knuckle, Pen Botter). Netero 3 P3. Panel: 1 P1, 8 P2, 13 P3, 5 Nit, 2 Note. 1 dropped. Reviewed against 192e108..6917c3d. Round 2BLOCKED. 25 addressed, 10 silent. No review. Reviewed against 192e108..71420b2. About deep-reviewCRF = Coder Review Finding (P0-P4, Nit, Note)
|
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 71420b28f9
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| if err == nil || errors.Is(err, context.Canceled) || t.isClosed() { | ||
| return |
There was a problem hiding this comment.
When an SSE GET ends normally, parseSSE returns nil and this branch permanently exits the stream. That can happen in this implementation when the server closes a slow subscriber via sub.done on outbound overflow (acphttp/server/connection.go says this relies on client reconnects), or when an intermediary cleanly closes an idle stream; after that the connection-scoped stream is never reopened, so later session/new/session/load responses are only buffered server-side and SDK callers can wait forever. Treat EOF as reconnectable unless the transport is closing.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Good progress: 25 of 35 findings addressed in 71420b2. The P1 (silent subscriber drop), both routing issues (DEREM-6, DEREM-7), the zombie connection leak (DEREM-5), and many P3/Nit fixes all look right. Thank you for the thorough response.
Further review is blocked until the remaining 10 findings receive a response (fix, acknowledge, or contest). The panel has not yet verified the fixes; that happens in the next unblocked round.
Silent P2 findings (need response before panel review proceeds):
- DEREM-8 No limit on concurrent connections. Add
MaxConnectionsto Config, or explain why deployment-level limiting is sufficient. - DEREM-9 No test for client transport error paths (5 paths in doInitialize). Add error-path tests, or explain why the happy-path coverage is sufficient.
- DEREM-10 No test exercises the pre-subscribe replay buffer. Add a test that sends messages before a subscriber attaches, or explain.
- DEREM-12 Unbounded reconnect loop converts permanent server failure into silent client hang. Add a max retry or configurable timeout, or explain why infinite retry is correct.
Silent P3 findings:
- DEREM-15 Session-scoped streams never cleaned up until connection shutdown.
- DEREM-22 Clean server EOF permanently kills client SSE streams with no reconnect.
- DEREM-23 WaitGroup Add/Wait race in ensureSessionStream and openConnectionStream.
- DEREM-24
CanonicalIDnaming diverges from root SDK'scanonicalJSONRPCIDKey.
Silent Nit/Note findings:
- DEREM-33
Dialnaming implies network I/O. - DEREM-35
pendingmap has no eviction for orphaned entries.
For any finding you disagree with, a reply explaining why is sufficient to unblock. For findings you plan to defer, file a ticket and link it.
🤖 This review was automatically generated with Coder Agents.
2 participants
Implementation is based on:
The RFD is proposing streamable HTTP as a base network transport with an option to upgrade to WebSockets. I didn't implement WebSockets part yet, this will be a separate PR.
In our pool client we are trying it out and can successfully connect and talk to the Goose agent over HTTP: