From a8f0baaaa0f44a9154e426aeb8746a1d70d496ef Mon Sep 17 00:00:00 2001 From: yetone Date: Thu, 1 Oct 2026 03:52:23 +0800 Subject: [PATCH] =?UTF-8?q?gateway:=20a=20refusal=20after=20only=20thinkin?= =?UTF-8?q?g=20goes=20to=20the=20next=20account=20(#248,=20Koohoko:=20safe?= =?UTF-8?q?guards=20stopped=20the=20response=20above)=20=E2=80=94=20Opus?= =?UTF-8?q?=205.5=20at=20xhigh=20thought=2010=E2=80=9325s=20and=20then=20e?= =?UTF-8?q?nded=20with=20stop=5Freason=20refusal,=20and=20magpie=20had=20c?= =?UTF-8?q?ounted=20the=20first=20thinking=20delta=20as=20the=20reply=20ha?= =?UTF-8?q?ving=20started,=20so=20the=20stream=20was=20already=20Claude=20?= =?UTF-8?q?Code's=20and=20the=20refusal=20passed=20through=20instead=20of?= =?UTF-8?q?=20the=20next=20account=20being=20asked.=20Reasoning=20(Anthrop?= =?UTF-8?q?ic=20thinking,=20Responses=20reasoning=20and=20summaries,=20Cha?= =?UTF-8?q?t=20reasoning=5Fcontent,=20Gemini=20thought=20parts)=20is=20now?= =?UTF-8?q?=20held=20with=20the=20reply,=20up=20to=204=20minutes,=20and=20?= =?UTF-8?q?let=20through=20with=20it=20as=20soon=20as=20text=20or=20a=20ca?= =?UTF-8?q?ll=20comes.=20A=20ChatGPT=20bio=5Fpolicy=20translated=20for=20C?= =?UTF-8?q?laude=20Code=20or=20a=20Chat=20agent=20kept=20only=20its=20mess?= =?UTF-8?q?age,=20read=20as=20any=20failure,=20and=20set=20the=20account?= =?UTF-8?q?=20aside;=20the=20refusal's=20code=20now=20survives=20translati?= =?UTF-8?q?on.=20TestRefusalAfterThinkingFailsOver=20and=20TestThinkingHel?= =?UTF-8?q?dLonger=20fail=20without=20the=20change.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/gateway/anthropic.go | 8 +- internal/gateway/chat.go | 8 +- internal/gateway/fallback.go | 100 ++++++++++++++++- internal/gateway/ir.go | 3 + internal/gateway/refusal_test.go | 12 +- internal/gateway/refusal_thinking_test.go | 127 ++++++++++++++++++++++ internal/gateway/responses.go | 10 +- 7 files changed, 255 insertions(+), 13 deletions(-) create mode 100644 internal/gateway/refusal_thinking_test.go diff --git a/internal/gateway/anthropic.go b/internal/gateway/anthropic.go index 98fea28d..5bc4e16d 100644 --- a/internal/gateway/anthropic.go +++ b/internal/gateway/anthropic.go @@ -517,7 +517,7 @@ func decodeAnthropic(data string, emit func(Event)) error { } emit(Event{Kind: KUsage, Usage: ev.Usage.usage()}) case "error": - emit(Event{Kind: KError, Text: ev.Error.Message}) + emit(Event{Kind: KError, Text: ev.Error.Message, Code: refusedCode(data)}) } return nil } @@ -671,7 +671,11 @@ func (e *anthropicEncoder) event(ev Event) { e.index++ case KError: e.close() - e.w.event("error", map[string]any{"type": "error", "error": map[string]any{"type": "api_error", "message": ev.Text}}) + failed := map[string]any{"type": "api_error", "message": ev.Text} + if ev.Code != "" { + failed["code"] = ev.Code // a refusal, for the next account to be asked + } + e.w.event("error", map[string]any{"type": "error", "error": failed}) } e.col.add(ev) } diff --git a/internal/gateway/chat.go b/internal/gateway/chat.go index 85282995..0be181ce 100644 --- a/internal/gateway/chat.go +++ b/internal/gateway/chat.go @@ -610,7 +610,7 @@ func (d *chatDecoder) decode(data string, emit func(Event)) error { return nil } if ch.Error != nil { - emit(Event{Kind: KError, Text: ch.Error.Message}) + emit(Event{Kind: KError, Text: ch.Error.Message, Code: refusedCode(data)}) return nil } if !d.started { @@ -792,7 +792,11 @@ func (e *chatEncoder) event(ev Event) { "function": map[string]any{"arguments": ev.Text}}}}, nil, nil) } case KError: - e.w.event("", map[string]any{"error": map[string]any{"message": ev.Text, "type": "api_error"}}) + failed := map[string]any{"message": ev.Text, "type": "api_error"} + if ev.Code != "" { + failed["code"] = ev.Code // a refusal, for the next account to be asked + } + e.w.event("", map[string]any{"error": failed}) } e.col.add(ev) } diff --git a/internal/gateway/fallback.go b/internal/gateway/fallback.go index 3098ccf2..43c6cd82 100644 --- a/internal/gateway/fallback.go +++ b/internal/gateway/fallback.go @@ -560,6 +560,9 @@ type holdWriter struct { // buffered: the vendor said it holds the reply back for safety checks, // which may end in a refusal: held longer (holdBuffered) buffered bool + // thinking: the reply has reasoned but said nothing yet, which a + // refusal may still end: held longer (holdThinking) + thinking bool ended bool // the stream's last event was written: the reply is whole tail []byte // the end of the last write, for a marker split across two @@ -659,6 +662,14 @@ const ( // a stream's next event. const holdBuffered = 4 * time.Minute +// holdThinking is how long a stream is held while all it has is reasoning, +// shown to nobody yet: Claude refused Claude Code after 10–25s of thinking, +// which had let the stream through, so Claude Code got the refusal ("…'s +// safeguards stopped the response above") and the next account was never +// asked (#248). A reply that goes on to say something is let through, its +// reasoning with it, as soon as it does. +const holdThinking = 4 * time.Minute + // scan reads the held stream's events so far: an error before any content // fails it; content, or waiting too long for it, lets it through. func (h *holdWriter) scan() { @@ -675,6 +686,9 @@ func (h *holdWriter) scan() { case eventBuffering: h.buffered = true continue + case eventThinking: + h.thinking = true + continue case eventError: h.failure, h.failMsg = status, msg return @@ -689,6 +703,9 @@ func (h *holdWriter) scan() { if h.buffered { longest = holdBuffered } + if h.thinking { + longest = max(longest, holdThinking) + } if h.held.Len() > holdMost || time.Since(h.since) > longest { h.flow() } @@ -790,6 +807,7 @@ const ( eventError eventRefusal // the reply's end, by the vendor's safety filter eventBuffering // a lead saying the reply is held back for safety checks + eventThinking // reasoning, before anything is said: a refusal may yet end it ) // refusedStatus is what a refusal with nothing said is answered as: a @@ -834,6 +852,16 @@ type streamPart struct { } `json:"summary"` } +// thought tells whether a part is reasoning, said or not yet: a thinking +// block, a reasoning item or its summary. It says nothing to the reader. +func (p streamPart) thought() bool { + switch p.Type { + case "thinking", "redacted_thinking", "summary_text", "reasoning_text", "reasoning": + return true + } + return false +} + // said tells whether a part says anything an agent would show or act on: // text, reasoning, a call. What isn't known is taken as saying something. func (p streamPart) said() bool { @@ -956,6 +984,15 @@ func streamEvent(ev []byte) (kind, status int, msg string) { return eventRefusal, refusedStatus, note } return errOf(v.Error) + // reasoning, before a word of the reply: held with it, and longer, for + // a refusal after it to go to another as one with nothing said (#248) + case typ == "content_block_start" && v.ContentBlock.thought(), + typ == "content_block_delta" && anthropicDeltaThinks(v.Delta), + (typ == "response.output_item.added" || typ == "response.output_item.done") && v.Item.thought(), + (typ == "response.reasoning_summary_part.added" || typ == "response.reasoning_summary_part.done") && v.Part.thought(), + typ == "response.reasoning_summary_text.delta", typ == "response.reasoning_text.delta", + typ == "response.reasoning_summary_text.done", typ == "response.reasoning_text.done": + return eventThinking, 0, "" // what only frames a reply, before anything is said in it — held with // its start, so a refusal after it can still go to another case typ == "content_block_start" && !v.ContentBlock.said(), @@ -988,10 +1025,16 @@ func streamEvent(ev []byte) (kind, status int, msg string) { if r, ok := filterReasons[v.PromptFeedback.BlockReason]; ok { return refusal(r) } + thinks := false for _, c := range *v.Candidates { for _, p := range c.Content.Parts { var part struct { - Text string `json:"text"` + Text string `json:"text"` + Thought bool `json:"thought"` + } + if json.Unmarshal(p, &part) == nil && part.Thought && part.Text != "" { + thinks = true // its reasoning, said to nobody yet + continue } if json.Unmarshal(p, &part) != nil || part.Text != "" || !bytes.Contains(p, []byte(`"text"`)) { return eventContent, 0, "" // text, or a call @@ -1004,6 +1047,9 @@ func streamEvent(ev []byte) (kind, status int, msg string) { return eventContent, 0, "" } } + if thinks { + return eventThinking, 0, "" + } return eventLead, 0, "" case typ == "" && len(v.PromptFeedback.BlockReason) > 0: if r, ok := filterReasons[v.PromptFeedback.BlockReason]; ok { @@ -1028,11 +1074,19 @@ func streamEvent(ev []byte) (kind, status int, msg string) { return eventLead, 0, "" case typ == "" && v.Choices != nil: // a Chat chunk: the first says only who speaks + thinks := false for _, c := range *v.Choices { for k, x := range c.Delta { - if k != "role" && x != nil && x != "" { + if k == "role" || x == nil || x == "" { + continue + } + if k != "reasoning_content" && k != "reasoning" { return eventContent, 0, "" } + thinks = true + } + if thinks && c.FinishReason == nil { + continue } if c.FinishReason != nil { if r, ok := filterReasons[*c.FinishReason]; ok { @@ -1041,11 +1095,24 @@ func streamEvent(ev []byte) (kind, status int, msg string) { return eventContent, 0, "" } } + if thinks { + return eventThinking, 0, "" + } return eventLead, 0, "" } return eventContent, 0, "" } +// anthropicDeltaThinks tells whether an Anthropic content_block_delta is +// reasoning with something in it. +func anthropicDeltaThinks(raw json.RawMessage) bool { + var d struct { + Type string `json:"type"` + Thinking string `json:"thinking"` + } + return json.Unmarshal(raw, &d) == nil && d.Type == "thinking_delta" && d.Thinking != "" +} + // anthropicDeltaSays tells whether an Anthropic content_block_delta says // anything: text or reasoning, or a call's input. A signature doesn't. func anthropicDeltaSays(raw json.RawMessage) bool { @@ -1140,6 +1207,35 @@ func policyRefusal(raw []byte) (string, bool) { return "", false } +// refusedCode is the code of a stream's error event (its data) when it is +// the vendor's safety filter refusing, for a translation of it to keep: +// said in another protocol as only its message, ChatGPT's bio_policy read +// to Claude Code as any failure, and the account was set aside for it +// rather than the next asked as after a refusal (#248). +func refusedCode(data string) string { + var v struct { + Error json.RawMessage `json:"error"` + Response struct { + Error json.RawMessage `json:"error"` + } `json:"response"` + } + if json.Unmarshal([]byte(data), &v) != nil { + return "" + } + for _, raw := range []json.RawMessage{v.Response.Error, v.Error, json.RawMessage(data)} { + if len(raw) == 0 || string(raw) == "null" { + continue + } + if _, ok := policyRefusal(raw); ok { + if c := errorCode(raw); c != "" && c != "error" { + return c + } + return "content_filter" + } + } + return "" +} + // refusedReply tells whether a whole reply, not streamed, is the vendor's // safety filter refusing with nothing said: Anthropic's stop_reason // "refusal", Chat's finish_reason "content_filter", Responses' incomplete diff --git a/internal/gateway/ir.go b/internal/gateway/ir.go index ebe03257..32a3e4a3 100644 --- a/internal/gateway/ir.go +++ b/internal/gateway/ir.go @@ -162,6 +162,9 @@ type Event struct { Stop string // stop | length | tool | filter Usage Usage Hits []Hit + // Code: for KError, the vendor's code when it is its safety filter + // refusing (bio_policy, content_filter…), kept in translation (#248) + Code string } // Usage counts tokens. diff --git a/internal/gateway/refusal_test.go b/internal/gateway/refusal_test.go index b401d220..54ecebc1 100644 --- a/internal/gateway/refusal_test.go +++ b/internal/gateway/refusal_test.go @@ -254,20 +254,24 @@ func TestStreamEventRefusals(t *testing.T) { }{ {`event: content_block_start` + "\n" + `data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}`, eventLead}, {`event: content_block_start` + "\n" + `data: {"type":"content_block_start","index":0,"content_block":{"type":"tool_use","id":"t","name":"x","input":{}}}`, eventContent}, - {`event: content_block_start` + "\n" + `data: {"type":"content_block_start","index":0,"content_block":{"type":"thinking","thinking":""}}`, eventLead}, + {`event: content_block_start` + "\n" + `data: {"type":"content_block_start","index":0,"content_block":{"type":"thinking","thinking":""}}`, eventThinking}, {`event: content_block_delta` + "\n" + `data: {"type":"content_block_delta","index":0,"delta":{"type":"signature_delta","signature":"s"}}`, eventLead}, - {`event: content_block_delta` + "\n" + `data: {"type":"content_block_delta","index":0,"delta":{"type":"thinking_delta","thinking":"hm"}}`, eventContent}, + {`event: content_block_delta` + "\n" + `data: {"type":"content_block_delta","index":0,"delta":{"type":"thinking_delta","thinking":"hm"}}`, eventThinking}, {`event: content_block_delta` + "\n" + `data: {"type":"content_block_delta","index":0,"delta":{"type":"input_json_delta","partial_json":""}}`, eventContent}, {`event: message_delta` + "\n" + `data: {"type":"message_delta","delta":{"stop_reason":"refusal"}}`, eventRefusal}, {`event: message_delta` + "\n" + `data: {"type":"message_delta","delta":{"stop_reason":"end_turn"}}`, eventContent}, {`data: {"id":"c","choices":[{"index":0,"delta":{},"finish_reason":"content_filter"}]}`, eventRefusal}, {`data: {"id":"c","choices":[{"index":0,"delta":{"content":"hi"},"finish_reason":"content_filter"}]}`, eventContent}, {`data: {"id":"c","choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}`, eventContent}, - {`data: {"type":"response.output_item.added","item":{"type":"reasoning","summary":[]}}`, eventLead}, + {`data: {"type":"response.output_item.added","item":{"type":"reasoning","summary":[]}}`, eventThinking}, + {`data: {"type":"response.reasoning_summary_text.delta","delta":"**Plan**"}`, eventThinking}, + {`data: {"id":"c","choices":[{"index":0,"delta":{"reasoning_content":"hm"}}]}`, eventThinking}, + {`data: {"id":"c","choices":[{"index":0,"delta":{"reasoning_content":"hm"},"finish_reason":"content_filter"}]}`, eventRefusal}, + {`data: {"candidates":[{"content":{"parts":[{"text":"hm","thought":true}]}}]}`, eventThinking}, {`data: {"type":"response.output_item.added","item":{"type":"function_call","name":"x"}}`, eventContent}, {`data: {"type":"response.output_item.done","item":{"type":"message","content":[{"type":"output_text","text":"hi"}]}}`, eventContent}, {`data: {"type":"response.output_text.delta","delta":""}`, eventLead}, - {`data: {"type":"response.reasoning_summary_part.added","part":{"type":"summary_text","text":""}}`, eventLead}, + {`data: {"type":"response.reasoning_summary_part.added","part":{"type":"summary_text","text":""}}`, eventThinking}, {`data: {"type":"response.incomplete","response":{"status":"incomplete","incomplete_details":{"reason":"content_filter"}}}`, eventRefusal}, {`data: {"type":"response.incomplete","response":{"status":"incomplete","incomplete_details":{"reason":"max_output_tokens"}}}`, eventContent}, {`data: {"type":"response.completed","response":{"status":"completed"}}`, eventContent}, diff --git a/internal/gateway/refusal_thinking_test.go b/internal/gateway/refusal_thinking_test.go new file mode 100644 index 00000000..873ef423 --- /dev/null +++ b/internal/gateway/refusal_thinking_test.go @@ -0,0 +1,127 @@ +package gateway + +import ( + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/yetone/magpie/internal/provider" +) + +// Claude refusing after it has only thought, as Opus did Claude Code +// through a group (#248): thinking streamed, then stop_reason "refusal" +// with no word of the reply said. 381–707 tokens out, 200 on the routing +// page, and Claude Code's "…'s safeguards stopped the response above". +var anthropicThoughtThenRefused = sse(`event: message_start`+"\n"+`data: {"type":"message_start","message":{"id":"m1","role":"assistant","model":"m","content":[],"usage":{"input_tokens":2,"cache_read_input_tokens":32749}}}`, + `event: content_block_start`+"\n"+`data: {"type":"content_block_start","index":0,"content_block":{"type":"thinking","thinking":"","signature":""}}`, + `event: content_block_delta`+"\n"+`data: {"type":"content_block_delta","index":0,"delta":{"type":"thinking_delta","thinking":"The user wants a plan for"}}`, + `event: ping`+"\n"+`data: {"type":"ping"}`, + `event: content_block_delta`+"\n"+`data: {"type":"content_block_delta","index":0,"delta":{"type":"thinking_delta","thinking":" the next step."}}`, + `event: content_block_delta`+"\n"+`data: {"type":"content_block_delta","index":0,"delta":{"type":"signature_delta","signature":"sig"}}`, + `event: content_block_stop`+"\n"+`data: {"type":"content_block_stop","index":0}`, + `event: message_delta`+"\n"+`data: {"type":"message_delta","delta":{"stop_reason":"refusal","stop_sequence":null},"usage":{"output_tokens":381}}`, + `event: message_stop`+"\n"+`data: {"type":"message_stop"}`) + +var anthropicThoughtThenAnswered = sse(`event: message_start`+"\n"+`data: {"type":"message_start","message":{"id":"m1","role":"assistant","model":"m","content":[]}}`, + `event: content_block_start`+"\n"+`data: {"type":"content_block_start","index":0,"content_block":{"type":"thinking","thinking":"","signature":""}}`, + `event: content_block_delta`+"\n"+`data: {"type":"content_block_delta","index":0,"delta":{"type":"thinking_delta","thinking":"hm"}}`, + `event: content_block_stop`+"\n"+`data: {"type":"content_block_stop","index":0}`, + `event: content_block_start`+"\n"+`data: {"type":"content_block_start","index":1,"content_block":{"type":"text","text":""}}`, + `event: content_block_delta`+"\n"+`data: {"type":"content_block_delta","index":1,"delta":{"type":"text_delta","text":"from a"}}`, + `event: content_block_stop`+"\n"+`data: {"type":"content_block_stop","index":1}`, + `event: message_delta`+"\n"+`data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":4}}`, + `event: message_stop`+"\n"+`data: {"type":"message_stop"}`) + +const claudeCodeAsk = `{"model":"group/g","max_tokens":10,"stream":true,"thinking":{"type":"adaptive"},"messages":[{"role":"user","content":"hi"}]}` + +// A refusal after nothing but reasoning is one with nothing said: the +// reasoning was shown to nobody, and the next member answers (#248) — for +// Claude Code on Anthropic's API, Codex on Responses, and each protocol's +// own reasoning. +func TestRefusalAfterThinkingFailsOver(t *testing.T) { + chatThought := sse(`data: {"id":"c1","choices":[{"index":0,"delta":{"role":"assistant","reasoning_content":"thinking it over"}}]}`, + `data: {"id":"c1","choices":[{"index":0,"delta":{},"finish_reason":"content_filter"}]}`, + `data: [DONE]`) + responsesThought := sse(`event: response.created`+"\n"+`data: {"type":"response.created","response":{"id":"resp_1","status":"in_progress","output":[]}}`, + `event: response.output_item.added`+"\n"+`data: {"type":"response.output_item.added","output_index":0,"item":{"id":"rs_1","type":"reasoning","summary":[]}}`, + `event: response.reasoning_summary_part.added`+"\n"+`data: {"type":"response.reasoning_summary_part.added","item_id":"rs_1","output_index":0,"summary_index":0,"part":{"type":"summary_text","text":""}}`, + `event: response.reasoning_summary_text.delta`+"\n"+`data: {"type":"response.reasoning_summary_text.delta","item_id":"rs_1","output_index":0,"summary_index":0,"delta":"**Planning the step**"}`, + `event: response.reasoning_summary_text.done`+"\n"+`data: {"type":"response.reasoning_summary_text.done","item_id":"rs_1","output_index":0,"summary_index":0,"text":"**Planning the step**"}`, + `event: response.output_item.done`+"\n"+`data: {"type":"response.output_item.done","output_index":0,"item":{"id":"rs_1","type":"reasoning","summary":[{"type":"summary_text","text":"**Planning the step**"}]}}`, + `event: response.failed`+"\n"+`data: {"type":"response.failed","response":{"id":"resp_1","status":"failed","error":{"code":"bio_policy","message":"This content was flagged for possible biological risk."}}}`) + for _, x := range []struct { + name string + setup func(t *testing.T, s *scripted) + reply string + path string + ask string + }{ + {"claude to claude code", func(t *testing.T, s *scripted) { scriptedOn(t, "a", provider.Anthropic, s) }, anthropicThoughtThenRefused, "/v1/messages", claudeCodeAsk}, + {"claude to codex", func(t *testing.T, s *scripted) { scriptedOn(t, "a", provider.Anthropic, s) }, anthropicThoughtThenRefused, "/v1/responses", codexAsk}, + {"chat to claude code", func(t *testing.T, s *scripted) { scriptedOn(t, "a", provider.Chat, s) }, chatThought, "/v1/messages", claudeCodeAsk}, + {"responses to codex", func(t *testing.T, s *scripted) { responsesOn(t, "a", s) }, responsesThought, "/v1/responses", codexAsk}, + {"responses to claude code", func(t *testing.T, s *scripted) { responsesOn(t, "a", s) }, responsesThought, "/v1/messages", claudeCodeAsk}, + } { + t.Run(x.name, func(t *testing.T) { + fresh(t) + a := &scripted{replies: []reply{{200, "text/event-stream", x.reply}}} + b := &scripted{replies: []reply{{200, "text/event-stream", anthropicAnswer}}} + x.setup(t, a) + scriptedOn(t, "b", provider.Anthropic, b) + refusalGroup(t, "a/m", "b/m") + s := New() + code, body := sendTo(s, x.path, x.ask) + if code != 200 || !strings.Contains(body, "from b") || strings.Contains(body, `"refusal"`) || strings.Contains(body, "Planning") || + strings.Contains(body, "thinking it over") || strings.Contains(body, "user wants") || a.n != 1 || b.n != 1 { + t.Fatalf("%d %s (a %d, b %d)", code, body, a.n, b.n) + } + r := lastRoute(s) + if len(r.Tries) != 2 || r.Tries[0].Fail != failRefused || r.Tries[0].Rest != nil || r.Status != 200 { + t.Fatalf("tries: %+v", r.Tries) + } + }) + } + + // thinking that goes on to an answer comes through whole, as it came + fresh(t) + a := &scripted{replies: []reply{{200, "text/event-stream", anthropicThoughtThenAnswered}}} + b := &scripted{replies: []reply{{200, "text/event-stream", anthropicAnswer}}} + scriptedOn(t, "a", provider.Anthropic, a) + scriptedOn(t, "b", provider.Anthropic, b) + refusalGroup(t, "a/m", "b/m") + if code, body := sendTo(New(), "/v1/messages", claudeCodeAsk); code != 200 || body != anthropicThoughtThenAnswered || b.n != 0 { + t.Fatalf("answered: %d %s (b %d)", code, body, b.n) + } +} + +// Reasoning keeps the stream held past the 15s a quiet one is: Claude +// thought for 16–25s before refusing (#248). Past holdThinking it goes +// through, as it did. +func TestThinkingHeldLonger(t *testing.T) { + events := strings.SplitAfter(anthropicThoughtThenRefused, "\n\n") + start := func(ago time.Duration) *holdWriter { + h := newHoldWriter(httptest.NewRecorder(), true) + h.Header().Set("Content-Type", "text/event-stream") + h.WriteHeader(http.StatusOK) + h.since = time.Now().Add(-ago) + return h + } + // the reply begins at once; its thinking goes on past holdLongest + h := start(0) + h.Write([]byte(events[0] + events[1])) + h.since = time.Now().Add(-holdLongest - 10*time.Second) + for _, ev := range events[2:] { + h.Write([]byte(ev)) + } + h.settle() + if h.passing || !h.refused || !h.failed() { + t.Fatalf("thought 25s then refused: passing %v refused %v", h.passing, h.refused) + } + h = start(holdThinking + time.Second) + h.Write([]byte(events[0] + events[1] + events[2])) + if !h.passing { + t.Fatal("reasoning held past holdThinking") + } +} diff --git a/internal/gateway/responses.go b/internal/gateway/responses.go index 0bd03896..e8d6db67 100644 --- a/internal/gateway/responses.go +++ b/internal/gateway/responses.go @@ -568,7 +568,7 @@ func (d *responsesDecoder) decode(data string, emit func(Event)) error { } case "response.completed", "response.incomplete", "response.failed": if ev.Response.Error != nil { - emit(Event{Kind: KError, Text: ev.Response.Error.Message}) + emit(Event{Kind: KError, Text: ev.Response.Error.Message, Code: refusedCode(data)}) return nil } stop := "stop" @@ -594,7 +594,7 @@ func (d *responsesDecoder) decode(data string, emit func(Event)) error { if ev.Error != nil { msg = ev.Error.Message } - emit(Event{Kind: KError, Text: msg}) + emit(Event{Kind: KError, Text: msg, Code: refusedCode(data)}) } return nil } @@ -764,7 +764,11 @@ func (e *responsesEncoder) event(ev Event) { } case KError: e.closeItem() - e.send("response.failed", map[string]any{"response": e.response("failed", map[string]any{"error": map[string]any{"code": "server_error", "message": ev.Text}})}) + code := "server_error" + if ev.Code != "" { + code = ev.Code + } + e.send("response.failed", map[string]any{"response": e.response("failed", map[string]any{"error": map[string]any{"code": code, "message": ev.Text}})}) } e.col.add(ev) }