mirror of
https://github.com/lukaszraczylo/traefik-llmgateway.git
synced 2026-10-12 04:54:45 +00:00
fix(anthropic): merge parallel tool results, ordinal stream tool indices, array system content, strict stream decode
Review round 1 fixes (1 Critical + 5 Important) on the anthropic
adapter/translator.
- CRITICAL: consecutive role:"tool" messages each became their own
Anthropic user turn. Anthropic requires alternating user/assistant
turns and 400s on two consecutive same-role messages; parallel tool
calls (one assistant message with N tool_calls followed by N
role:"tool" messages) are the standard agentic transcript shape.
anthropicRequestFromOpenAI now accumulates consecutive tool messages
into one Anthropic user turn carrying all of their tool_result
blocks (anthropicToolResultBlock replaces anthropicToolResultMessage).
Added chat_parallel_tool_results fixture.
- Streaming tool_calls[].index passed Anthropic's raw content-block
index through, so a tool_use block after a text block emitted
index:1 instead of the required contiguous 0-based ordinal.
anthropicStreamState now tracks a per-stream tool-call ordinal
counter plus a content-block-index -> ordinal map, populated on
content_block_start and consulted on input_json_delta. Added the
stream_text_then_tool fixture (text block at index 0, tool_use at
index 1) proving the emitted index is 0.
- A system message with array-form content ([{"type":"text",...}])
was silently dropped (only a string content was handled). Added
systemTextFromContentParts: text parts join with "\n"; a non-text
part returns a *translateError instead of being dropped. Added
chat_system_array_content (success) and
chat_system_array_content_error fixtures.
- Malformed JSON on a recognized stream event (message_start,
content_block_start, content_block_delta, message_delta) was
swallowed as (nil, nil), which could silently corrupt a client's
reassembled tool-call arguments. All four now return an error
wrapping errUpstream; forwardStream already propagates that and
withholds the terminal [DONE]. The unrecognized-event-type and
error-event branches are unchanged. Added
TestAnthropicStreamState_MalformedEventReturnsError.
- providerAdapter.injectAuth's doc comment claimed a universal
empty-key no-op; updated to state the per-type divergence
(openai-type: keyless is a no-op; anthropic-type: the constructor
rejects an empty key, so injectAuth is never called with one).
- Added case-insensitive-Content-Type coverage for both adapters'
streaming-vs-JSON gate (previously untested): one subtest each in
provider_openai_test.go / provider_anthropic_test.go using a fake
upstream that sends "Text/Event-Stream; charset=utf-8", asserting
the streaming path (not the JSON fallback) is taken.
Gate: go test ./... -count=1 and -race both green; gofmt, go vet,
fieldalignment, golangci-lint (incl. gosec), govulncheck all clean.
This commit is contained in:
@@ -181,6 +181,52 @@ func TestAnthropicAdapter_ChatCompletion_Streaming(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestAnthropicAdapter_ChatCompletion_Streaming_CaseInsensitiveContentType
|
||||
// proves the streaming-vs-JSON-fallback gate matches an upstream's
|
||||
// Content-Type case-insensitively — review fix, mirroring the same fix
|
||||
// in the openai-type adapter. Distinguishing evidence: the streaming path
|
||||
// extracts real usage from the SSE events and sets Content-Type to the
|
||||
// sseWriter's own lowercase "text/event-stream"; the JSON fallback path
|
||||
// would copy the upstream's mixed-case header verbatim and fail to
|
||||
// decode a body of raw SSE text as a single Anthropic response.
|
||||
func TestAnthropicAdapter_ChatCompletion_Streaming_CaseInsensitiveContentType(t *testing.T) {
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "Text/Event-Stream; charset=utf-8")
|
||||
w.WriteHeader(http.StatusOK)
|
||||
f, ok := w.(http.Flusher)
|
||||
if !ok {
|
||||
t.Fatalf("httptest ResponseWriter does not implement http.Flusher")
|
||||
}
|
||||
for _, frame := range anthropicStreamFrames {
|
||||
fmt.Fprintf(w, "event: %s\ndata: %s\n\n", frame.event, frame.data)
|
||||
f.Flush()
|
||||
}
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
a, err := newAnthropicAdapter("p1", srv.URL, "sk-ant-test")
|
||||
if err != nil {
|
||||
t.Fatalf("newAnthropicAdapter: %v", err)
|
||||
}
|
||||
rec := newRecordingResponseWriter()
|
||||
|
||||
req := map[string]any{
|
||||
"model": "claude-opus-5",
|
||||
"stream": true,
|
||||
"messages": []any{map[string]any{"role": "user", "content": "hello"}},
|
||||
}
|
||||
u, err := a.chatCompletion(context.Background(), rec, req)
|
||||
if err != nil {
|
||||
t.Fatalf("chatCompletion: %v", err)
|
||||
}
|
||||
if u.prompt != 6 || u.completion != 3 {
|
||||
t.Errorf("usage = %+v, want {prompt:6 completion:3} (proves the streaming path, not JSON fallback, was taken)", u)
|
||||
}
|
||||
if got := rec.header.Get("Content-Type"); got != "text/event-stream" {
|
||||
t.Errorf("Content-Type on client response = %q, want %q (sseWriter's own header; JSON fallback would copy the upstream's mixed-case value verbatim)", got, "text/event-stream")
|
||||
}
|
||||
}
|
||||
|
||||
// TestAnthropicAdapter_ChatCompletion_Upstream429 proves a non-2xx
|
||||
// upstream response returns a *providerHTTPError with nothing written to
|
||||
// the client, mirroring the openai-type adapter's contract.
|
||||
|
||||
@@ -364,6 +364,49 @@ func TestOpenAIAdapter_ChatCompletion_Streaming_UpstreamIgnoresStreamFallsBackTo
|
||||
}
|
||||
}
|
||||
|
||||
// TestOpenAIAdapter_ChatCompletion_Streaming_CaseInsensitiveContentType
|
||||
// proves the streaming-vs-JSON-fallback gate matches an upstream's
|
||||
// Content-Type case-insensitively — review fix: Content-Type values are
|
||||
// case-insensitive per RFC 7231, but the gate previously did a
|
||||
// case-sensitive substring match against "text/event-stream" and would
|
||||
// wrongly fall back to the JSON path for an upstream that sent, e.g.,
|
||||
// "Text/Event-Stream". Distinguishing evidence: the streaming path
|
||||
// extracts real usage from the SSE chunks and sets Content-Type to the
|
||||
// sseWriter's own lowercase "text/event-stream"; the JSON fallback path
|
||||
// would copy the upstream's mixed-case header verbatim and fail to parse
|
||||
// usage out of raw SSE text.
|
||||
func TestOpenAIAdapter_ChatCompletion_Streaming_CaseInsensitiveContentType(t *testing.T) {
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "Text/Event-Stream; charset=utf-8")
|
||||
w.WriteHeader(http.StatusOK)
|
||||
f, ok := w.(http.Flusher)
|
||||
if !ok {
|
||||
t.Fatalf("httptest ResponseWriter does not implement http.Flusher")
|
||||
}
|
||||
for _, frame := range streamFrames {
|
||||
fmt.Fprintf(w, "data: %s\n\n", frame)
|
||||
f.Flush()
|
||||
}
|
||||
fmt.Fprint(w, "data: [DONE]\n\n")
|
||||
f.Flush()
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
a := newOpenAIAdapter("p1", srv.URL, "sk-test")
|
||||
rec := newRecordingResponseWriter()
|
||||
|
||||
u, err := a.chatCompletion(context.Background(), rec, map[string]any{"model": "gpt-5", "stream": true})
|
||||
if err != nil {
|
||||
t.Fatalf("chatCompletion: %v", err)
|
||||
}
|
||||
if u.prompt != 7 || u.completion != 9 {
|
||||
t.Errorf("usage = %+v, want {prompt:7 completion:9} (proves the streaming path, not JSON fallback, was taken)", u)
|
||||
}
|
||||
if got := rec.header.Get("Content-Type"); got != "text/event-stream" {
|
||||
t.Errorf("Content-Type on client response = %q, want %q (sseWriter's own header; JSON fallback would copy the upstream's mixed-case value verbatim)", got, "text/event-stream")
|
||||
}
|
||||
}
|
||||
|
||||
// TestOpenAIAdapter_ChatCompletion_Upstream429 proves a non-2xx upstream
|
||||
// response returns a *providerHTTPError carrying the status and body,
|
||||
// with nothing written to the client — the caller decides envelope vs
|
||||
|
||||
+6
-2
@@ -91,8 +91,12 @@ type providerAdapter interface {
|
||||
// trimmed.
|
||||
base() string
|
||||
// injectAuth sets whatever headers r needs to authenticate against the
|
||||
// provider. It is a no-op when the provider's resolved API key is
|
||||
// empty — a keyless upstream is a valid configuration, not an error.
|
||||
// provider. Behavior is type-specific: an openai-type adapter treats
|
||||
// a resolved-empty API key as a valid keyless upstream, so injectAuth
|
||||
// is a no-op in that case. An anthropic-type adapter's constructor
|
||||
// rejects an empty key outright (ruling c), so its injectAuth always
|
||||
// sets auth headers — there is no keyless case for it to be a no-op
|
||||
// over.
|
||||
injectAuth(r *http.Request)
|
||||
// chatCompletion proxies req to the provider's chat completions
|
||||
// endpoint and writes the (possibly streamed) response to w. req has
|
||||
|
||||
@@ -0,0 +1,13 @@
|
||||
{
|
||||
"model": "claude-opus-5",
|
||||
"messages": [
|
||||
{"role": "user", "content": "What's the weather in Paris and Tokyo?"},
|
||||
{"role": "assistant", "content": null, "tool_calls": [
|
||||
{"id": "call_1", "type": "function", "function": {"name": "get_weather", "arguments": "{\"location\":\"Paris\"}"}},
|
||||
{"id": "call_2", "type": "function", "function": {"name": "get_weather", "arguments": "{\"location\":\"Tokyo\"}"}}
|
||||
]},
|
||||
{"role": "tool", "tool_call_id": "call_1", "content": "72F and sunny"},
|
||||
{"role": "tool", "tool_call_id": "call_2", "content": "50F and rainy"}
|
||||
],
|
||||
"max_tokens": 200
|
||||
}
|
||||
@@ -0,0 +1,15 @@
|
||||
{
|
||||
"model": "claude-opus-5",
|
||||
"messages": [
|
||||
{"role": "user", "content": "What's the weather in Paris and Tokyo?"},
|
||||
{"role": "assistant", "content": [
|
||||
{"type": "tool_use", "id": "call_1", "name": "get_weather", "input": {"location": "Paris"}},
|
||||
{"type": "tool_use", "id": "call_2", "name": "get_weather", "input": {"location": "Tokyo"}}
|
||||
]},
|
||||
{"role": "user", "content": [
|
||||
{"type": "tool_result", "tool_use_id": "call_1", "content": "72F and sunny"},
|
||||
{"type": "tool_result", "tool_use_id": "call_2", "content": "50F and rainy"}
|
||||
]}
|
||||
],
|
||||
"max_tokens": 200
|
||||
}
|
||||
@@ -0,0 +1,11 @@
|
||||
{
|
||||
"model": "claude-opus-5",
|
||||
"messages": [
|
||||
{"role": "system", "content": [
|
||||
{"type": "text", "text": "You are a helpful assistant."},
|
||||
{"type": "image_url", "image_url": {"url": "data:image/png;base64,AAAA"}}
|
||||
]},
|
||||
{"role": "user", "content": "Hi"}
|
||||
],
|
||||
"max_tokens": 100
|
||||
}
|
||||
@@ -0,0 +1 @@
|
||||
{"errContains": "system message content part type \"image_url\" is not supported for anthropic models", "notSupported": false}
|
||||
@@ -0,0 +1,11 @@
|
||||
{
|
||||
"model": "claude-opus-5",
|
||||
"messages": [
|
||||
{"role": "system", "content": [
|
||||
{"type": "text", "text": "You are a helpful assistant."},
|
||||
{"type": "text", "text": "Always be concise."}
|
||||
]},
|
||||
{"role": "user", "content": "Hi"}
|
||||
],
|
||||
"max_tokens": 100
|
||||
}
|
||||
@@ -0,0 +1,8 @@
|
||||
{
|
||||
"model": "claude-opus-5",
|
||||
"system": "You are a helpful assistant.\nAlways be concise.",
|
||||
"messages": [
|
||||
{"role": "user", "content": "Hi"}
|
||||
],
|
||||
"max_tokens": 100
|
||||
}
|
||||
+27
@@ -0,0 +1,27 @@
|
||||
event: message_start
|
||||
data: {"type":"message_start","message":{"id":"msg_01MIX","type":"message","role":"assistant","content":[],"model":"claude-opus-5","stop_reason":null,"stop_sequence":null,"usage":{"input_tokens":8,"output_tokens":1}}}
|
||||
|
||||
event: content_block_start
|
||||
data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}
|
||||
|
||||
event: content_block_delta
|
||||
data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Let me check."}}
|
||||
|
||||
event: content_block_stop
|
||||
data: {"type":"content_block_stop","index":0}
|
||||
|
||||
event: content_block_start
|
||||
data: {"type":"content_block_start","index":1,"content_block":{"type":"tool_use","id":"toolu_MIX1","name":"get_weather","input":{}}}
|
||||
|
||||
event: content_block_delta
|
||||
data: {"type":"content_block_delta","index":1,"delta":{"type":"input_json_delta","partial_json":"{\"location\":\"Paris\"}"}}
|
||||
|
||||
event: content_block_stop
|
||||
data: {"type":"content_block_stop","index":1}
|
||||
|
||||
event: message_delta
|
||||
data: {"type":"message_delta","delta":{"stop_reason":"tool_use","stop_sequence":null},"usage":{"output_tokens":6}}
|
||||
|
||||
event: message_stop
|
||||
data: {"type":"message_stop"}
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
{"prompt": 8, "completion": 6}
|
||||
@@ -0,0 +1,7 @@
|
||||
[
|
||||
{"id":"chatcmpl-msg_01MIX","object":"chat.completion.chunk","created":1734000000,"model":"gw-anthropic-model","choices":[{"index":0,"delta":{"role":"assistant"},"finish_reason":null}]},
|
||||
{"id":"chatcmpl-msg_01MIX","object":"chat.completion.chunk","created":1734000000,"model":"gw-anthropic-model","choices":[{"index":0,"delta":{"content":"Let me check."},"finish_reason":null}]},
|
||||
{"id":"chatcmpl-msg_01MIX","object":"chat.completion.chunk","created":1734000000,"model":"gw-anthropic-model","choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"id":"toolu_MIX1","type":"function","function":{"name":"get_weather","arguments":""}}]},"finish_reason":null}]},
|
||||
{"id":"chatcmpl-msg_01MIX","object":"chat.completion.chunk","created":1734000000,"model":"gw-anthropic-model","choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"function":{"arguments":"{\"location\":\"Paris\"}"}}]},"finish_reason":null}]},
|
||||
{"id":"chatcmpl-msg_01MIX","object":"chat.completion.chunk","created":1734000000,"model":"gw-anthropic-model","choices":[{"index":0,"delta":{},"finish_reason":"tool_calls"}]}
|
||||
]
|
||||
+103
-31
@@ -218,24 +218,45 @@ func anthropicMessageFromOpenAI(msg map[string]any, role string) (map[string]any
|
||||
return map[string]any{"role": role, "content": blocks}, nil
|
||||
}
|
||||
|
||||
// anthropicToolResultMessage maps one OpenAI role:"tool" message to an
|
||||
// Anthropic user message carrying a single tool_result content block, per
|
||||
// the mapping table's "messages[role=tool]" row. content passes through
|
||||
// unchanged — Anthropic's tool_result.content accepts a string, matching
|
||||
// the common case, and this translator does not need to further interpret
|
||||
// it.
|
||||
func anthropicToolResultMessage(msg map[string]any) map[string]any {
|
||||
// anthropicToolResultBlock maps one OpenAI role:"tool" message to a single
|
||||
// Anthropic {"type":"tool_result","tool_use_id","content"} content block,
|
||||
// per the mapping table's "messages[role=tool]" row. content passes
|
||||
// through unchanged — Anthropic's tool_result.content accepts a string,
|
||||
// matching the common case, and this translator does not need to further
|
||||
// interpret it. The caller (anthropicRequestFromOpenAI) accumulates
|
||||
// consecutive tool messages' blocks into one Anthropic user turn — see
|
||||
// its doc comment.
|
||||
func anthropicToolResultBlock(msg map[string]any) map[string]any {
|
||||
toolCallID, _ := msg["tool_call_id"].(string)
|
||||
return map[string]any{
|
||||
"role": "user",
|
||||
"content": []any{map[string]any{
|
||||
"type": "tool_result",
|
||||
"tool_use_id": toolCallID,
|
||||
"content": msg["content"],
|
||||
}},
|
||||
"type": "tool_result",
|
||||
"tool_use_id": toolCallID,
|
||||
"content": msg["content"],
|
||||
}
|
||||
}
|
||||
|
||||
// systemTextFromContentParts joins an OpenAI system message's array-form
|
||||
// content ([{"type":"text","text":...}, ...]) into plain text, joined
|
||||
// with "\n". Any part that is not {"type":"text"} has no equivalent in
|
||||
// Anthropic's plain-string top-level "system" field, so it is a
|
||||
// *translateError rather than being silently dropped.
|
||||
func systemTextFromContentParts(parts []any) (string, error) {
|
||||
texts := make([]string, 0, len(parts))
|
||||
for _, part := range parts {
|
||||
pm, ok := part.(map[string]any)
|
||||
if !ok {
|
||||
return "", &translateError{msg: "system message content[] entries must be objects"}
|
||||
}
|
||||
ptype, _ := pm["type"].(string)
|
||||
if ptype != "text" {
|
||||
return "", &translateError{msg: fmt.Sprintf("system message content part type %q is not supported for anthropic models", ptype)}
|
||||
}
|
||||
text, _ := pm["text"].(string)
|
||||
texts = append(texts, text)
|
||||
}
|
||||
return strings.Join(texts, "\n"), nil
|
||||
}
|
||||
|
||||
// anthropicToolChoiceFromOpenAI maps OpenAI's "tool_choice" to Anthropic's
|
||||
// tool_choice object, and reports whether "tools" must be dropped from the
|
||||
// request. Per the mapping table, tool_choice:"none" has no valid
|
||||
@@ -291,7 +312,9 @@ func anthropicToolsFromOpenAI(toolsRaw []any) []any {
|
||||
// mapping table. It returns a *translateError — never wrapped — for a
|
||||
// field Anthropic has no equivalent for (n>1, logit_bias, logprobs) or a
|
||||
// content part this translator cannot map (a non-data-URI image URL, a
|
||||
// malformed tool_calls argument string).
|
||||
// malformed tool_calls argument string). Consecutive role:"tool" messages
|
||||
// are merged into one Anthropic user turn carrying all of their
|
||||
// tool_result blocks — see anthropicToolResultBlock's doc comment.
|
||||
func anthropicRequestFromOpenAI(req map[string]any) (map[string]any, error) {
|
||||
if v, ok := req["n"]; ok {
|
||||
if n, ok2 := toFloat64(v); ok2 && n > 1 {
|
||||
@@ -313,6 +336,21 @@ func anthropicRequestFromOpenAI(req map[string]any) (map[string]any, error) {
|
||||
msgsRaw, _ := req["messages"].([]any)
|
||||
var systemParts []string
|
||||
anthMsgs := make([]any, 0, len(msgsRaw))
|
||||
// pendingToolResults accumulates consecutive role:"tool" messages'
|
||||
// blocks. Anthropic requires alternating user/assistant turns and
|
||||
// 400s on two consecutive same-role messages, and parallel tool calls
|
||||
// (one assistant message with N tool_calls, followed by N role:"tool"
|
||||
// messages) are the standard agentic transcript shape — so those N
|
||||
// messages must collapse into one Anthropic user turn carrying N
|
||||
// tool_result blocks, not N separate user turns.
|
||||
var pendingToolResults []any
|
||||
flushPendingToolResults := func() {
|
||||
if len(pendingToolResults) == 0 {
|
||||
return
|
||||
}
|
||||
anthMsgs = append(anthMsgs, map[string]any{"role": "user", "content": pendingToolResults})
|
||||
pendingToolResults = nil
|
||||
}
|
||||
for _, m := range msgsRaw {
|
||||
msg, ok := m.(map[string]any)
|
||||
if !ok {
|
||||
@@ -320,19 +358,28 @@ func anthropicRequestFromOpenAI(req map[string]any) (map[string]any, error) {
|
||||
}
|
||||
switch role, _ := msg["role"].(string); role {
|
||||
case "system":
|
||||
if s, ok := msg["content"].(string); ok {
|
||||
systemParts = append(systemParts, s)
|
||||
switch c := msg["content"].(type) {
|
||||
case string:
|
||||
systemParts = append(systemParts, c)
|
||||
case []any:
|
||||
text, err := systemTextFromContentParts(c)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
systemParts = append(systemParts, text)
|
||||
}
|
||||
case "user", "assistant":
|
||||
flushPendingToolResults()
|
||||
am, err := anthropicMessageFromOpenAI(msg, role)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
anthMsgs = append(anthMsgs, am)
|
||||
case "tool":
|
||||
anthMsgs = append(anthMsgs, anthropicToolResultMessage(msg))
|
||||
pendingToolResults = append(pendingToolResults, anthropicToolResultBlock(msg))
|
||||
}
|
||||
}
|
||||
flushPendingToolResults()
|
||||
if len(systemParts) > 0 {
|
||||
out["system"] = strings.Join(systemParts, "\n\n")
|
||||
}
|
||||
@@ -519,10 +566,19 @@ func reMarshalToolInput(raw json.RawMessage) (string, error) {
|
||||
// running usage total. A caller constructs one per stream and calls
|
||||
// translate once per upstream sseEvent, in order.
|
||||
type anthropicStreamState struct {
|
||||
id string
|
||||
model string
|
||||
created int64
|
||||
u usage
|
||||
// toolOrdinals maps an Anthropic content-block index to the 0-based
|
||||
// ordinal of that tool call among all tool_use blocks seen so far in
|
||||
// this message. OpenAI's tool_calls[].index must be contiguous
|
||||
// starting at 0 over the tool-call array; Anthropic's content-block
|
||||
// index also counts any preceding non-tool_use blocks (a tool_use
|
||||
// block right after a text block is content-block index 1, but it is
|
||||
// still the first — index 0 — tool call).
|
||||
toolOrdinals map[int]int
|
||||
id string
|
||||
model string
|
||||
u usage
|
||||
created int64
|
||||
nextToolOrdinal int
|
||||
}
|
||||
|
||||
// newAnthropicStreamState returns an anthropicStreamState for one
|
||||
@@ -530,7 +586,7 @@ type anthropicStreamState struct {
|
||||
// created the response timestamp every emitted chunk carries — the
|
||||
// adapter fixes both once, before the first upstream event arrives.
|
||||
func newAnthropicStreamState(model string, created int64) *anthropicStreamState {
|
||||
return &anthropicStreamState{model: model, created: created}
|
||||
return &anthropicStreamState{model: model, created: created, toolOrdinals: map[int]int{}}
|
||||
}
|
||||
|
||||
// usage returns the prompt/completion token counts captured so far, from
|
||||
@@ -636,15 +692,18 @@ type anthropicErrorEventPayload struct {
|
||||
// error — translate's caller must stop reading the stream after this
|
||||
// return, but forwards the returned chunk to the client first.
|
||||
//
|
||||
// Any other or malformed event is ignored (nil, nil) rather than erroring
|
||||
// — an event this translator does not recognize is not necessarily a
|
||||
// broken stream.
|
||||
// An unrecognized event type is ignored (nil, nil) — this translator does
|
||||
// not treat an event it does not know about as a broken stream. Malformed
|
||||
// data on a recognized event type is different: it returns an error
|
||||
// wrapping errUpstream instead of silently producing nothing, since a
|
||||
// dropped input_json_delta (say) would otherwise corrupt the client's
|
||||
// reassembled tool-call arguments without either side noticing.
|
||||
func (st *anthropicStreamState) translate(ev sseEvent) ([][]byte, error) {
|
||||
switch ev.event {
|
||||
case "message_start":
|
||||
var p anthropicMessageStartPayload
|
||||
if err := json.Unmarshal(ev.data, &p); err != nil {
|
||||
return nil, nil
|
||||
return nil, fmt.Errorf("%w: decode anthropic message_start event: %w", errUpstream, err)
|
||||
}
|
||||
st.id = p.Message.ID
|
||||
st.u.prompt = p.Message.Usage.InputTokens
|
||||
@@ -653,13 +712,16 @@ func (st *anthropicStreamState) translate(ev sseEvent) ([][]byte, error) {
|
||||
case "content_block_start":
|
||||
var p anthropicContentBlockStartPayload
|
||||
if err := json.Unmarshal(ev.data, &p); err != nil {
|
||||
return nil, nil
|
||||
return nil, fmt.Errorf("%w: decode anthropic content_block_start event: %w", errUpstream, err)
|
||||
}
|
||||
if p.ContentBlock.Type != "tool_use" {
|
||||
return nil, nil
|
||||
}
|
||||
ordinal := st.nextToolOrdinal
|
||||
st.toolOrdinals[p.Index] = ordinal
|
||||
st.nextToolOrdinal++
|
||||
delta := map[string]any{"tool_calls": []any{map[string]any{
|
||||
"index": p.Index,
|
||||
"index": ordinal,
|
||||
"id": p.ContentBlock.ID,
|
||||
"type": "function",
|
||||
"function": map[string]any{
|
||||
@@ -672,14 +734,24 @@ func (st *anthropicStreamState) translate(ev sseEvent) ([][]byte, error) {
|
||||
case "content_block_delta":
|
||||
var p anthropicContentBlockDeltaPayload
|
||||
if err := json.Unmarshal(ev.data, &p); err != nil {
|
||||
return nil, nil
|
||||
return nil, fmt.Errorf("%w: decode anthropic content_block_delta event: %w", errUpstream, err)
|
||||
}
|
||||
switch p.Delta.Type {
|
||||
case "text_delta":
|
||||
return [][]byte{st.chunk(map[string]any{"content": p.Delta.Text}, nil)}, nil
|
||||
case "input_json_delta":
|
||||
ordinal, ok := st.toolOrdinals[p.Index]
|
||||
if !ok {
|
||||
// Defensive: an input_json_delta for a content-block index
|
||||
// this translator never saw a tool_use content_block_start
|
||||
// for. Fall back to the raw content-block index rather than
|
||||
// dropping the fragment — a wrong index is recoverable by
|
||||
// the client's own reassembly; a silently dropped argument
|
||||
// fragment is not.
|
||||
ordinal = p.Index
|
||||
}
|
||||
delta := map[string]any{"tool_calls": []any{map[string]any{
|
||||
"index": p.Index,
|
||||
"index": ordinal,
|
||||
"function": map[string]any{"arguments": p.Delta.PartialJSON},
|
||||
}}}
|
||||
return [][]byte{st.chunk(delta, nil)}, nil
|
||||
@@ -690,7 +762,7 @@ func (st *anthropicStreamState) translate(ev sseEvent) ([][]byte, error) {
|
||||
case "message_delta":
|
||||
var p anthropicMessageDeltaPayload
|
||||
if err := json.Unmarshal(ev.data, &p); err != nil {
|
||||
return nil, nil
|
||||
return nil, fmt.Errorf("%w: decode anthropic message_delta event: %w", errUpstream, err)
|
||||
}
|
||||
st.u.completion = p.Usage.OutputTokens
|
||||
reason := anthropicFinishReason(p.Delta.StopReason)
|
||||
|
||||
@@ -270,6 +270,16 @@ func TestAnthropicStreamGolden_ToolUse(t *testing.T) {
|
||||
runStreamGolden(t, "stream_tooluse")
|
||||
}
|
||||
|
||||
// TestAnthropicStreamGolden_TextThenTool covers a text block followed by
|
||||
// a tool_use block (Anthropic content-block index 0 then 1) and proves
|
||||
// the emitted tool_calls[0].index is the 0-based tool-call ordinal (0),
|
||||
// not the raw Anthropic content-block index (1) — review fix: OpenAI's
|
||||
// contract requires tool_calls[].index to be contiguous from 0 over the
|
||||
// tool-call array, regardless of how many non-tool_use blocks preceded it.
|
||||
func TestAnthropicStreamGolden_TextThenTool(t *testing.T) {
|
||||
runStreamGolden(t, "stream_text_then_tool")
|
||||
}
|
||||
|
||||
// TestAnthropicStreamState_ErrorEvent proves an "error" SSE event
|
||||
// translates to one OpenAI-shaped {"error":{...}} chunk and a non-nil
|
||||
// terminal error, per the stream mapping table's error row — built
|
||||
@@ -307,6 +317,33 @@ func TestAnthropicStreamState_ErrorEvent(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestAnthropicStreamState_MalformedEventReturnsError proves a malformed
|
||||
// (invalid-JSON) data payload on each recognized event type returns an
|
||||
// error wrapping errUpstream, rather than being swallowed as (nil, nil) —
|
||||
// review fix: a dropped input_json_delta would otherwise silently corrupt
|
||||
// the client's reassembled tool-call arguments. The "error" event type
|
||||
// itself already returned an error before this fix and is covered by
|
||||
// TestAnthropicStreamState_ErrorEvent, not repeated here.
|
||||
func TestAnthropicStreamState_MalformedEventReturnsError(t *testing.T) {
|
||||
const corrupt = `{"type":"message_start","message":{`
|
||||
|
||||
for _, event := range []string{"message_start", "content_block_start", "content_block_delta", "message_delta"} {
|
||||
t.Run(event, func(t *testing.T) {
|
||||
st := newAnthropicStreamState(goldenGatewayModel, goldenCreated)
|
||||
chunks, err := st.translate(sseEvent{event: event, data: []byte(corrupt)})
|
||||
if err == nil {
|
||||
t.Fatalf("translate(%s, corrupt data): want error, got nil (chunks=%v)", event, chunks)
|
||||
}
|
||||
if !errors.Is(err, errUpstream) {
|
||||
t.Errorf("translate(%s, corrupt data): err = %v, want errors.Is(err, errUpstream)", event, err)
|
||||
}
|
||||
if len(chunks) != 0 {
|
||||
t.Errorf("translate(%s, corrupt data): chunks = %v, want none", event, chunks)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestAnthropicListModels_404NotSpeciallyHandled proves listModels
|
||||
// returns a plain *providerHTTPError on a 404 (an older Anthropic API
|
||||
// version that predates the Models endpoint) rather than any special
|
||||
|
||||
Reference in New Issue
Block a user