feat: stream AI response tokens via SSE#248
Conversation
This prevents holding up the HTTP handler goroutine for the duration of inference. Co-authored-by: codex <codex@users.noreply.github.com>
|
@pisum-sativum is attempting to deploy a commit to the ankanmisra's projects Team on Vercel. A member of the Team first needs to authorize it. |
|
Hi @pisum-sativum, thanks for opening this PR. Every contribution helps MicroAI-Paygate grow. If you find the project useful, consider starring the repository — it helps others discover it. Star MicroAI-Paygate on GitHub Looking forward to reviewing this PR. |
github-actions
Bot
added
go
📝 WalkthroughWalkthroughThis PR changes AI summarization from a single JSON response to SSE streaming end to end. Providers now expose streaming APIs, the gateway streams text and receipts, cache handling uses streamed summaries, and the web client/tests parse SSE payloads. ChangesStreaming summarization flow
Estimated code review effort: 4 (Complex) | ~60 minutes Sequence Diagram(s)sequenceDiagram
participant Client
participant Gateway
participant AIProvider
participant Redis
Client->>Gateway: POST /summarize
Gateway->>AIProvider: GenerateStream(ctx, prompt)
loop chunks
AIProvider-->>Gateway: stream.Recv() text
Gateway-->>Client: data: {"text": ...}
end
Gateway->>Gateway: accumulate fullSummary
Gateway->>Gateway: generate receipt
Gateway-->>Client: data: {"receipt": ...}
Gateway-->>Client: data: [DONE]
Gateway->>Redis: storeInCache(full_summary)
Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 inconclusive)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Warning There were issues while running some tools. Please review the errors and either fix the tool's configuration or disable the tool if it's a critical failure. 🔧 golangci-lint (2.12.2)level=error msg="[linters_context] typechecking error: pattern ./...: directory prefix . does not contain main module or its selected dependencies" Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
|
@AnkanMisra I have done the necessary changes pls check and merge with gssoc and gssoc:approved tags. |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 7
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
gateway/cache.go (1)
204-208: 🎯 Functional Correctness | 🔴 Critical | ⚡ Quick winRemove the unused
bodyBytesvariable ingateway/cache.go:207. It causes a Go compile error (declared and not used: bodyBytes) and blocks the build.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@gateway/cache.go` around lines 204 - 208, The handler in cache-related code reads writer.body into bodyBytes but never uses it, causing a compile failure. Remove the unused bodyBytes assignment from the status/body extraction block in the response-wrapping logic around writer.ResponseWriter.Status() so only the needed statusCode is kept, or use the value if it is intended for later caching logic.
🧹 Nitpick comments (3)
gateway/timeout_test.go (1)
66-66: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winConsider adding a mid-stream cancellation regression test.
This test only verifies that
GenerateStreamfails fast when the upstream never responds before the context deadline. The PR's acceptance criteria also require that a client disconnect (context cancellation) mid-stream stops the upstream LLM call once tokens have already started flowing — that path (stream.Recv()returning an error after some chunks) isn't covered here or elsewhere in this diff.
As per coding guidelines, "Gateway handlers, config parsing, middleware, timeout behavior, receipts, cache/rate limits, and error paths must have table tests or targeted regression tests."🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@gateway/timeout_test.go` at line 66, Add a targeted regression test around GenerateStream in gateway/timeout_test.go for mid-stream client cancellation. The current test only covers a pre-response timeout; create a case where the upstream starts streaming, then the context is canceled after a few chunks and stream.Recv() returns an error. Verify the provider stops the upstream LLM call promptly and that the cancellation path is handled by the GenerateStream flow rather than only the initial deadline path.Source: Coding guidelines
tests/e2e.test.ts (1)
89-91: 🗄️ Data Integrity & Integration | 🔵 Trivial | ⚡ Quick winStrengthen SSE assertions to cover the receipt contract.
The test verifies the stream contains a
data:prefix and a[DONE]terminator, but doesn't assert the receipt payload is actually delivered, nor thatContent-Typereflects the new streaming response. Since receipt delivery moved from theX-402-Receiptheader to the final SSE chunk, this is the change most worth locking down in a regression test.✅ Proposed stronger assertions
+ expect(res.headers.get("content-type")).toContain("text/event-stream"); const textStr = await res.text(); expect(textStr).toContain("data: {"); + expect(textStr).toContain('"receipt"'); expect(textStr).toContain("[DONE]");As per coding guidelines, "Changes to gateway routes, headers, status codes, receipts, or response bodies require matching tests and updates to
gateway/openapi.yaml, README sections, and the web client if applicable."🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/e2e.test.ts` around lines 89 - 91, Strengthen the SSE contract test in tests/e2e.test.ts by asserting the streaming response is actually delivering the receipt payload, not just a generic data prefix and [DONE] terminator. Update the existing receipt-related assertions around res.text() to verify the final SSE chunk contains the expected receipt content, and add a Content-Type check for the streaming response. Use the test case in the e2e flow that exercises the receipt delivery path so the regression is covered where the response moved from the X-402-Receipt header to the final SSE event.Source: Coding guidelines
web/src/hooks/use-x402.ts (1)
112-117: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winConsider throttling per-chunk state updates.
Calling
update({ summary: text })on every SSE chunk causes asetStateper token, which can produce very frequent re-renders for long/fast streams. Consider batching updates (e.g., viarequestAnimationFrameor a small time-based throttle) so the UI still feels progressive without re-rendering on every token.Separately, if the summary display region uses
aria-live, verify that rapid text mutations don't spam screen readers with announcements — this may need debouncing for a11y as well.As per coding guidelines, "Preserve accessibility for forms/buttons/status messages in the web client; prevent hydration or uncontrolled-input issues."
Also applies to: 284-288
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@web/src/hooks/use-x402.ts` around lines 112 - 117, The per-chunk summary updates in use-x402’s readSummarizeSuccess callback are triggering a state update on every SSE token, which can cause excessive re-renders and noisy aria-live announcements. Update the summary handling in use-x402 to batch or throttle the update({ summary: text }) calls (for example with requestAnimationFrame or a short debounce) so the UI remains progressive without firing setState on every chunk. Also check any related summary rendering path to ensure the live region does not announce every rapid mutation and instead uses a debounced or throttled update strategy.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@gateway/internal/ai/ollama.go`:
- Around line 86-95: The ollamaStream.Recv method currently treats any
non-response chunk as a generic missing-field failure; update it to explicitly
detect Ollama error chunks from /api/generate by checking for an "error" field
in the decoded chunk and returning that message as the error. Keep the existing
decode handling, but in ollamaStream.Recv prefer returning the streamed error
text before falling back to the invalid response error when neither "response"
nor "error" is present.
In `@gateway/internal/ai/openrouter.go`:
- Around line 110-146: The openRouterStream Recv() path is ignoring SSE error
payloads and only extracting delta.content, so mid-stream OpenRouter failures
can be treated like normal empty output. Update Recv() in openrouter.go to
detect chunks containing an error or a finish_reason of "error" before returning
content, and return a non-nil error instead of continuing. Keep the existing
chunk parsing flow around json.Unmarshal, choices, and delta, but add explicit
handling for the error case so callers can surface the stream failure.
In `@gateway/main.go`:
- Around line 546-561: The SSE response encoding is inconsistent in
gateway/main.go: text chunks use json.Marshal in the streaming loop, but the
error and receipt events in this function and streamResultAndReceipt are still
built with fmt.Fprintf and %q. Replace those hand-built payloads with JSON
marshaling for the error/receipt objects so all SSE messages use valid JSON
escaping and match the text-chunk encoding path.
- Around line 563-591: The receipt generation, storage, and final SSE emission
logic is duplicated between handleSummarize and streamResultAndReceipt, which
can easily drift over time. Extract the repeated GenerateReceipt,
storeReceiptWithContext, and final receipt/[DONE] write sequence into a shared
helper such as emitReceiptEvent that accepts the c writer/flusher plus
paymentCtx, recovered address, request path, requestBody, and responseBody.
Update both call sites to use the helper so any future fix to receipt formatting
or emission behavior is applied consistently in one place.
- Around line 664-672: Move the SSE header writes in the streaming handler so
they happen only after the http.Flusher assertion succeeds; currently the
Content-Type, Cache-Control, and Connection headers are set before the check,
causing respondError to return a JSON error with text/event-stream headers when
flushing is unsupported. Update the same streaming logic used in the handler
around the flusher check, keeping the header setup after the ok branch and
before normal streaming output.
- Around line 582-588: The receipt flow has moved from the X-402-Receipt header
to the SSE final event in the response path, but the contract still assumes the
header in gateway/openapi.yaml, README.md, and web/src/lib/x402-client.ts.
Either restore the header in the gateway handler alongside the SSE receipt for
backward compatibility, or update the openapi spec, docs, and x402 client
parsing to read the receipt from the SSE payload using the same
receipt/receiptJSON/base64 flow.
In `@web/src/lib/x402-client.ts`:
- Around line 82-107: The SSE parsing in x402Client’s read loop is dropping the
final buffered message and may miss the receipt when the stream ends without a
trailing newline. After the while loop, flush the decoder once without stream
mode and process any remaining buffer as a final line so the last data: payload
is handled; use the existing x402Client/reader.read logic and receipt parsing
path to locate the fix. Also avoid silently swallowing JSON parse failures in
the line parsing try/catch—log or warn so malformed SSE chunks are diagnosable.
---
Outside diff comments:
In `@gateway/cache.go`:
- Around line 204-208: The handler in cache-related code reads writer.body into
bodyBytes but never uses it, causing a compile failure. Remove the unused
bodyBytes assignment from the status/body extraction block in the
response-wrapping logic around writer.ResponseWriter.Status() so only the needed
statusCode is kept, or use the value if it is intended for later caching logic.
---
Nitpick comments:
In `@gateway/timeout_test.go`:
- Line 66: Add a targeted regression test around GenerateStream in
gateway/timeout_test.go for mid-stream client cancellation. The current test
only covers a pre-response timeout; create a case where the upstream starts
streaming, then the context is canceled after a few chunks and stream.Recv()
returns an error. Verify the provider stops the upstream LLM call promptly and
that the cancellation path is handled by the GenerateStream flow rather than
only the initial deadline path.
In `@tests/e2e.test.ts`:
- Around line 89-91: Strengthen the SSE contract test in tests/e2e.test.ts by
asserting the streaming response is actually delivering the receipt payload, not
just a generic data prefix and [DONE] terminator. Update the existing
receipt-related assertions around res.text() to verify the final SSE chunk
contains the expected receipt content, and add a Content-Type check for the
streaming response. Use the test case in the e2e flow that exercises the receipt
delivery path so the regression is covered where the response moved from the
X-402-Receipt header to the final SSE event.
In `@web/src/hooks/use-x402.ts`:
- Around line 112-117: The per-chunk summary updates in use-x402’s
readSummarizeSuccess callback are triggering a state update on every SSE token,
which can cause excessive re-renders and noisy aria-live announcements. Update
the summary handling in use-x402 to batch or throttle the update({ summary: text
}) calls (for example with requestAnimationFrame or a short debounce) so the UI
remains progressive without firing setState on every chunk. Also check any
related summary rendering path to ensure the live region does not announce every
rapid mutation and instead uses a debounced or throttled update strategy.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: cbf57f8a-f79d-4a52-80d1-a79cc5c3c9ff
📒 Files selected for processing (12)
gateway/cache.gogateway/cache_integration_test.gogateway/errors_test.gogateway/internal/ai/ollama.gogateway/internal/ai/openrouter.gogateway/internal/ai/provider.gogateway/main.gogateway/receipt_store_integration_test.gogateway/timeout_test.gotests/e2e.test.tsweb/src/hooks/use-x402.tsweb/src/lib/x402-client.ts
| func (s *ollamaStream) Recv() (string, error) { | ||
| var chunk map[string]interface{} | ||
| if err := s.decoder.Decode(&chunk); err != nil { | ||
| return "", err // will return io.EOF when stream ends | ||
| } | ||
| if response, ok := chunk["response"].(string); ok { | ||
| return response, nil | ||
| } | ||
| return "", fmt.Errorf("invalid response from Ollama: missing response field") | ||
| } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "=== relevant files ==="
git ls-files | rg 'gateway/internal/ai/ollama.go|go.mod|go.sum|README|openapi|ollama|ai'
echo
echo "=== ollama.go excerpt ==="
cat -n gateway/internal/ai/ollama.go | sed -n '1,220p'
echo
echo "=== dependency versions mentioning ollama ==="
rg -n '"ollama|github.com/ollama' go.mod go.sum gateway -g '!**/node_modules/**'
echo
echo "=== repo mentions of /api/generate or error field ==="
rg -n 'api/generate|"\berror\b"|missing response field|ollama' gateway README.md . -g '!**/node_modules/**'Repository: AnkanMisra/MicroAI-Paygate
Length of output: 6137
🌐 Web query:
Ollama /api/generate streaming error field contract error chunk response field
💡 Result:
When using the Ollama /api/generate endpoint with streaming enabled (the default), the API uses newline-delimited JSON (NDJSON) [1][2]. Each chunk in the stream is an independent JSON object sent on a new line [2][3]. If an error occurs mid-stream, Ollama sends an error object as a JSON chunk within the NDJSON stream [4][5]. This error chunk typically contains an "error" field describing the issue [5][6][7]. Because the HTTP connection is already open and partial data may have been sent, the API does not change the HTTP status code when this occurs [5]. To handle this, client implementations must parse each line of the stream independently [2][3]. If a chunk contains an "error" field, the client should treat it as an error and stop processing the stream [5][7]. Example of an error chunk in a stream: {"error":"an error was encountered while running the model"} [5] If you prefer to receive a single, standard JSON response instead of a stream—which simplifies error handling by allowing the API to return a traditional HTTP error status—you can disable streaming by setting "stream": false in your request body [1][8].
Citations:
- 1: https://docs.ollama.com/api/streaming
- 2: https://www.mintlify.com/ollama/ollama/api/streaming
- 3: Streaming multiple json objects at the same time ollama/ollama#1728
- 4: https://docs.ollama.com/api/errors
- 5: https://github.com/ollama/ollama/blob/main/docs/api/errors.mdx
- 6: https://github.com/ollama/ollama/blob/main/docs/openapi.yaml
- 7: https://deepwiki.com/ollama/ollama-python/5.3-error-handling
- 8: https://github.com/ollama/ollama/blob/main/docs%2Fapi.md
Surface Ollama stream errors
/api/generate streams can return {"error":"..."} chunks; return that message here instead of the generic “missing response field” error.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@gateway/internal/ai/ollama.go` around lines 86 - 95, The ollamaStream.Recv
method currently treats any non-response chunk as a generic missing-field
failure; update it to explicitly detect Ollama error chunks from /api/generate
by checking for an "error" field in the decoded chunk and returning that message
as the error. Keep the existing decode handling, but in ollamaStream.Recv prefer
returning the streamed error text before falling back to the invalid response
error when neither "response" nor "error" is present.
| func (s *openRouterStream) Recv() (string, error) { | ||
| for { | ||
| line, err := s.reader.ReadBytes('\n') | ||
| if err != nil { | ||
| return "", err | ||
| } | ||
| line = bytes.TrimSpace(line) | ||
| if len(line) == 0 { | ||
| continue | ||
| } | ||
| if bytes.HasPrefix(line, []byte("data: ")) { | ||
| data := bytes.TrimPrefix(line, []byte("data: ")) | ||
| if string(data) == "[DONE]" { | ||
| return "", io.EOF | ||
| } | ||
| var chunk map[string]interface{} | ||
| if err := json.Unmarshal(data, &chunk); err != nil { | ||
| continue // skip invalid json | ||
| } | ||
| choices, ok := chunk["choices"].([]interface{}) | ||
| if !ok || len(choices) == 0 { | ||
| continue | ||
| } | ||
| choice, ok := choices[0].(map[string]interface{}) | ||
| if !ok { | ||
| continue | ||
| } | ||
| delta, ok := choice["delta"].(map[string]interface{}) | ||
| if !ok { | ||
| continue | ||
| } | ||
| if content, ok := delta["content"].(string); ok { | ||
| return content, nil | ||
| } | ||
| } | ||
| } | ||
| } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== File listing ==\n'
git ls-files 'gateway/internal/ai/openrouter.go' 'gateway/internal/ai/*' | sed -n '1,120p'
printf '\n== Outline ==\n'
ast-grep outline gateway/internal/ai/openrouter.go --view expanded || true
printf '\n== Relevant lines from gateway/internal/ai/openrouter.go ==\n'
cat -n gateway/internal/ai/openrouter.go | sed -n '1,240p'Repository: AnkanMisra/MicroAI-Paygate
Length of output: 7140
🌐 Web query:
OpenRouter streaming API SSE error payload format data: {"error":...} finish_reason error model version streaming chat completions
💡 Result:
OpenRouter handles streaming errors differently depending on whether tokens have already been sent to the client [1][2]. If an error occurs before any tokens are streamed, the API returns a standard JSON error response with the appropriate HTTP status code [1][2]. If an error occurs mid-stream (after tokens have been sent), the HTTP status code remains 200 OK because headers were already transmitted [1][2]. In this case, the error is delivered as a Server-Sent Event (SSE) containing a unified JSON payload [1][2]. The structure of this mid-stream error SSE payload is as follows [2]: data: {"id":"cmpl-abc123","object":"chat.completion.chunk","created":1234567890,"model":"model-name","provider":"provider-name","error":{"code":"error_code","message":"error message"},"choices":[{"index":0,"delta":{"content":""},"finish_reason":"error"}]} Key characteristics of this payload include: - The error object is included at the top level of the JSON alongside standard fields like id, object, and model [1][2]. - A choices array is provided with finish_reason set to "error" to formally terminate the stream [1][2]. - The stream is terminated by the server immediately after this event is sent [3][2]. OpenRouter normalizes finish_reason values across different models, and "error" is one of the standardized values used to indicate an abnormal stream termination [4].
Citations:
- 1: https://openrouter.ai/docs/api/reference/streaming.md
- 2: https://openrouter.ai/docs/api/reference/errors-and-debugging.mdx
- 3: https://openrouter.ai/docs/api/reference/errors-and-debugging
- 4: https://openrouter.ai/docs/api/reference/overview
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== Search for OpenRouter stream handling in repo ==\n'
rg -n "finish_reason|\\[DONE\\]|data: \\{\"error\"|openrouter" gateway -g '*.go' -g '*.md' -g '*.yaml' | sed -n '1,200p'Repository: AnkanMisra/MicroAI-Paygate
Length of output: 2763
🏁 Script executed:
#!/bin/bash
set -euo pipefail
python3 - <<'PY'
import json, textwrap, sys, re
from pathlib import Path
p = Path('gateway/internal/ai/openrouter.go')
text = p.read_text()
for i, line in enumerate(text.splitlines(), 1):
if 100 <= i <= 170:
print(f"{i:4}: {line}")
PYRepository: AnkanMisra/MicroAI-Paygate
Length of output: 2302
Surface OpenRouter stream errors in gateway/internal/ai/openrouter.go. Mid-stream SSE errors can arrive as a chunk with error and finish_reason: "error", but Recv() only returns delta.content, which can turn the failure into an empty/normal result instead of an error.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@gateway/internal/ai/openrouter.go` around lines 110 - 146, The
openRouterStream Recv() path is ignoring SSE error payloads and only extracting
delta.content, so mid-stream OpenRouter failures can be treated like normal
empty output. Update Recv() in openrouter.go to detect chunks containing an
error or a finish_reason of "error" before returning content, and return a
non-nil error instead of continuing. Keep the existing chunk parsing flow around
json.Unmarshal, choices, and delta, but add explicit handling for the error case
so callers can surface the stream failure.
| for { | ||
| text, err := stream.Recv() | ||
| if errors.Is(err, io.EOF) { | ||
| break | ||
| } | ||
| if err != nil { | ||
| fmt.Fprintf(c.Writer, "data: {\"error\": %q}\n\n", err.Error()) | ||
| flusher.Flush() | ||
| return | ||
| } | ||
|
|
||
| fullSummary += text | ||
| chunkBytes, _ := json.Marshal(map[string]string{"text": text}) | ||
| fmt.Fprintf(c.Writer, "data: %s\n\n", chunkBytes) | ||
| flusher.Flush() | ||
| } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Inconsistent SSE payload encoding: json.Marshal for text chunks vs. manual %q for error/receipt events.
Line 558 correctly uses json.Marshal for the text chunk, but error events (Line 552, 571, 577) and the receipt event (Line 586) are hand-built with fmt.Fprintf(..., "%q", ...). Go's %q uses Go string-literal escaping, not strict JSON escaping (e.g. some control-character escapes like \a, \v aren't valid JSON), so an upstream/receipt error containing such characters could produce invalid JSON on the wire and break the client's SSE JSON parsing for that chunk.
♻️ Suggested fix
- fmt.Fprintf(c.Writer, "data: {\"error\": %q}\n\n", err.Error())
+ errBytes, _ := json.Marshal(map[string]string{"error": err.Error()})
+ fmt.Fprintf(c.Writer, "data: %s\n\n", errBytes)Apply the same pattern to the other %q-based error/receipt writes in this function and in streamResultAndReceipt.
Also applies to: 569-588
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@gateway/main.go` around lines 546 - 561, The SSE response encoding is
inconsistent in gateway/main.go: text chunks use json.Marshal in the streaming
loop, but the error and receipt events in this function and
streamResultAndReceipt are still built with fmt.Fprintf and %q. Replace those
hand-built payloads with JSON marshaling for the error/receipt objects so all
SSE messages use valid JSON escaping and match the text-chunk encoding path.
| // 4. Generate Receipt | ||
| responseMap := map[string]interface{}{ | ||
| "result": fullSummary, | ||
| } | ||
| responseBody, _ := json.Marshal(responseMap) | ||
|
|
||
| receipt, err := GenerateReceipt(*paymentCtx, verifyResp.RecoveredAddress, c.Request.URL.Path, requestBody, responseBody) | ||
| if err != nil { | ||
| fmt.Fprintf(c.Writer, "data: {\"error\": \"receipt_generation_failed\"}\n\n") | ||
| flusher.Flush() | ||
| return | ||
| } | ||
|
|
||
| if err := storeReceiptWithContext(c.Request.Context(), receipt, getReceiptTTL()); err != nil { | ||
| fmt.Fprintf(c.Writer, "data: {\"error\": \"receipt_store_failed\"}\n\n") | ||
| flusher.Flush() | ||
| return | ||
| } | ||
|
|
||
| receiptJSON, _ := json.Marshal(receipt) | ||
| receiptBase64 := base64.StdEncoding.EncodeToString(receiptJSON) | ||
|
|
||
| // Send receipt as final event | ||
| fmt.Fprintf(c.Writer, "data: {\"receipt\": %q}\n\n", receiptBase64) | ||
| fmt.Fprintf(c.Writer, "data: [DONE]\n\n") | ||
| flusher.Flush() | ||
|
|
||
| // Store for cache middleware | ||
| c.Set("full_summary", fullSummary) |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick win
Duplicate receipt-generation/emit logic between handleSummarize and streamResultAndReceipt.
The receipt generate → store → emit-as-final-SSE-event → [DONE] sequence (Lines 563-588) is copy-pasted almost verbatim into streamResultAndReceipt (Lines 678-703). Extracting a shared helper (e.g. emitReceiptEvent(c, flusher, paymentCtx, recoveredAddr, path, requestBody, responseBody)) would prevent the two paths from drifting (e.g. one gets the %q-escaping fix above and the other doesn't).
Also applies to: 662-705
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@gateway/main.go` around lines 563 - 591, The receipt generation, storage, and
final SSE emission logic is duplicated between handleSummarize and
streamResultAndReceipt, which can easily drift over time. Extract the repeated
GenerateReceipt, storeReceiptWithContext, and final receipt/[DONE] write
sequence into a shared helper such as emitReceiptEvent that accepts the c
writer/flusher plus paymentCtx, recovered address, request path, requestBody,
and responseBody. Update both call sites to use the helper so any future fix to
receipt formatting or emission behavior is applied consistently in one place.
| c.Writer.Header().Set("Content-Type", "text/event-stream") | ||
| c.Writer.Header().Set("Cache-Control", "no-cache") | ||
| c.Writer.Header().Set("Connection", "keep-alive") | ||
|
|
||
| flusher, ok := c.Writer.(http.Flusher) | ||
| if !ok { | ||
| respondError(c, 500, "streaming_unsupported", fmt.Errorf("streaming unsupported")) | ||
| return fmt.Errorf("streaming unsupported") | ||
| } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Same header-order issue as handleSummarize: SSE headers set before the http.Flusher check.
Mirrors the issue flagged at Lines 535-544 in this file — if c.Writer doesn't support http.Flusher, respondError's JSON body inherits the already-set text/event-stream Content-Type.
🐛 Proposed fix
- c.Writer.Header().Set("Content-Type", "text/event-stream")
- c.Writer.Header().Set("Cache-Control", "no-cache")
- c.Writer.Header().Set("Connection", "keep-alive")
-
- flusher, ok := c.Writer.(http.Flusher)
- if !ok {
- respondError(c, 500, "streaming_unsupported", fmt.Errorf("streaming unsupported"))
- return fmt.Errorf("streaming unsupported")
- }
+ flusher, ok := c.Writer.(http.Flusher)
+ if !ok {
+ respondError(c, 500, "streaming_unsupported", fmt.Errorf("streaming unsupported"))
+ return fmt.Errorf("streaming unsupported")
+ }
+
+ c.Writer.Header().Set("Content-Type", "text/event-stream")
+ c.Writer.Header().Set("Cache-Control", "no-cache")
+ c.Writer.Header().Set("Connection", "keep-alive")📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| c.Writer.Header().Set("Content-Type", "text/event-stream") | |
| c.Writer.Header().Set("Cache-Control", "no-cache") | |
| c.Writer.Header().Set("Connection", "keep-alive") | |
| flusher, ok := c.Writer.(http.Flusher) | |
| if !ok { | |
| respondError(c, 500, "streaming_unsupported", fmt.Errorf("streaming unsupported")) | |
| return fmt.Errorf("streaming unsupported") | |
| } | |
| flusher, ok := c.Writer.(http.Flusher) | |
| if !ok { | |
| respondError(c, 500, "streaming_unsupported", fmt.Errorf("streaming unsupported")) | |
| return fmt.Errorf("streaming unsupported") | |
| } | |
| c.Writer.Header().Set("Content-Type", "text/event-stream") | |
| c.Writer.Header().Set("Cache-Control", "no-cache") | |
| c.Writer.Header().Set("Connection", "keep-alive") |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@gateway/main.go` around lines 664 - 672, Move the SSE header writes in the
streaming handler so they happen only after the http.Flusher assertion succeeds;
currently the Content-Type, Cache-Control, and Connection headers are set before
the check, causing respondError to return a JSON error with text/event-stream
headers when flushing is unsupported. Update the same streaming logic used in
the handler around the flusher check, keeping the header setup after the ok
branch and before normal streaming output.
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
|
@codex review |
AnkanMisra
added
enhancement
ChatGPT Codex Connector
![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: 8306ef593c
ℹ️ 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 result, ok := resp["result"].(string); ok { | ||
| // Instead of parsing the SSE stream from bodyBytes, we retrieve the full summary | ||
| // which handleSummarize saves in the gin context. | ||
| if fullSummaryVal, exists := c.Get("full_summary"); exists { |
There was a problem hiding this comment.
This new cache-miss branch no longer uses the bodyBytes := writer.body.Bytes() assignment just above it; cd gateway && go test ./... fails with ./cache.go:207:3: declared and not used: bodyBytes, so the gateway package cannot build. Remove the assignment or use it when parsing/caching the SSE result.
Useful? React with 👍 / 👎.
| if err := json.Unmarshal(w2.Body.Bytes(), &resp2); err != nil { | ||
| t.Fatalf("Failed to unmarshal response 2: %v", err) | ||
| parseSSEResultForTest := func(body []byte) string { | ||
| lines := strings.Split(string(body), "\n") |
There was a problem hiding this comment.
This new SSE parser calls strings.Split, strings.HasPrefix, and strings.Trim*, but gateway/cache_integration_test.go does not import strings; after the production compile error is fixed, the gateway test package still fails to compile with undefined: strings. Add the import so the cache integration test can run.
Useful? React with 👍 / 👎.
| if receiptHeader == "" { | ||
| t.Fatal("missing X-402-Receipt header") | ||
| var receiptBase64 string | ||
| lines := strings.Split(createResp.Body.String(), "\n") |
There was a problem hiding this comment.
This new receipt-stream parser also uses the strings package without importing it, so go test ./... reports undefined: strings in gateway/receipt_store_integration_test.go. Add the import before relying on this updated integration test.
Useful? React with 👍 / 👎.
| receiptBase64 := base64.StdEncoding.EncodeToString(receiptJSON) | ||
|
|
||
| // Send receipt as final event | ||
| fmt.Fprintf(c.Writer, "data: {\"receipt\": %q}\n\n", receiptBase64) |
There was a problem hiding this comment.
Moving the receipt into the SSE body here changes the signed summarize success contract, but the TypeScript SDK still reads the body as JSON in sdk/typescript/src/client.ts::readSuccess and only decodes receipts from X-402-Receipt in sdk/typescript/src/protocol/microai.ts::readReceipt; any SDK summarize() call against this gateway will now throw invalid JSON or lose the receipt. Update the SDK/smoke consumers to parse the SSE stream, or preserve the old JSON/header contract for clients that do not opt into streaming.
Useful? React with 👍 / 👎.
| if (parsed.text) { | ||
| summary += parsed.text; | ||
| if (onChunk) onChunk(summary); | ||
| } else if (parsed.receipt) { | ||
| receipt = safeDecodeReceiptHeader(parsed.receipt); |
There was a problem hiding this comment.
When the gateway emits data: {"error": ...} after a 200 stream has started (for example AI stream errors or receipt generation/storage failures in handleSummarize), this parser ignores the error object and returns the partial summary with a null receipt, causing the hook to record a successful paid summary. Track [DONE] and throw on parsed error events so users are not shown partial/no-receipt failures as success.
Useful? React with 👍 / 👎.
| fullSummary += text | ||
| chunkBytes, _ := json.Marshal(map[string]string{"text": text}) | ||
| fmt.Fprintf(c.Writer, "data: %s\n\n", chunkBytes) | ||
| flusher.Flush() |
There was a problem hiding this comment.
In the normal gateway stack, RequestTimeoutMiddleware wraps c.Writer in responseWriterShim; that shim's Flush() only flushes the original writer and never copies the buffered bytes until c.Next() finishes, so these per-token flushes do not reach the client and the entire SSE response is still held in memory until inference completes. Either exclude streaming routes from the buffering middleware or teach the shim to flush buffered chunks safely.
Useful? React with 👍 / 👎.
| responseMap := map[string]interface{}{ | ||
| "result": fullSummary, | ||
| } | ||
| responseBody, _ := json.Marshal(responseMap) |
There was a problem hiding this comment.
The receipt still signs a synthetic JSON body {"result": fullSummary} even though clients now receive an SSE payload containing text, receipt, and [DONE] events, so service.response_hash no longer proves the actual bytes returned for this paid request. Receipt verification against the delivered response will fail or require clients to know about an undocumented alternate body; hash a canonical streamed payload or keep the JSON body/header contract for receipt hashing.
Useful? React with 👍 / 👎.
| const textStr = await res.text(); | ||
| expect(textStr).toContain("data: {"); | ||
| expect(textStr).toContain("[DONE]"); |
There was a problem hiding this comment.
This replacement weakens the signed-payment E2E to pass for any 200 response that contains one data: line and [DONE], even if the summary is empty or the receipt event is missing/invalid. Parse the SSE stream here and assert at least one text chunk plus a decodable receipt so regressions in the new success contract are caught.
Useful? React with 👍 / 👎.
| if err != nil { | ||
| return "", err |
There was a problem hiding this comment.
A bare io.EOF from ReadBytes is returned exactly like the [DONE] sentinel, and handleSummarize treats any io.EOF as a clean completion; if the OpenRouter connection drops mid-stream or returns a non-SSE body, the gateway signs and caches a partial or empty summary as a successful paid response. Track whether [DONE] was actually received and return an error for EOF before that sentinel.
Useful? React with 👍 / 👎.
| if err := json.Unmarshal([]byte(data), &chunk); err == nil { | ||
| if text, ok := chunk["text"].(string); ok { | ||
| result += text |
There was a problem hiding this comment.
After the missing import is fixed, this parser only accumulates text fields from data: SSE chunks, but the test's mock OpenRouter server still returns the old non-streaming JSON shape, so the handler produces no text chunks and the cache integration test fails with empty results whenever Redis is available. Change the mock response to stream data: {"choices":[{"delta":{"content":...}}]} and [DONE] events.
Useful? React with 👍 / 👎.
AnkanMisra
left a comment
AnkanMisra
left a comment
There was a problem hiding this comment.
- Fix the go lint and go test
- Fix the bugs codex found
Co-authored-by: codex <codex@users.noreply.github.com>
Co-authored-by: codex <codex@users.noreply.github.com>
|
@AnkanMisra I have done the necessary changes kindly check and merge. |
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@gateway/middleware.go`:
- Around line 141-156: The buffering flush path in flushTo still races the
timeout handling because it does not honor bw.closed and the ctx.Done() branch
writes the 504 directly to origWriter without the shared mutex. Update flushTo
in the buffered writer so it returns immediately when closed is set, and make
the timeout response go through the same lock before calling WriteHeader on
origWriter. Keep the behavior aligned with Write and WriteHeader so SSE Flush
cannot interleave buffered output or an extra status write after timeout.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 36eaeef0-3354-4536-b7d3-27d51a46b1c4
📒 Files selected for processing (8)
gateway/cache.gogateway/cache_integration_test.gogateway/internal/ai/openrouter.gogateway/main.gogateway/middleware.gogateway/receipt_store_integration_test.gotests/e2e.test.tsweb/src/lib/x402-client.ts
💤 Files with no reviewable changes (1)
- gateway/cache.go
🚧 Files skipped from review as they are similar to previous changes (6)
- tests/e2e.test.ts
- gateway/receipt_store_integration_test.go
- web/src/lib/x402-client.ts
- gateway/internal/ai/openrouter.go
- gateway/main.go
- gateway/cache_integration_test.go
| b.mu.Lock() | ||
| defer b.mu.Unlock() | ||
| if !b.flushed { | ||
| for k, vv := range b.head { | ||
| for _, v := range vv { | ||
| w.Header().Add(k, v) | ||
| } | ||
| } | ||
| w.WriteHeader(b.Status()) | ||
| b.flushed = true | ||
| } | ||
| if b.buf.Len() > 0 { | ||
| _, _ = w.Write(b.buf.Bytes()) | ||
| b.buf.Reset() | ||
| } | ||
| w.WriteHeader(b.Status()) | ||
| _, _ = w.Write(b.buf.Bytes()) | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🔴 Critical | 🏗️ Heavy lift
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
rg -nP -C4 'flushTo|closed|WriteHeader\(http\.StatusGatewayTimeout\)' gateway/middleware.goRepository: AnkanMisra/MicroAI-Paygate
Length of output: 3165
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "=== gateway/middleware.go (relevant sections) ==="
sed -n '1,320p' gateway/middleware.go | cat -n | sed -n '1,320p' | rg -n '(^ *[0-9]+ *|flushTo|closed|finished|ctx.Done|WriteHeader\(http.StatusGatewayTimeout\)|Flush\(\))'
echo
echo "=== gateway/main.go (SSE/flush call sites) ==="
rg -n -C3 'flusher\.Flush\(|Flush\(\)|streamResultAndReceipt|ResponseWriterShim|bufferedWriter|middleware' gateway/main.go gateway -g '!gateway/middleware.go'Repository: AnkanMisra/MicroAI-Paygate
Length of output: 34149
flushTo still races the timeout response and should stop after closed
bw.closed is respected by Write/WriteHeader, but flushTo ignores it and the ctx.Done() branch writes the 504 directly to origWriter without taking the same lock. An SSE Flush() can still interleave with the timeout path and send buffered bytes or a superfluous WriteHeader into the middle of the stream. Guard flushTo on closed and have the timeout branch serialize through the same mutex before writing the 504.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@gateway/middleware.go` around lines 141 - 156, The buffering flush path in
flushTo still races the timeout handling because it does not honor bw.closed and
the ctx.Done() branch writes the 504 directly to origWriter without the shared
mutex. Update flushTo in the buffered writer so it returns immediately when
closed is set, and make the timeout response go through the same lock before
calling WriteHeader on origWriter. Keep the behavior aligned with Write and
WriteHeader so SSE Flush cannot interleave buffered output or an extra status
write after timeout.
Source: Coding guidelines
|
@codex review |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 39367db7bb
ℹ️ 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".
| responseBody, _ := json.Marshal(responseMap) | ||
|
|
||
| receipt, err := GenerateReceipt(*paymentCtx, verifyResp.RecoveredAddress, c.Request.URL.Path, requestBody, responseBody) |
There was a problem hiding this comment.
For Accept: text/event-stream requests, this signs the JSON body {"result": fullSummary} even though the response actually sent below is an SSE stream of data: events plus the receipt event. Any client/SDK that verifies service.response_hash against the payload it received will reject the receipt or, worse, treat the receipt as proving bytes that were never returned. Define and hash a canonical streamed payload (or keep the receipt out-of-band and document the logical hash) so the signed receipt matches the SSE contract.
Useful? React with 👍 / 👎.
| func (rws *responseWriterShim) Flush() { | ||
| rws.bw.flushTo(rws.orig) |
There was a problem hiding this comment.
When handleSummarize streams tokens, each Flush() now commits the buffered 200 text/event-stream response to the client, but RequestTimeoutMiddleware can still hit ctx.Done() afterward and write the JSON 504 body on the same connection. For streams that run longer than AI_REQUEST_TIMEOUT_SECONDS, this appends a non-SSE JSON timeout to an already-started SSE response instead of a clean SSE error/close. Track that the response was flushed and avoid sending the timeout response once headers/body have been committed.
Useful? React with 👍 / 👎.
| if receiptHeader == "" { | ||
| t.Fatal("missing X-402-Receipt header") | ||
| var receiptBase64 string | ||
| lines := strings.Split(createResp.Body.String(), "\n") |
There was a problem hiding this comment.
This test now searches the summarize response body for SSE receipt events, but createReq never sets Accept: text/event-stream, so the handler's success path returns JSON with X-402-Receipt instead of SSE. Also, the unchanged OpenRouter mock above still returns non-stream JSON to GenerateStream, which makes the request fail with 502 before this parsing. Either set Accept and make the mock emit data: chunks plus [DONE], or keep asserting the JSON/header response.
Useful? React with 👍 / 👎.
| if wantsSSE { | ||
| c.Writer.Header().Set("Content-Type", "text/event-stream") | ||
| fmt.Fprintf(c.Writer, "data: {\"error\": \"receipt_generation_failed\"}\n\n") | ||
| if flusher, ok := c.Writer.(http.Flusher); ok { flusher.Flush() } | ||
| } |
There was a problem hiding this comment.
When a cache hit is served to a client that did not send Accept: text/event-stream and receipt generation fails, this branch only writes an error for SSE callers and otherwise just returns to CacheMiddleware, which logs and aborts without sending a response. That regresses the old generateAndSendReceipt behavior and can leave non-SSE paid cache hits looking like empty/default 200 responses during receipt key or store failures; call respondError for the non-SSE path before returning.
Useful? React with 👍 / 👎.
| } | ||
| if err := json.Unmarshal(w2.Body.Bytes(), &resp2); err != nil { | ||
| t.Fatalf("Failed to unmarshal response 2: %v", err) | ||
| parseSSEResultForTest := func(body []byte) string { |
There was a problem hiding this comment.
The new assertion helper parses only data: events, but the requests built by makeRequest above still omit Accept: text/event-stream, so both the cache miss and cache hit take the JSON response path and this parser returns an empty string. Set the Accept header in the helper, or keep this test on the JSON/header contract and assert result from the JSON body.
Useful? React with 👍 / 👎.
| return fetch(`${getGatewayUrl()}/api/ai/summarize`, { | ||
| method: "POST", | ||
| headers: { "Content-Type": "application/json", ...headers }, | ||
| headers: { "Content-Type": "application/json", "Accept": "text/event-stream", ...headers }, |
There was a problem hiding this comment.
This makes the web client negotiate text/event-stream by default, but gateway/openapi.yaml and the README still document successful summarizes as application/json with the receipt in X-402-Receipt only. Anyone generating clients from the spec or following the examples will call res.json()/read the header and fail against the new default web flow; add the SSE media type, text/receipt/[DONE] events, and updated examples to the API docs.
Useful? React with 👍 / 👎.
| // generateAndSendReceipt sends error response if it fails? | ||
| // No, it returns error, we might have already written status if we aren't careful. | ||
| // Let's implement generateAndSendReceipt to handle sending response. | ||
| wantsSSE := c.GetHeader("Accept") == "text/event-stream" |
There was a problem hiding this comment.
This exact comparison misses valid Accept values such as text/event-stream, application/json or text/event-stream;q=1, so standards-compliant clients that request SSE alongside fallbacks receive JSON instead and their stream parser fails. Parse the comma-separated media ranges case-insensitively instead of requiring the header to be exactly one string.
Useful? React with 👍 / 👎.
2 participants
Resolves #243
This prevents holding up the HTTP handler goroutine for the duration of inference.
Summary
Providerinterface and backend implementations (OpenRouter, Ollama) to support returning an incremental token stream viaGenerateStream.handleSummarizeGo handler to send Server-Sent Events (SSE) to the client, preventing long HTTP handler blocks and reducing memory pressure during inference.CacheMiddlewareto seamlessly serve cache hits via identical SSE streams and asynchronously store completed SSE outputs to Redis.x402-client.tsto consume the stream chunks natively usingres.body.getReader(), mapping incremental data to progressive UI state updates inuse-x402.ts.Type Of Change
Affected Areas
gateway/)verifier/)web/)tests/,run_e2e.sh)bench/)deploy/, Docker, env, workflows)Contributor Checklist
Verification
List the exact commands you ran and their result.
Screenshots
Required for visible web UI changes. Remove this section if not applicable.
(N/A - the UI layout remains the same, text simply streams in instead of loading all at once)
Notes For Reviewers
X-402-ReceiptHTTP header, but is sent instead as the final data payload in the SSE stream (as{"receipt": "<base64>"}).CacheMiddlewarewas carefully updated so that cache hits continue to look like an SSE stream to clients for API contract stability.