Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
115 changes: 111 additions & 4 deletions internal/app/anthropic_upstream.go
Original file line number Diff line number Diff line change
Expand Up @@ -137,8 +137,15 @@ func pipeAnthropicStream(ctx context.Context, w http.ResponseWriter, rc io.Reade
// observeAnthropicStreamEvent 让 stats 与 message_start/stop 计数正确
// 累计——peek 消费过的帧不再二次进 reader.Read() 通道,所以这里必须补
// 一次观察。
if err := FlushPeekedBytes(w, peek.Consumed); err != nil {
return true, err
// peek 消费的首帧也要过大小写还原(帧尾常与首个 content_block_start 同行,
// 而主循环的还原按行生效),否则首个 tool_use 的 name 会被原样透小写。
for _, res := range peek.Consumed {
if res.line != "" {
res.line = restoreAnthropicStreamLineCase(res.line)
}
if _, err := io.WriteString(w, res.line); err != nil {
return true, err
}
}
sawMessageStop := false
observeLine := func(line string) {
Expand Down Expand Up @@ -202,7 +209,10 @@ func pipeAnthropicStream(ctx context.Context, w http.ResponseWriter, rc io.Reade
pendingErr := result.err
if line != "" {
observeLine(line)
frameBuf.WriteString(line)
// 免费层占位工具名大小写还原:仅改写恰为小写占位名的
// tool_use name 帧,其余字节原样,保证客户端(大小写敏感)
// 不再把 stub 的 read/glob/bash/grep 当成未注册工具。
frameBuf.WriteString(restoreAnthropicStreamLineCase(line))
// 空行 = 帧边界:整帧一次写出再 Flush。
if strings.TrimRight(line, "\r\n") == "" {
if err := flushFrame(); err != nil {
Expand Down Expand Up @@ -239,6 +249,62 @@ func pipeAnthropicStream(ctx context.Context, w http.ResponseWriter, rc io.Reade
}
}

// restoreAnthropicStreamLineCase 对一行 SSE(含结尾 \n)做免费层占位工具的大小写
// 还原:仅当它是 "data: " 帧、事件为 content_block_start / content_block_delta 且
// 携带恰为小写占位名的 tool_use name 时改写该行,其余行(含 content_block_stop、
// input_json_delta 的 partial_json 文本、非 data 行)原样返回。改写只命中 keep 的
// 名字段,不动其它字节(换行风格、字段顺序保持上游原样),供 byte-relay 路径在
// 写给客户端前调用。幂等。
func restoreAnthropicStreamLineCase(line string) string {
payload, ok := strings.CutPrefix(line, "data: ")
if !ok {
return line
}
trimmed := strings.TrimRight(payload, "\r\n")
if !strings.HasPrefix(trimmed, "{") {
return line
}
var evt map[string]any
if json.Unmarshal([]byte(trimmed), &evt) != nil {
return line
}
changed := false
if cb, ok := evt["content_block"].(map[string]any); ok {
if typ, _ := cb["type"].(string); typ == "tool_use" {
if n, _ := cb["name"].(string); n != "" {
if r := restoreToolNameCase(n); r != n {
cb["name"] = r
changed = true
}
}
}
}
// 兜底:某些上游把起始块放在 event.delta.content_block 而非顶层
// content_block;同样覆盖 delta 里偶发出现的 tool_use name。
if !changed {
if delta, ok := evt["delta"].(map[string]any); ok {
if cb, ok := delta["content_block"].(map[string]any); ok {
if typ, _ := cb["type"].(string); typ == "tool_use" {
if n, _ := cb["name"].(string); n != "" {
if r := restoreToolNameCase(n); r != n {
cb["name"] = r
changed = true
}
}
}
}
}
}
if !changed {
return line
}
b, err := json.Marshal(evt)
if err != nil {
return line
}
return "data: " + string(b) + "\n"
}

// observeAnthropicStreamEvent 旁路解析一行 SSE,累计 usage 与流统计。
func observeAnthropicStreamEvent(stats *logging.StreamStats, fullUsage map[string]any, line string) {
payload, ok := strings.CutPrefix(line, "data: ")
Expand Down Expand Up @@ -271,6 +337,45 @@ func observeAnthropicStreamEvent(stats *logging.StreamStats, fullUsage map[strin
}
}

// restoreAnthropicBodyToolCase 对一个完整的 Anthropic Messages JSON body 做免费层
// 占位工具名的大小写还原:遍历 content 数组,把恰为小写占位名(bash/glob/grep/
// read)的 tool_use block 的 name 还原为规范 PascalCase,assistant / tool_result /
// 其它块与所有其它字节原样。无法解析或无命中时原样返回(幂等)。供非流式
// byte-relay 在写给客户端前调用。
func restoreAnthropicBodyToolCase(body []byte) []byte {
var m map[string]any
if json.Unmarshal(body, &m) != nil {
return body
}
content, ok := m["content"].([]any)
if !ok {
return body
}
changed := false
for _, c := range content {
block, ok := c.(map[string]any)
if !ok {
continue
}
if typ, _ := block["type"].(string); typ != "tool_use" {
continue
}
if n, _ := block["name"].(string); n != "" {
if r := restoreToolNameCase(n); r != n {
block["name"] = r
changed = true
}
}
}
if !changed {
return body
}
if b, err := json.Marshal(m); err == nil {
return b
}
return body
}

// relayAnthropicBuffered 非流式直通与直通路径错误透传(含流式请求下的上游
// 非 2xx):buffered 读回上游体,以 application/json + 原状态码保真写回,
// 并解析 usage 记入 token 统计;上游错误体同时记入去重日志。
Expand All @@ -291,7 +396,9 @@ func relayAnthropicBuffered(ctx context.Context, w http.ResponseWriter, rc io.Re
}
w.Header().Set("Content-Type", "application/json; charset=utf-8")
w.WriteHeader(status)
w.Write(body)
// 非流式直通路径在写回前还原免费层占位工具的小写 tool_use name,
// 与流式 byte-relay 的 restoreAnthropicStreamLineCase 对应(幂等)。
w.Write(restoreAnthropicBodyToolCase(body))

if status >= 200 && status < 300 {
var raw map[string]any
Expand Down
11 changes: 7 additions & 4 deletions internal/app/chat_protocol_routing_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -968,10 +968,13 @@ func TestDispatch_ChatToResponsesMemory(t *testing.T) {
}

func TestResponsesSSEToChatStream_ToolArgumentsShareIndex(t *testing.T) {
// 工具名用中性的 Bash(非小写占位名):本用例专测「同 output_index 的参数
// 共享」,与工具名大小写无关——小写 "bash" 会被免费层占位大小写还原为
// "Bash"(见 restoreToolNameCase),不应作为本用例的夹具值。
sse := "event: response.created\ndata: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_7\",\"model\":\"gpt-x\"}}\n\n" +
"event: response.output_item.added\ndata: {\"type\":\"response.output_item.added\",\"output_index\":1,\"item\":{\"type\":\"function_call\",\"id\":\"fc_a\",\"call_id\":\"call_a\",\"name\":\"bash\"}}\n\n" +
"event: response.output_item.added\ndata: {\"type\":\"response.output_item.added\",\"output_index\":1,\"item\":{\"type\":\"function_call\",\"id\":\"fc_a\",\"call_id\":\"call_a\",\"name\":\"Bash\"}}\n\n" +
"event: response.function_call_arguments.delta\ndata: {\"type\":\"response.function_call_arguments.delta\",\"output_index\":1,\"item_id\":\"fc_a\",\"delta\":\"{\\\"cmd\\\":\\\"free -h\\\"}\"}\n\n" +
"event: response.output_item.added\ndata: {\"type\":\"response.output_item.added\",\"output_index\":2,\"item\":{\"type\":\"function_call\",\"id\":\"fc_b\",\"call_id\":\"call_b\",\"name\":\"bash\"}}\n\n" +
"event: response.output_item.added\ndata: {\"type\":\"response.output_item.added\",\"output_index\":2,\"item\":{\"type\":\"function_call\",\"id\":\"fc_b\",\"call_id\":\"call_b\",\"name\":\"Bash\"}}\n\n" +
// no item_id here: matched by output_index
"event: response.function_call_arguments.delta\ndata: {\"type\":\"response.function_call_arguments.delta\",\"output_index\":2,\"delta\":\"{\\\"cmd\\\":\\\"uptime\\\"}\"}\n\n" +
"event: response.completed\ndata: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_7\"}}\n\n"
Expand Down Expand Up @@ -1017,10 +1020,10 @@ func TestResponsesSSEToChatStream_ToolArgumentsShareIndex(t *testing.T) {
if len(names) != 2 {
t.Fatalf("want 2 tool calls, got indices %v (arguments %v): %s", names, args, body)
}
if names[0] != "bash" || args[0] != `{"cmd":"free -h"}` {
if names[0] != "Bash" || args[0] != `{"cmd":"free -h"}` {
t.Fatalf("tool call 0 = %q %q", names[0], args[0])
}
if names[1] != "bash" || args[1] != `{"cmd":"uptime"}` {
if names[1] != "Bash" || args[1] != `{"cmd":"uptime"}` {
t.Fatalf("tool call 1 = %q %q", names[1], args[1])
}
}
Expand Down
8 changes: 6 additions & 2 deletions internal/app/chat_to_anthropic.go
Original file line number Diff line number Diff line change
Expand Up @@ -194,7 +194,7 @@ func chatMessagesToAnthropic(messages []Message) (string, []map[string]any) {
for _, tc := range msg.ToolCalls {
input := parseToolCallArguments(tc.Function.Arguments)
blocks = append(blocks, map[string]any{
"type": "tool_use", "id": tc.ID, "name": tc.Function.Name, "input": input,
"type": "tool_use", "id": tc.ID, "name": restoreToolNameCase(tc.Function.Name), "input": input,
})
}
appendBlocks("assistant", blocks)
Expand Down Expand Up @@ -896,7 +896,11 @@ func (st *anthropicToChatState) handleLine(line string) {
st.toolIndices[idx] = toolIdx
tool := &anthropicToolState{}
st.toolStates[idx] = tool
name, _ := cb["name"].(string)
// 免费层小写占位工具名在此还原为客户端注册的规范大小写
// (bash/glob/grep/read -> Bash/Glob/Grep/Read),避免 Claude
// Code 等大小写敏感客户端报 "No such tool available"。
rawName, _ := cb["name"].(string)
name := restoreToolNameCase(rawName)
id, _ := cb["id"].(string)
// 缓存 start 块的 initial input(常见 {});不要立刻 emit 给 chat 端
// —— OpenAI 客户端会 concat 所有 arguments 片段,若 start 下发了
Expand Down
6 changes: 3 additions & 3 deletions internal/app/chat_to_responses_upstream.go
Original file line number Diff line number Diff line change
Expand Up @@ -473,7 +473,7 @@ func aggregateResponsesStreamToChat(body []byte, model string, wantReasoning boo
acc.callID = callID
}
if n, _ := item["name"].(string); n != "" {
acc.name = n
acc.name = restoreToolNameCase(n)
}
if args, _ := item["arguments"].(string); args != "" && acc.args == "" {
acc.args = args
Expand Down Expand Up @@ -1113,7 +1113,7 @@ func (st *responsesToChatState) handleLine(line string) {
st.ensureRole()
st.emitChunk(map[string]any{"tool_calls": []any{map[string]any{
"index": toolIdx, "id": callID, "type": "function",
"function": map[string]any{"name": toString(item["name"]), "arguments": ""},
"function": map[string]any{"name": restoreToolNameCase(toString(item["name"])), "arguments": ""},
}}}, "", nil)
}
case "response.function_call_arguments.delta", "response.tool_call_arguments.delta":
Expand Down Expand Up @@ -1155,7 +1155,7 @@ func (st *responsesToChatState) handleLine(line string) {
// 并把 item 上的完整 arguments 作为单段增量发完。
if !st.toolAnnounced[toolIdx] {
st.toolAnnounced[toolIdx] = true
name := toString(item["name"])
name := restoreToolNameCase(toString(item["name"]))
st.ensureRole()
st.emitChunk(map[string]any{"tool_calls": []any{map[string]any{
"index": toolIdx, "id": callID, "type": "function",
Expand Down
11 changes: 8 additions & 3 deletions internal/app/claude.go
Original file line number Diff line number Diff line change
Expand Up @@ -740,7 +740,9 @@ func openAIToClaudeResponse(chatBody []byte, model string, wantReasoning bool) [
}
case "tool_use":
id, _ := blk["id"].(string)
name, _ := blk["name"].(string)
rawName, _ := blk["name"].(string)
// 免费层小写占位工具名还原为规范大小写(见 restoreToolNameCase)。
name := restoreToolNameCase(rawName)
input := blk["input"]
if input == nil {
input = map[string]any{}
Expand Down Expand Up @@ -783,7 +785,7 @@ func openAIToClaudeResponse(chatBody []byte, model string, wantReasoning bool) [
content = append(content, ClaudeContent{
Type: "tool_use",
ID: tc.ID,
Name: tc.Function.Name,
Name: restoreToolNameCase(tc.Function.Name),
Input: input,
})
}
Expand Down Expand Up @@ -1496,7 +1498,10 @@ loop:
callID = "toolu_" + randomString(12)
}
fn, _ := tc["function"].(map[string]any)
name, _ := fn["name"].(string)
rawName, _ := fn["name"].(string)
// 免费层小写占位工具名还原为规范大小写
// (见 restoreToolNameCase)。
name := restoreToolNameCase(rawName)
toolCallAccumulator[upstreamIndex] = map[string]string{
"id": callID,
"name": name,
Expand Down
9 changes: 6 additions & 3 deletions internal/app/claude_responses.go
Original file line number Diff line number Diff line change
Expand Up @@ -731,7 +731,9 @@ func responsesOutputToClaudeBlocks(output []any, wantReasoning bool) ([]ClaudeCo
if callID == "" {
callID, _ = item["id"].(string)
}
name, _ := item["name"].(string)
rawName, _ := item["name"].(string)
// 免费层小写占位工具名还原为规范大小写(见 restoreToolNameCase)。
name := restoreToolNameCase(rawName)
if name == "" {
continue
}
Expand Down Expand Up @@ -829,7 +831,8 @@ func responsesOutputToClaudeBlocks(output []any, wantReasoning bool) ([]ClaudeCo
content = append(content, ClaudeContent{Type: "text", Text: joinedText})
}
for _, tl := range tools {
content = append(content, ClaudeContent{Type: "tool_use", ID: tl.id, Name: tl.name, Input: tl.input})
// 免费层小写占位工具名还原为规范大小写(见 restoreToolNameCase)。
content = append(content, ClaudeContent{Type: "tool_use", ID: tl.id, Name: restoreToolNameCase(tl.name), Input: tl.input})
}
if len(content) == 0 {
content = append(content, ClaudeContent{Type: "text", Text: ""})
Expand Down Expand Up @@ -1284,7 +1287,7 @@ func claudeResponsesStreamHandler(ctx context.Context, w http.ResponseWriter, rc
emitEvent("content_block_start", map[string]any{
"type": "content_block_start", "index": b.claudeIndex,
"content_block": map[string]any{
"type": "tool_use", "id": b.toolID, "name": b.toolName, "input": map[string]any{},
"type": "tool_use", "id": b.toolID, "name": restoreToolNameCase(b.toolName), "input": map[string]any{},
},
})
if _, exists := indexOfToolOrder(toolOrder, b.claudeIndex); !exists {
Expand Down
Loading
Loading