fix(proxy): filtra chunks SSE sin id y corta en [DONE]
Grok Build siempre usa streaming. El upstream (OpenCode Zen) envía chunks no-estándar (inference-cost sin id, cost tras [DONE]) que rompen el deserializador serde de Rust con 'missing field id'. streamResponse ahora filtra eventos SSE que no tengan campo id y deja de forwardear después de [DONE].
This commit is contained in:
+28
-4
@@ -197,7 +197,8 @@ func (p *Proxy) ForwardChatCompletion(w http.ResponseWriter, r *http.Request, bo
|
|||||||
return p.nonStreamResponse(w, resp, req.Model)
|
return p.nonStreamResponse(w, resp, req.Model)
|
||||||
}
|
}
|
||||||
|
|
||||||
// streamResponse forwards SSE chunks directly from upstream to client.
|
// streamResponse forwards SSE chunks from upstream to client, filtering out
|
||||||
|
// non-standard events (missing `id`) and stopping after [DONE].
|
||||||
func (p *Proxy) streamResponse(w http.ResponseWriter, resp *http.Response) error {
|
func (p *Proxy) streamResponse(w http.ResponseWriter, resp *http.Response) error {
|
||||||
w.Header().Set("Content-Type", "text/event-stream")
|
w.Header().Set("Content-Type", "text/event-stream")
|
||||||
w.Header().Set("Cache-Control", "no-cache")
|
w.Header().Set("Cache-Control", "no-cache")
|
||||||
@@ -212,17 +213,40 @@ func (p *Proxy) streamResponse(w http.ResponseWriter, resp *http.Response) error
|
|||||||
scanner := bufio.NewScanner(resp.Body)
|
scanner := bufio.NewScanner(resp.Body)
|
||||||
scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024)
|
scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024)
|
||||||
|
|
||||||
|
done := false
|
||||||
for scanner.Scan() {
|
for scanner.Scan() {
|
||||||
line := scanner.Text()
|
line := scanner.Text()
|
||||||
// Write the line. If the client disconnects, stop silently.
|
if done {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
|
if strings.HasPrefix(line, "data: ") {
|
||||||
|
data := strings.TrimPrefix(line, "data: ")
|
||||||
|
if data == "[DONE]" {
|
||||||
|
if _, err := fmt.Fprint(w, line+"\n"); err != nil {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
flusher.Flush()
|
||||||
|
done = true
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
|
// Only forward SSE events that have an "id" field — filters out
|
||||||
|
// non-standard chunks (e.g. x-opencode-type, standalone cost)
|
||||||
|
var obj map[string]any
|
||||||
|
if json.Unmarshal([]byte(data), &obj) == nil {
|
||||||
|
if _, ok := obj["id"]; !ok {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
if _, err := fmt.Fprint(w, line+"\n"); err != nil {
|
if _, err := fmt.Fprint(w, line+"\n"); err != nil {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
flusher.Flush()
|
flusher.Flush()
|
||||||
}
|
}
|
||||||
|
|
||||||
// Scanner errors (e.g., connection reset) are not real errors
|
|
||||||
// after streaming has started — the client likely disconnected.
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user