diff --git a/internal/gateway/anthropic.go b/internal/gateway/anthropic.go index 3e7c4821..5693f137 100644 --- a/internal/gateway/anthropic.go +++ b/internal/gateway/anthropic.go @@ -507,30 +507,60 @@ func buildAnthropic(r *Request, model string) []byte { } // anthropicDecoder leaves out the blocks of Anthropic's own server tools — -// a web search it ran — whose input is no call of the client's. -type anthropicDecoder struct{ server map[int]bool } +// a web search it ran — whose input is no call of the client's. A +// tool_use block whose start carries its whole input, as a relay in front +// of another vendor's model may send it with no input_json_delta after +// (蓝猫 on Discord), keeps that input; Anthropic's own start has an empty +// one, its deltas after. +type anthropicDecoder struct { + server map[int]bool + whole map[int]bool // the tool_use blocks whose start carried their input +} func (d *anthropicDecoder) decode(data string, emit func(Event)) error { var ev struct { Type string `json:"type"` Index int `json:"index"` ContentBlock struct { - Type string `json:"type"` + Type string `json:"type"` + Input json.RawMessage `json:"input"` } `json:"content_block"` + Delta struct { + Type string `json:"type"` + } `json:"delta"` } if json.Unmarshal([]byte(data), &ev) == nil { switch ev.Type { case "content_block_start": d.server[ev.Index] = ev.ContentBlock.Type == "server_tool_use" + d.whole[ev.Index] = false + if ev.ContentBlock.Type == "tool_use" && fullInput(ev.ContentBlock.Input) { + if err := decodeAnthropic(data, emit); err != nil { + return err + } + d.whole[ev.Index] = true + emit(Event{Kind: KToolArgs, Text: string(ev.ContentBlock.Input)}) + return nil + } case "content_block_delta", "content_block_stop": if d.server[ev.Index] { return nil } + if d.whole[ev.Index] && ev.Delta.Type == "input_json_delta" { + return nil // the input came whole already + } } } return decodeAnthropic(data, emit) } +// fullInput says whether a tool_use start's input is the call's own: an +// object with something in it. +func fullInput(in json.RawMessage) bool { + var m map[string]any + return json.Unmarshal(in, &m) == nil && len(m) > 0 +} + // decodeAnthropic turns an Anthropic event stream into events. func decodeAnthropic(data string, emit func(Event)) error { var ev struct { @@ -639,9 +669,14 @@ func stopFromAnthropic(s string) string { // the second is Claude 4.5+ running into its context window // before max_tokens: the reply is cut short all the same return "length" - case "tool_use": + case "tool_use", + // an Anthropic-shaped relay in front of another vendor's model, + // passing on OpenAI's reasons as they came (蓝猫 on Discord) + "tool_calls", "function_call": return "tool" - case "refusal": + case "length": + return "length" + case "refusal", "content_filter": return "filter" } return "stop" diff --git a/internal/gateway/continuation.go b/internal/gateway/continuation.go index be281e0d..77129762 100644 --- a/internal/gateway/continuation.go +++ b/internal/gateway/continuation.go @@ -17,6 +17,7 @@ package gateway import ( "context" + "encoding/json" "errors" "io" "net" @@ -37,6 +38,19 @@ const streamRetries = 2 // without waiting on a vendor that keeps the connection open after it. var errStreamCut = errors.New("stream cut mid-reply") +// errEndedShort is an Anthropic stream that ended, its connection closed +// cleanly, before it said its reply had: the passthrough's word for it. +var errEndedShort = errors.New("the reply ended before it was complete") + +// anthropicEnds says whether an Anthropic stream's event ends the reply: +// its message_stop (a stop_reason is upstreamStop's), or an error. +func anthropicEnds(data string) bool { + var v struct { + Type string `json:"type"` + } + return json.Unmarshal([]byte(data), &v) == nil && lastEvent(v.Type) +} + // groupTryKey marks a try of a routing group's member. A member whose // stream breaks with an error mid-reply is failed as it used to be — the // group answers the agent's retry with another of its members (#733) — @@ -328,8 +342,11 @@ func (s *Server) streamTranslated(w http.ResponseWriter, r *http.Request, p prov t := &textCallSee{names: names, see: see} attemptSee, held = t.event, t.release } + ended := false // the upstream said its reply ended serr = readSSEAlive(rd, func(_, data string) error { - u.add(Usage{Upstream: upstreamOf([]byte(data))}) + st := upstreamStop([]byte(data)) + u.add(Usage{Upstream: upstreamOf([]byte(data)), Stop: st}) + ended = ended || st != "" || actual == provider.Anthropic && anthropicEnds(data) if err := dec(data, attemptSee); err != nil { return err } @@ -342,6 +359,14 @@ func (s *Server) streamTranslated(w http.ResponseWriter, r *http.Request, p prov if errors.Is(serr, errStreamCut) { serr = nil } + if serr == nil && failed == "" && !ended && actual == provider.Anthropic && r.Context().Err() == nil { + // an Anthropic stream that just stopped — no + // stop_reason, no message_stop — is a reply cut + // short, not a finished one: a relay's (蓝猫 on + // Discord) read as whole ended the agent's turn a + // few words in, with nothing to say why + serr = errEndedShort + } } } if serr == nil && failed == "" { @@ -387,6 +412,9 @@ func (s *Server) streamTranslated(w http.ResponseWriter, r *http.Request, p prov // the upstream died mid-reply: say so in the client's own // protocol instead of finishing as if all went well failed = cutMidReply(p.Name, serr) + if errors.Is(serr, errEndedShort) { + failed = p.Name + ": " + errEndedShort.Error() + } } if failed != "" { // the same refusal as the 429's, said inside the reply diff --git a/internal/gateway/gateway.go b/internal/gateway/gateway.go index 7a6d97ea..2096a37c 100644 --- a/internal/gateway/gateway.go +++ b/internal/gateway/gateway.go @@ -2167,7 +2167,7 @@ func (s *Server) serve(w http.ResponseWriter, r *http.Request, from provider.Pro Input: call.Usage.Input, Output: call.Usage.Output, CacheRead: call.Usage.CacheRead, CacheWrite: call.Usage.CacheWrite, CacheWrite1h: call.Usage.CacheWrite1h, Reasoning: call.Usage.Reasoning, Effort: sent, Millis: call.Millis, Status: call.Status, TTFT: call.TTFT, FirstText: call.FirstText, Sent: sentMs, Session: sessionOf(r.Header), NativeSession: nativeSessionOf(r.Header), Kind: call.Kind, - RequestID: call.Usage.RequestID, ResponseID: call.Usage.ResponseID, Endpoint: endpointOf(r, from, call.To), Archive: call.archiveName()} + RequestID: call.Usage.RequestID, ResponseID: call.Usage.ResponseID, Endpoint: endpointOf(r, from, call.To), Stop: call.Usage.Stop, Archive: call.archiveName()} failedWith(&rec, call.Status, call.Error, call.Usage.ErrType) withBodies(&rec, &call) appendUsage(r, rec) @@ -3663,7 +3663,7 @@ func decoder(proto provider.Protocol) func(data string, emit func(Event)) error d := &codeAssistDecoder{} return d.decode } - d := &anthropicDecoder{server: map[int]bool{}} + d := &anthropicDecoder{server: map[int]bool{}, whole: map[int]bool{}} return d.decode } diff --git a/internal/gateway/ir.go b/internal/gateway/ir.go index ea0b0aa8..a5c4256d 100644 --- a/internal/gateway/ir.go +++ b/internal/gateway/ir.go @@ -220,6 +220,10 @@ type Usage struct { // ResponseID is the final client response ID, independent of request headers. ResponseID string `json:"response_id,omitempty"` ErrType string `json:"err_type,omitempty"` + // Stop is why the upstream said its reply ended, in its own words + // (stop_reason, finish_reason, a Responses status): "" when it said + // none, as a stream that just stops does (upstreamStop) + Stop string `json:"stop,omitempty"` } // prompt is every token the prompt came to, as OpenAI's and Gemini's @@ -263,6 +267,9 @@ func (u *Usage) add(v Usage) { if v.ErrType != "" { u.ErrType = v.ErrType } + if v.Stop != "" { + u.Stop = v.Stop + } } // Result is a whole reply, for non-streaming clients. @@ -354,7 +361,10 @@ func (c *collector) add(ev Event) { func (c *collector) finish() Result { c.closeTool() - if c.res.Stop == "" { + // a reply of tool calls ended "stop" (a relay's end_turn beside its + // tool_use blocks, 蓝猫 on Discord) is the client's to run the calls + // of: told it stopped, an agent ends its turn there + if c.res.Stop == "" || c.res.Stop == "stop" { c.res.Stop = "stop" for _, p := range c.res.Parts { if p.Kind == ToolCall { diff --git a/internal/gateway/relay_stop_test.go b/internal/gateway/relay_stop_test.go new file mode 100644 index 00000000..b9047036 --- /dev/null +++ b/internal/gateway/relay_stop_test.go @@ -0,0 +1,165 @@ +package gateway + +import ( + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/yetone/magpie/internal/provider" + "github.com/yetone/magpie/internal/usage" +) + +// Anthropic-shaped relays in front of other vendors' models (Grok's, as +// 蓝猫 on Discord had it) end their streams in ways Anthropic's own API +// doesn't: a reply of tool calls whose stop_reason says end_turn, or +// OpenAI's "tool_calls"; a tool_use block whose content_block_start +// carries the whole input and no input_json_delta after it; a stream that +// just stops, with no stop_reason and no message_stop. An agent told such +// a reply ended its turn ends it, a few words in, as if done. + +// relayAnthropic serves a translated request's model from an Anthropic +// upstream answering with events, as they are, and returns the Chat +// stream magpie made of it. +func relayAnthropic(t *testing.T, events ...string) string { + t.Helper() + fresh(t) + up := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + io.Copy(io.Discard, r.Body) + w.Header().Set("Content-Type", "text/event-stream") + for _, ev := range events { + io.WriteString(w, "data: "+ev+"\n\n") + } + })) + t.Cleanup(up.Close) + p := provider.Provider{ID: "up", Name: "UP", Key: "k", Models: []string{"grok-4"}, Anthropic: up.URL} + if err := provider.Save(p); err != nil { + t.Fatal(err) + } + code, body := post(t, "/v1/chat/completions", `{"model":"up/grok-4","stream":true,"messages":[{"role":"user","content":"list files"}],`+ + `"tools":[{"type":"function","function":{"name":"Bash","parameters":{"type":"object","properties":{"command":{"type":"string"}}}}}]}`) + if code != 200 { + t.Fatalf("status %d: %s", code, body) + } + return body +} + +// chatFinish is a Chat stream's finish_reason, chatArgs the arguments of +// its tool calls, run together. +func chatFinish(body string) string { + var out string + for _, ev := range events(body) { + for _, c := range choices(ev) { + if s, _ := c["finish_reason"].(string); s != "" { + out = s + } + } + } + return out +} + +func chatArgs(body string) string { + var sb strings.Builder + for _, ev := range events(body) { + for _, c := range choices(ev) { + d, _ := c["delta"].(map[string]any) + calls, _ := d["tool_calls"].([]any) + for _, x := range calls { + call, _ := x.(map[string]any) + fn, _ := call["function"].(map[string]any) + s, _ := fn["arguments"].(string) + sb.WriteString(s) + } + } + } + return sb.String() +} + +const ( + relayToolStart = `{"type":"content_block_start","index":1,"content_block":{"type":"tool_use","id":"toolu_1","name":"Bash","input":{}}}` + relayToolArgs = `{"type":"content_block_delta","index":1,"delta":{"type":"input_json_delta","partial_json":"{\"command\":\"ls\"}"}}` + relayToolStop = `{"type":"content_block_stop","index":1}` + relayEnd = `{"type":"message_stop"}` +) + +func relayStop(reason string) string { + return `{"type":"message_delta","delta":{"stop_reason":"` + reason + `"},"usage":{"output_tokens":9}}` +} + +// A reply of tool calls is the agent's to run, whatever stop_reason the +// relay gave it. +func TestRelayToolCallsStopForTools(t *testing.T) { + for _, reason := range []string{"end_turn", "tool_calls", "stop"} { + t.Run(reason, func(t *testing.T) { + body := relayAnthropic(t, anthStart("grok-4", 5), anthText("Let me look."), relayToolStart, relayToolArgs, relayToolStop, relayStop(reason), relayEnd) + if got := chatFinish(body); got != "tool_calls" { + t.Errorf("finish_reason %q, want tool_calls: %s", got, body) + } + if got := chatArgs(body); got != `{"command":"ls"}` { + t.Errorf("arguments %q", got) + } + }) + } +} + +// A tool_use block that starts with its whole input, no delta after it, +// keeps that input. +func TestRelayToolInputInStart(t *testing.T) { + start := `{"type":"content_block_start","index":0,"content_block":{"type":"tool_use","id":"toolu_1","name":"Bash","input":{"command":"ls"}}}` + body := relayAnthropic(t, anthStart("grok-4", 5), start, `{"type":"content_block_stop","index":0}`, relayStop("tool_use"), relayEnd) + if got := chatArgs(body); got != `{"command":"ls"}` { + t.Errorf("arguments %q, want the start's input: %s", got, body) + } + if got := chatFinish(body); got != "tool_calls" { + t.Errorf("finish_reason %q", got) + } +} + +// A stream that stops with no stop_reason and no message_stop is a reply +// cut short, and the client is told so rather than handed a finished one. +func TestRelayStreamEndedShort(t *testing.T) { + body := relayAnthropic(t, anthStart("grok-4", 5), anthText("Let me look at the")) + if chatFinish(body) == "stop" { + t.Errorf("a cut reply finished as whole: %s", body) + } + if !strings.Contains(body, "ended before it was complete") { + t.Errorf("no error for the cut reply: %s", body) + } + // a stream with a stop_reason and no message_stop said it was done + body = relayAnthropic(t, anthStart("grok-4", 5), anthText("Done."), relayStop("end_turn")) + if got := chatFinish(body); got != "stop" || strings.Contains(body, "ended before") { + t.Errorf("finish_reason %q: %s", got, body) + } +} + +// The upstream's own stop reason is kept with the request, for a reply +// that ended too soon to be told apart by: the relay's word, not magpie's. +func TestRelayStopReasonRecorded(t *testing.T) { + relayAnthropic(t, anthStart("grok-4", 5), anthText("Done."), relayStop("end_turn"), relayEnd) + recs := usage.Load(time.Time{}) + if len(recs) != 1 || recs[0].Stop != "end_turn" { + t.Fatalf("records %+v, want stop end_turn", recs) + } + // and a reply relayed as it came, to an agent of the same protocol + fresh(t) + up := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + io.Copy(io.Discard, r.Body) + w.Header().Set("Content-Type", "text/event-stream") + for _, ev := range []string{anthStart("grok-4", 5), anthText("Done."), relayStop("tool_calls"), relayEnd} { + io.WriteString(w, "data:"+ev+"\r\n\r\n") + } + })) + defer up.Close() + if err := provider.Save(provider.Provider{ID: "up", Name: "UP", Key: "k", Models: []string{"grok-4"}, Anthropic: up.URL}); err != nil { + t.Fatal(err) + } + if code, body := post(t, "/v1/messages", `{"model":"up/grok-4","stream":true,"max_tokens":9,"messages":[{"role":"user","content":"hi"}]}`); code != 200 { + t.Fatalf("status %d: %s", code, body) + } + recs = usage.Load(time.Time{}) + if len(recs) != 1 || recs[0].Stop != "tool_calls" { + t.Fatalf("records %+v, want stop tool_calls", recs) + } +} diff --git a/internal/gateway/sniff.go b/internal/gateway/sniff.go index 59a7aca8..5a37d94f 100644 --- a/internal/gateway/sniff.go +++ b/internal/gateway/sniff.go @@ -127,6 +127,9 @@ func (s *usageSniffer) parse(b []byte) { if up := upstreamOf(b); up != "" { s.u.Upstream = up } + if st := upstreamStop(b); st != "" { + s.u.Stop = st + } switch s.proto { case provider.Gemini: // Factory's generateContent chunks, the same usageMetadata Code diff --git a/internal/gateway/upstream.go b/internal/gateway/upstream.go index 51343aad..096a904e 100644 --- a/internal/gateway/upstream.go +++ b/internal/gateway/upstream.go @@ -38,3 +38,51 @@ func upstreamOf(b []byte) string { } return "" } + +// upstreamStop is why a reply (a JSON body, or one event of its stream) +// says it ended, in the upstream's own words: an Anthropic stop_reason, a +// Chat finish_reason, a Responses reply's status and the reason it was +// incomplete. Kept with the request, so a reply that ended too soon shows +// what the upstream said of it — a relay's "end_turn" after a few words, +// or nothing at all (蓝猫 on Discord). "" when this one says none. +func upstreamStop(b []byte) string { + if !bytes.Contains(b, []byte(`stop_reason"`)) && !bytes.Contains(b, []byte(`finish_reason"`)) && !bytes.Contains(b, []byte(`"response.`)) { + return "" + } + var v struct { + Type string `json:"type"` + StopReason string `json:"stop_reason"` + Delta struct { + StopReason string `json:"stop_reason"` + } `json:"delta"` + Choices []struct { + FinishReason string `json:"finish_reason"` + } `json:"choices"` + Response struct { + Status string `json:"status"` + IncompleteDetails *struct { + Reason string `json:"reason"` + } `json:"incomplete_details"` + } `json:"response"` + } + if json.Unmarshal(b, &v) != nil { + return "" + } + switch { + case v.Delta.StopReason != "": + return v.Delta.StopReason + case v.StopReason != "": + return v.StopReason + case strings.HasPrefix(v.Type, "response.") && v.Response.Status != "" && v.Response.Status != "in_progress" && v.Response.Status != "queued": + if d := v.Response.IncompleteDetails; d != nil && d.Reason != "" { + return v.Response.Status + ": " + d.Reason + } + return v.Response.Status + } + for _, c := range v.Choices { + if c.FinishReason != "" { + return c.FinishReason + } + } + return "" +} diff --git a/internal/gui/assets/app.css b/internal/gui/assets/app.css index 551b677a..83562b01 100644 --- a/internal/gui/assets/app.css +++ b/internal/gui/assets/app.css @@ -914,6 +914,7 @@ body.window .field.solo { grid-column: auto; } .led .src.official::before { content: ""; width: 6px; height: 3.5px; border-left: 1.5px solid currentColor; border-bottom: 1.5px solid currentColor; transform: translateY(-1px) rotate(-45deg); } .led .src.access-api { --badge-ink: var(--tk-f); } .led .src.access-subscription { --badge-ink: var(--tk-k); } +.led .src.proto { --badge-ink: var(--muted); font-weight: 500; } .led .duration { font-weight: 550; } .led .duration-fast { color: var(--tk-s); } .led .duration-slow { color: var(--amber); } diff --git a/internal/gui/assets/app.js b/internal/gui/assets/app.js index dd874c08..b01a7e39 100644 --- a/internal/gui/assets/app.js +++ b/internal/gui/assets/app.js @@ -13622,6 +13622,28 @@ function ledServed(r) { return k; } +// the protocol a request was sent upstream in, told by the path it went +// out on, and the one its agent spoke when that was another: the +// endpoint reads "/v1/chat/completions → /v1/messages" for a request +// translated, the agent's own path alone for one sent as it came (蓝猫 on +// Discord). null for a request with no path kept, a session file's. +const ledProtoOf = (path) => /\/messages\b/.test(path) ? "Anthropic" : /\/responses\b/.test(path) ? "Responses" + : /\/chat\/completions\b/.test(path) ? "Chat" : /generateContent|\/generate\b/.test(path) ? "Gemini" : ""; +function ledProtos(r) { + if (!r.ep || r.source === "log") return null; + const [a, b] = String(r.ep).split(" → "); + const from = ledProtoOf(a), to = b ? ledProtoOf(b) : from; + if (!to) return null; + return { from: from || to, to }; +} +function ledProtoBadge(r) { + const p = ledProtos(r); + if (!p) return null; + const k = el("span", "src proto proto-" + p.to.toLowerCase(), p.from !== p.to ? p.from + " → " + p.to : p.to); + k.title = p.from !== p.to ? t("The agent spoke {from}; sent upstream as {to}", p) : t("Sent upstream as {to}, as the agent spoke it", p); + return k; +} + // a request that failed: told by its status, or, for one read from a // session file, which records none, by the error that ended it const ledFailed_ = (r) => r.status >= 400 || !!r.err; @@ -13669,6 +13691,11 @@ function ledDetail(r, cols) { if (ledRowSpeed(r)) add("Speed", t("{n} tok/s", { n: ledNum(Math.round(ledRowSpeed(r))) })); add("Request ID", r.rid); add("Endpoint", r.ep); + const protos = ledProtos(r); + if (protos) add("Protocol", protos.from !== protos.to ? protos.from + " → " + protos.to : protos.to); + // why the upstream said its reply ended, in its own words: a reply that + // ended too soon is told apart by it (蓝猫 on Discord) + add("Upstream stop reason", r.stop); // the provider an aggregator (OpenRouter …) said answered behind it add("Upstream provider", r.upstream); add("Session ID", r.session); @@ -14424,6 +14451,8 @@ function renderLedger() { badge.title = t(access); badges.append(badge); } + const proto = ledProtoBadge(r); + if (proto) badges.append(proto); if (local && r.session_account) { const k = el("span", "src local", t("Local session")); k.title = t("Read from the agent's session file. The account is shown only when local metadata identifies it; no service provider is inferred."); diff --git a/internal/gui/assets/i18n.js b/internal/gui/assets/i18n.js index 707f3bec..33dd877f 100644 --- a/internal/gui/assets/i18n.js +++ b/internal/gui/assets/i18n.js @@ -2811,6 +2811,10 @@ const I18N = { "Upstream: {upstream}": "上游：{upstream}", "The aggregator passed the request on to {upstream}, as its reply says: the provider that actually answered it.": "聚合供应商把这条请求转给了 {upstream}，由它实际作答（聚合供应商的回复里写明）。", "Upstream provider": "上游", + "Protocol": "协议", + "Upstream stop reason": "上游结束原因", + "The agent spoke {from}; sent upstream as {to}": "agent 用 {from} 协议请求；以 {to} 协议发往上游", + "Sent upstream as {to}, as the agent spoke it": "以 {to} 协议发往上游，与 agent 所用协议相同", "{sent} is a routing group of the remote magpie, and it routed the request to {served}: the group picking one of its models, not the vendor swapping the model.": "{sent} 是远程 magpie 的路由组，请求被路由给 {served}：这是路由组选择了组内模型，并非服务商替换了模型。", "Open Routing": "打开路由", "The vendor was asked for {sent}, but its reply says {served} answered: likely another model. The same model under a dated name or spelled otherwise isn't marked.": "向服务商请求的是 {sent}，但回复说是 {served} 作答，很可能换成了别的模型。（同一模型只是带了日期或写法不同，不会这样标出。）", @@ -6173,6 +6177,10 @@ const I18N = { "Upstream: {upstream}": "上流：{upstream}", "The aggregator passed the request on to {upstream}, as its reply says: the provider that actually answered it.": "アグリゲーターはこのリクエストを {upstream} に回したと応答に書いています。実際に応答したプロバイダーです。", "Upstream provider": "上流プロバイダー", + "Protocol": "プロトコル", + "Upstream stop reason": "上流の終了理由", + "The agent spoke {from}; sent upstream as {to}": "エージェントは {from} で要求し、上流へは {to} で送信", + "Sent upstream as {to}, as the agent spoke it": "エージェントと同じ {to} で上流へ送信", "{sent} is a routing group of the remote magpie, and it routed the request to {served}: the group picking one of its models, not the vendor swapping the model.": "{sent} はリモート magpie のルーティンググループで、リクエストを {served} にルーティングしました。グループがモデルを選んだもので、ベンダーがモデルを差し替えたのではありません。", "Open Routing": "ルーティングを開く", "The vendor was asked for {sent}, but its reply says {served} answered: likely another model. The same model under a dated name or spelled otherwise isn't marked.": "ベンダーには {sent} をリクエストしましたが、応答では {served} が答えたとあり、別のモデルに替えられた可能性があります。（同じモデルに日付が付いただけ、または表記が違うだけなら表示しません。）", @@ -9535,6 +9543,10 @@ const I18N = { "Upstream: {upstream}": "Upstream: {upstream}", "The aggregator passed the request on to {upstream}, as its reply says: the provider that actually answered it.": "Der Aggregator hat die Anfrage laut seiner Antwort an {upstream} weitergegeben: der Anbieter, der sie tatsächlich beantwortet hat.", "Upstream provider": "Upstream-Anbieter", + "Protocol": "Protokoll", + "Upstream stop reason": "Endgrund des Upstreams", + "The agent spoke {from}; sent upstream as {to}": "Der Agent sprach {from}; an den Upstream als {to} gesendet", + "Sent upstream as {to}, as the agent spoke it": "An den Upstream als {to} gesendet, wie der Agent es sprach", "{sent} is a routing group of the remote magpie, and it routed the request to {served}: the group picking one of its models, not the vendor swapping the model.": "{sent} ist eine Weiterleitungsgruppe des entfernten magpie und hat die Anfrage an {served} weitergeleitet: Die Gruppe wählt eines ihrer Modelle, nicht der Anbieter tauscht das Modell.", "Open Routing": "Weiterleitung öffnen", "The vendor was asked for {sent}, but its reply says {served} answered: likely another model. The same model under a dated name or spelled otherwise isn't marked.": "Beim Anbieter wurde {sent} angefragt, laut Antwort hat aber {served} geantwortet – wahrscheinlich ein anderes Modell. (Dasselbe Modell mit Datum oder anderer Schreibweise wird nicht markiert.)", diff --git a/internal/gui/tests/README.md b/internal/gui/tests/README.md index 1f320de6..ef1f3cf0 100644 --- a/internal/gui/tests/README.md +++ b/internal/gui/tests/README.md @@ -1851,3 +1851,15 @@ string has zh, ja and de. English and Chinese, Chromium and WebKit. ```sh node --test internal/gui/tests/lane-limit.test.cjs ``` + +`ledger-protocol.test.cjs` checks the protocol each request was sent +upstream in, on Usage › Requests (蓝猫 on Discord): a row translated from +its agent's protocol is tagged "Chat → Anthropic", one sent as it came +"Anthropic" or "Responses", one read from a session file nothing. A +request's details say Protocol and the upstream's own stop reason, when it +gave one. A click opens the details and moves nothing; no left border; +every string has zh, ja and de. English and Chinese, Chromium and WebKit. + +```sh +node --test internal/gui/tests/ledger-protocol.test.cjs +``` diff --git a/internal/gui/tests/ledger-protocol.test.cjs b/internal/gui/tests/ledger-protocol.test.cjs new file mode 100644 index 00000000..c46881c2 --- /dev/null +++ b/internal/gui/tests/ledger-protocol.test.cjs @@ -0,0 +1,128 @@ +// Run with Node's test runner and Playwright on the module path; see README.md. +// The protocol each request was sent upstream in (蓝猫 on Discord: 「请求」 +// 列表里每条请求显示它用的是 Anthropic / Responses / Chat 哪种协议): the +// Usage page's Requests list tags each row with it, "Chat → Anthropic" +// for one translated from the agent's own, and a request's details name +// it beside the upstream's own stop reason. A row read from a session +// file, which has no path, has no tag. No click moves the page, no +// left-border accent. English and Chinese; the API is faked. +const assert = require("node:assert/strict"); +const fs = require("node:fs/promises"); +const path = require("node:path"); +const { test } = require("node:test"); +const { chromium, webkit } = require("playwright"); +const reader = require("./reader.cjs"); + +const assets = path.resolve(__dirname, "../assets"); +const now = new Date(); +const at = (i) => new Date(now.getTime() - (i + 1) * 60e3).toISOString(); +const row = (i, ep, stop, extra) => ({ + t: at(i), agent: "opencode", agentName: "OpenCode", icon: "opencode", provider: "relay", providerName: "Relay", + req: "relay/grok-4", model: "grok-4", served: "grok-4", in: 300, out: 40, ms: 2000, status: 200, rid: "r-" + i, + ...(ep ? { ep } : {}), ...(stop ? { stop } : {}), ...extra, +}); +const ROWS = [ + row(0, "/v1/chat/completions → /v1/messages", "end_turn"), + row(1, "/v1/messages", "tool_use"), + row(2, "/v1/responses", ""), + row(3, "", "", { source: "log", status: 0, provider: "", providerName: "" }), +]; + +function serve(lang) { + const state = { agents: [{ id: "opencode", name: "OpenCode", path: "/test/opencode.json", fields: [] }], profiles: [], settings: { lang, theme: "light" } }; + return async (route) => { + const url = new URL(route.request().url()); + const json = (data) => route.fulfill({ json: data }); + if (url.pathname === "/boot.js") return route.fulfill({ contentType: "text/javascript", body: `window.bootPrefs = {lang:"${lang}",theme:"light",web:true};` }); + if (url.pathname === "/wails/runtime.js") return route.fulfill({ contentType: "text/javascript", body: "export const Window = {};" }); + if (url.pathname === "/api/state") return json(state); + if (url.pathname === "/api/gateway/trace") { + if (url.searchParams.get("wait")) await new Promise((r) => setTimeout(r, 20e3)); + return json({ mine: true, now: now.toISOString(), seq: 1, totals: { requests: 0, rerouted: 0, errors: 0 }, routes: [] }); + } + if (url.pathname === "/api/groups") return json({ groups: [], models: [] }); + if (url.pathname === "/api/providers") return json({ providers: [], excluded: [], gateway: { running: true, window: true } }); + if (url.pathname === "/api/plugins") return json({ plugins: [] }); + if (url.pathname === "/api/usage/requests") return json({ period: url.searchParams.get("period"), rows: ROWS, offset: 0, total: ROWS.length, calls: ROWS.length, errors: 0, input: 1200, output: 160, cache_read: 0, cache_write: 0, reasoning: 0, cost: 0, unpriced: ROWS.length, agents: [{ id: "opencode", name: "OpenCode", icon: "opencode" }] }); + if (url.pathname === "/api/usage/requests/content") return json({ found: false, why: "read" }); + if (url.pathname === "/api/usage/quotas") return json([]); + if (url.pathname === "/api/usage") return json({ calls: 4, errors: 0, input: 1200, output: 160, cache_read: 0, cache_write: 0, reasoning: 0, unpriced: 4, cost: 0, bucket: "day", series: [], agents: [], models: [], path: "~/.config/magpie/usage.jsonl" }); + if (url.pathname === "/api/sessions") return json({ sessions: [], dirs: [] }); + if (url.pathname === "/api/sessions/stats") return json({ from: "", to: "", days: [], agents: {} }); + if (url.pathname.startsWith("/api/")) return json({}); + const file = path.join(assets, url.pathname === "/" ? "index.html" : url.pathname); + const contentType = { ".html": "text/html", ".js": "text/javascript", ".css": "text/css", ".svg": "image/svg+xml", ".png": "image/png" }[path.extname(file)]; + await route.fulfill({ body: await fs.readFile(file), contentType }); + }; +} + +const want = { + en: { proto: "Protocol", stop: "Upstream stop reason", translated: /The agent spoke Chat; sent upstream as Anthropic/ }, + zh: { proto: "协议", stop: "上游结束原因", translated: /agent 用 Chat 协议请求；以 Anthropic 协议发往上游/ }, +}; + +for (const engine of (process.env.BROWSER ? [process.env.BROWSER] : ["chromium", "webkit"])) { + for (const lang of ["en", "zh"]) { + test(`${engine} ${lang}: each request names its protocol`, async (t) => { + const browser = await (engine === "webkit" ? webkit.launch() : chromium.launch({ channel: "chromium" })); + const context = await browser.newContext({ viewport: { width: 1280, height: 760 }, reducedMotion: "reduce" }); + const page = await context.newPage(); + page.setDefaultTimeout(5000); + const errors = []; + page.on("pageerror", (e) => errors.push(e.message)); + await page.route("**/*", serve(lang)); + t.after(async () => { + if (process.env.ARTIFACT_DIR) { + await fs.mkdir(process.env.ARTIFACT_DIR, { recursive: true }); + await page.screenshot({ path: path.join(process.env.ARTIFACT_DIR, `${engine}-${lang}-protocol.png`), fullPage: true }); + } + await browser.close(); + }); + await page.goto("http://magpie.test/?view=usage"); + await page.locator("#usageTab .opt").nth(1).click(); + const rows = page.locator(".led tbody tr.led-row"); + await rows.nth(ROWS.length - 1).waitFor({ timeout: 15000 }); + + // the list: every request the gateway logged says its protocol + const tags = await rows.evaluateAll((rs) => rs.map((r) => r.querySelector(".src.proto")?.textContent || "")); + assert.deepEqual(tags, ["Chat → Anthropic", "Anthropic", "Responses", ""]); + const look = await page.locator(".led .src.proto").evaluateAll((es) => es.map((e) => { + const cell = e.closest("td").getBoundingClientRect(), b = e.getBoundingClientRect(), cs = getComputedStyle(e); + return { inCell: b.left >= cell.left - 0.5 && b.right <= cell.right + 0.5 && b.width > 0, full: e.clientWidth >= e.scrollWidth, border: cs.borderLeftWidth, top: cs.borderTopWidth, title: e.title }; + })); + for (const l of look) { + assert(l.inCell && l.full, JSON.stringify(l)); + assert.equal(l.border, l.top, "no left-border accent"); + } + assert.match(look[0].title, want[lang].translated); + + // a request's details: the protocol, and what the upstream said of its end + const top = () => page.locator("#view-usage").evaluate((v) => v.scrollTop); + const dd = async (i, dt) => { + await reader.inView(page, rows.nth(i)); + const was = await top(); + const scrolled = await page.evaluate(() => [scrollX, scrollY]); + await rows.nth(i).click(); + await page.waitForTimeout(200); + assert.equal(await top(), was, "a click moved the page"); + assert.deepEqual(await page.evaluate(() => [scrollX, scrollY]), scrolled); + const box = page.locator(".led-detail").last(); + await box.waitFor(); + const v = await box.evaluate((b, dts) => dts.map((dt) => [...b.querySelectorAll("dt")].find((d) => d.textContent === dt)?.nextElementSibling?.textContent || ""), dt); + await rows.nth(i).click(); // closed again + return v; + }; + assert.deepEqual(await dd(0, [want[lang].proto, want[lang].stop]), ["Chat → Anthropic", "end_turn"]); + assert.deepEqual(await dd(1, [want[lang].proto, want[lang].stop]), ["Anthropic", "tool_use"]); + assert.deepEqual(await dd(2, [want[lang].proto, want[lang].stop]), ["Responses", ""]); + assert.deepEqual(await dd(3, [want[lang].proto, want[lang].stop]), ["", ""]); + if (process.env.ARTIFACT_DIR) await page.screenshot({ path: path.join(process.env.ARTIFACT_DIR, `${engine}-${lang}-protocol-list.png`) }); + for (const l of ["zh", "ja", "de"]) { + const missing = await page.evaluate((l) => ["Protocol", "Upstream stop reason", + "The agent spoke {from}; sent upstream as {to}", "Sent upstream as {to}, as the agent spoke it"].filter((k) => !I18N[l][k]), l); + assert.deepEqual(missing, [], `every string has its ${l}`); + } + assert.deepEqual(errors, []); + }); + } +} diff --git a/internal/usage/usage.go b/internal/usage/usage.go index 35c776a0..cdc4cfae 100644 --- a/internal/usage/usage.go +++ b/internal/usage/usage.go @@ -95,6 +95,11 @@ type Record struct { // ResponseID is the response object ID actually sent to the client. ResponseID string `json:"response_id,omitempty"` Endpoint string `json:"ep,omitempty"` + // Stop is why the upstream said its reply ended, in its own words + // (stop_reason, finish_reason …): "" when it said none, a stream that + // just stopped. Told apart by it, a reply that ended too soon is the + // upstream's own end_turn or a cut (蓝猫 on Discord). + Stop string `json:"stop,omitempty"` // Session is the conversation the call was part of, as its agent names // it (X-Magpie-Session, or the session header Claude Code, Codex or // OpenCode sends): several sessions on one model told apart