diff --git a/docs/reference.md b/docs/reference.md index 658eac71..a9b3a070 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -260,8 +260,13 @@ first text (`usage.DecodeOf`): hidden reasoning is written before the stream shows anything, so counting it in a window that starts after it read gpt-6.1-sol at hundreds of tok/s. A reply that reasoned and then wrote only tool calls, one not streamed, and a burst (under 100 ms, or over 10,000 tok/s) -tell no speed. History is read the same way, as it keeps the reasoning and the -first text. +tell no speed. The window is no longer than the reply's content took to come +(`flow_ms`, from a tenth of its bytes to nine tenths, scaled to the whole): a +vendor that sends the first words, holds the rest and lets it go in one burst +gives a flow of next to none, so such a reply tells no speed instead of its +tokens over the wait. History is read the same way, as it keeps the reasoning, +the first text and the flow; a record from before the flow was kept is read +over its whole window. It lists **accounts** too: each Codex, Claude or other subscription account's tokens and cost, by the account that actually answered — the one that took diff --git a/internal/gateway/codex_backend.go b/internal/gateway/codex_backend.go index 9d1e870c..03c71ed5 100644 --- a/internal/gateway/codex_backend.go +++ b/internal/gateway/codex_backend.go @@ -432,6 +432,8 @@ func (s *Server) codexUpstream(w http.ResponseWriter, r *http.Request, rest stri t.TitleLink = &link } t.Output, t.Reasoning, t.TTFT, t.FirstText = out, uu.Reasoning, ttft, text + flow := first.flowFor(uu.Reasoning) + t.Flow, t.Tries[0].Flow = flow, flow t.Usage = routeUsage("openai", model, uu) if prompt != nil { t.Prompt = prompt.calibrated(promptCounted(t.Usage), catalog.ContextOf(model)) @@ -578,6 +580,7 @@ func (s *Server) codexUpstream(w http.ResponseWriter, r *http.Request, rest stri Millis: time.Since(start).Milliseconds(), Fallback: resetNote} call.TTFT, call.FirstText = first.ms() uu.add(sniff.usage()) + call.Flow = first.flowFor(uu.Reasoning) call.Usage, served = uu, uu.Served errType := "" if res.StatusCode >= 400 { @@ -591,7 +594,7 @@ func (s *Server) codexUpstream(w http.ResponseWriter, r *http.Request, rest stri rec := usage.Record{RouteID: tr.ID, Time: start, Agent: call.Agent, Provider: call.Provider, Host: provider.HostOf(base), Model: call.Model, Requested: call.Model, Served: served, Input: uu.Input, Output: uu.Output, CacheRead: uu.CacheRead, CacheWrite: uu.CacheWrite, - Reasoning: uu.Reasoning, Effort: effort, Millis: call.Millis, TTFT: call.TTFT, FirstText: call.FirstText, Status: call.Status, Session: sessionOf(r.Header), NativeSession: nativeSessionOf(r.Header), Kind: call.Kind, + Reasoning: uu.Reasoning, Effort: effort, Millis: call.Millis, TTFT: call.TTFT, FirstText: call.FirstText, Flow: call.Flow, Status: call.Status, Session: sessionOf(r.Header), NativeSession: nativeSessionOf(r.Header), Kind: call.Kind, RequestID: requestID(res.Header), ResponseID: uu.ResponseID, Endpoint: r.URL.Path} failedWith(&rec, call.Status, call.Error, errType) appendUsage(r, rec) diff --git a/internal/gateway/gateway.go b/internal/gateway/gateway.go index abd1bbc1..511165e5 100644 --- a/internal/gateway/gateway.go +++ b/internal/gateway/gateway.go @@ -275,8 +275,12 @@ type Call struct { // TTFT: ms from the request to its reply's first content — text, // reasoning or a tool call — and FirstText to its first text, when // it was streamed (#196) - TTFT int64 `json:"ttft,omitempty"` - FirstText int64 `json:"firstText,omitempty"` + TTFT int64 `json:"ttft,omitempty"` + FirstText int64 `json:"firstText,omitempty"` + // Flow: the ms its content took to come, from where its speed is + // counted (firstToken.flowMs): next to none for a reply that came in + // a burst, so it tells no speed + Flow int64 `json:"flow,omitempty"` Error string `json:"error,omitempty"` Fallback string `json:"fallback,omitempty"` // providers that failed first, and why Usage Usage `json:"usage"` @@ -1954,6 +1958,8 @@ func (s *Server) serve(w http.ResponseWriter, r *http.Request, from provider.Pro asName := provider.SentNameOnIn(wiresOf(r.Context()), c.p.ID, accountAgent(c.p), c.model, sent) try.Swapped, try.Routed = swapped(asName, call.Usage.Served), usage.GroupRouted(asName, call.Usage.Served) try.TTFT, try.FirstText = hw.first.ms() + try.Flow = hw.first.flowFor(call.Usage.Reasoning) + call.Flow = try.Flow // the request's, from when it came as its ms are: the time before // this try, the ones that failed first, is in it call.TTFT, call.FirstText = sinceStart(began.Sub(start), hw.first.first), sinceStart(began.Sub(start), hw.first.text) @@ -1975,7 +1981,7 @@ func (s *Server) serve(w http.ResponseWriter, r *http.Request, from provider.Pro // rest, a long prompt being slow to start anywhere call.Status, call.Error = http.StatusGatewayTimeout, fmt.Sprintf("%s: no first token in %ds", c.p.Name, g.FirstToken) try.Status, try.Error, try.Fail = call.Status, call.Error, failSlow - call.TTFT, call.FirstText = 0, 0 + call.TTFT, call.FirstText, call.Flow = 0, 0, 0 telemetry.attempt(call, try, c.p.ID, sent, hw, capture) s.trace.update(tr, func(t *Route) { t.Tries[len(t.Tries)-1] = try }) if other == nil { @@ -2065,7 +2071,7 @@ func (s *Server) serve(w http.ResponseWriter, r *http.Request, from provider.Pro Requested: call.Model, Served: call.Usage.Served, Upstream: call.Usage.Upstream, 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: time.Since(began).Milliseconds(), Status: call.Status, - TTFT: try.TTFT, FirstText: try.FirstText, Session: sessionOf(r.Header), NativeSession: nativeSessionOf(r.Header), Kind: call.Kind, + TTFT: try.TTFT, FirstText: try.FirstText, Flow: try.Flow, Session: sessionOf(r.Header), NativeSession: nativeSessionOf(r.Header), Kind: call.Kind, RequestID: call.Usage.RequestID, Endpoint: endpointOf(r, from, call.To), Archive: call.archiveName()} failedWith(&rec, call.Status, call.Error, call.Usage.ErrType) // what this account answered is its refusal, the reply @@ -2102,7 +2108,7 @@ func (s *Server) serve(w http.ResponseWriter, r *http.Request, from provider.Pro Requested: call.Model, Served: call.Usage.Served, 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: time.Since(began).Milliseconds(), Status: call.Status, - TTFT: try.TTFT, FirstText: try.FirstText, Session: sessionOf(r.Header), NativeSession: nativeSessionOf(r.Header), Kind: call.Kind, + TTFT: try.TTFT, FirstText: try.FirstText, Flow: try.Flow, Session: sessionOf(r.Header), NativeSession: nativeSessionOf(r.Header), Kind: call.Kind, RequestID: call.Usage.RequestID, Endpoint: endpointOf(r, from, call.To), Archive: call.archiveName()} failedWith(&rec, call.Status, call.Error, call.Usage.ErrType) // what this account answered is its refusal, the reply @@ -2415,7 +2421,7 @@ func (s *Server) serve(w http.ResponseWriter, r *http.Request, from provider.Pro t.Usage = append(t.Usage, routeUsage(call.Provider, model, call.Usage)...) } t.Tokens = call.Usage.Input + call.Usage.Output + call.Usage.CacheRead + call.Usage.CacheWrite - t.Output, t.Reasoning, t.TTFT, t.FirstText = call.Usage.Output, call.Usage.Reasoning, call.TTFT, call.FirstText + t.Output, t.Reasoning, t.TTFT, t.FirstText, t.Flow = call.Usage.Output, call.Usage.Reasoning, call.TTFT, call.FirstText, call.Flow if prompt != nil { t.Prompt = prompt.calibrated(promptCounted(t.Usage), window) } @@ -2431,7 +2437,7 @@ func (s *Server) serve(w http.ResponseWriter, r *http.Request, from provider.Pro Requested: call.Model, Served: call.Usage.Served, Upstream: call.Usage.Upstream, 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, + TTFT: call.TTFT, FirstText: call.FirstText, Flow: call.Flow, 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), Stop: call.Usage.Stop, Archive: call.archiveName()} failedWith(&rec, call.Status, call.Error, call.Usage.ErrType) withBodies(&rec, &call) diff --git a/internal/gateway/trace.go b/internal/gateway/trace.go index 8ebaf381..c259e946 100644 --- a/internal/gateway/trace.go +++ b/internal/gateway/trace.go @@ -67,6 +67,8 @@ type Route struct { // counts: streamed replies only (#196) TTFT int64 `json:"ttft,omitempty"` FirstText int64 `json:"firstText,omitempty"` + // Flow: the ms its content took to come (Call.Flow) + Flow int64 `json:"flow,omitempty"` // Served: the model the reply says answered, as the last try has it; // Swapped: another than the one that try asked for; Routed: that try // asked another magpie's routing group, and Served is its member @@ -248,6 +250,8 @@ type Try struct { // first text, when it streamed any (#196) TTFT int64 `json:"ttft,omitempty"` FirstText int64 `json:"firstText,omitempty"` + // Flow: the ms its content took to come (Call.Flow) + Flow int64 `json:"flow,omitempty"` // Served: the model its reply said answered, when it named one; // Swapped: another model than Model, not just its dated name; Routed: // Model is another magpie's routing group, and Served the member it diff --git a/internal/gateway/ttft.go b/internal/gateway/ttft.go index 985ee70f..d42f692c 100644 --- a/internal/gateway/ttft.go +++ b/internal/gateway/ttft.go @@ -17,7 +17,8 @@ import ( // that is all tool calls has no text to wait for. const firstMost = 8 << 20 -// firstToken times a stream's first content and first text from start. +// firstToken times a stream's first content and first text from start, +// and when the rest of its content came (flowMs). type firstToken struct { start time.Time first time.Duration // to the first content; 0 until it comes @@ -26,6 +27,16 @@ type firstToken struct { began bool read int pend []byte + // flow: how many bytes of content events had come by each time, one + // mark a millisecond, from the first content on + flow []flowMark + sum int +} + +// flowMark is how many bytes of content events had come by at. +type flowMark struct { + at time.Duration + sum int } // see reads what was written of the reply. @@ -48,23 +59,32 @@ func (f *firstToken) see(b []byte) { return } f.pend = append(f.pend, b...) + d := max(time.Since(f.start), time.Millisecond) for !f.off { end := eventEnd(f.pend) if end < 0 { break } - content, text := firstKind(f.pend[:end]) + ev := f.pend[:end] f.pend = f.pend[end:] + if f.text > 0 { + // the first text is known: what is left to tell is when the + // rest came, which a quicker look tells + if flowContent(ev) { + f.mark(d, len(ev)) + } + continue + } + content, text := firstKind(ev) if !content && !text { continue } - d := max(time.Since(f.start), time.Millisecond) if f.first == 0 { f.first = d } + f.mark(d, len(ev)) if text { f.text = d - f.off, f.pend = true, nil // all there is to know } } if len(f.pend) == 0 { @@ -72,6 +92,67 @@ func (f *firstToken) see(b []byte) { } } +// flowMost is the most marks kept: a reply that took longer than this many +// distinct milliseconds to come has its flow untold (flowMs). +const flowMost = 1 << 16 + +// mark counts n bytes of content as come by d. +func (f *firstToken) mark(d time.Duration, n int) { + f.sum += n + if k := len(f.flow); k > 0 && f.flow[k-1].at == d { + f.flow[k-1].sum = f.sum + return + } + if len(f.flow) >= flowMost { + f.off, f.pend, f.flow = true, nil, nil + return + } + f.flow = append(f.flow, flowMark{d, f.sum}) +} + +// flowContent: a server-sent event of a reply that carries content, told by +// a look rather than a parse: a delta of any protocol, or Gemini's parts. +// The end of a reply (Responses' response.completed, which says the whole +// reply again) carries none. +func flowContent(ev []byte) bool { + return bytes.Contains(ev, []byte("delta")) || bytes.Contains(ev, []byte(`"parts"`)) +} + +// flowMs is how long the content of a reply took to come from its decode's +// start from (its first content, or first text, as ms): the time its middle +// 80% of content bytes took, scaled to the whole, so a reply that came +// evenly gets its whole window, and one that came in a burst — the vendor +// wrote it all at once after its first content (John on Discord: 1,367 +// tok/s from Kimi Code) — gets next to none, and tells no speed +// (usage.DecodeOf). At least 1 when told; 0 when it can't be: no content +// after from, or a stream read too far to keep (flowMost, firstMost). +func (f *firstToken) flowMs(from int64) int64 { + if from <= 0 || len(f.flow) == 0 || f.read > firstMost { + return 0 + } + base := 0 + for _, m := range f.flow { + if m.at.Milliseconds() >= from { + break + } + base = m.sum + } + total := f.sum - base + if total <= 0 { + return 0 + } + at := func(share int) time.Duration { + for _, m := range f.flow { + if (m.sum-base)*100 >= total*share && m.at.Milliseconds() >= from { + return m.at + } + } + return f.flow[len(f.flow)-1].at + } + span := at(90) - at(10) + return max(span.Milliseconds()*10/8, 1) +} + // ms are the two as milliseconds from start, 0 for none. func (f *firstToken) ms() (first, text int64) { return f.first.Milliseconds(), f.text.Milliseconds() @@ -196,3 +277,13 @@ func filled(raw json.RawMessage) bool { } return true } + +// flowFor is flowMs from where a reply's speed is counted from, as +// usage.DecodeOf counts it: its first text when it reasoned, else its +// first content. +func (f *firstToken) flowFor(reasoning int) int64 { + if reasoning > 0 { + return f.flowMs(f.text.Milliseconds()) + } + return f.flowMs(f.first.Milliseconds()) +} diff --git a/internal/gateway/ttft_test.go b/internal/gateway/ttft_test.go index 09d24a22..6925b81f 100644 --- a/internal/gateway/ttft_test.go +++ b/internal/gateway/ttft_test.go @@ -4,6 +4,7 @@ import ( "io" "net/http" "net/http/httptest" + "strconv" "strings" "testing" "time" @@ -231,7 +232,8 @@ func TestFirstTokenKinds(t *testing.T) { for i := 0; i < len(stream); i += 7 { f.see([]byte(stream[i:min(i+7, len(stream))])) } - if first, text := f.ms(); first < 1000 || text < first || !f.off || f.pend != nil { + // read on past its first text, for when the rest came (flowMs) + if first, text := f.ms(); first < 1000 || text < first || f.off || f.pend != nil || f.flowFor(0) < 1 { t.Errorf("split: %d %d %+v", first, text, f) } // a JSON reply is not read @@ -259,3 +261,135 @@ func TestCodexBackendTimesFirstToken(t *testing.T) { t.Errorf("usage %+v", u) } } + +// bursty is a vendor that writes each of its chunks a gap after the last, +// every event of a chunk in one write: a chunk of many is a reply it held +// back and let go at once. +type bursty struct { + chunks [][]string + gap time.Duration + header bool // no Content-Type, as the ChatGPT backend sends +} + +func (b *bursty) ServeHTTP(w http.ResponseWriter, r *http.Request) { + io.ReadAll(r.Body) + if b.header { + w.Header()["Content-Type"] = nil + } else { + w.Header().Set("Content-Type", "text/event-stream") + } + f := w.(http.Flusher) + for i, c := range b.chunks { + if i > 0 { + time.Sleep(b.gap) + } + io.WriteString(w, strings.Join(c, "\n\n")+"\n\n") + f.Flush() + } +} + +// burstChunks: a reply's first word at once, then its out tokens' worth +// of words n events in each later chunk, then its end with the usage. +func burstChunks(proto provider.Protocol, words, per, out int) [][]string { + ev := func(word string) string { + if proto == provider.Responses { + return "event: response.output_text.delta\ndata: " + `{"type":"response.output_text.delta","item_id":"msg_1","output_index":0,"content_index":0,"delta":"` + word + `"}` + } + return `data: {"id":"c1","model":"m1","choices":[{"index":0,"delta":{"content":"` + word + `"}}]}` + } + chunks := [][]string{{ev("hi")}} + for i := 1; i < words; i += per { + var c []string + for j := i; j < min(i+per, words); j++ { + c = append(c, ev(" word")) + } + chunks = append(chunks, c) + } + end := []string{`data: {"id":"c1","model":"m1","choices":[{"index":0,"delta":{},"finish_reason":"stop"}],"usage":{"prompt_tokens":5,"completion_tokens":` + strconv.Itoa(out) + `}}`, `data: [DONE]`} + if proto == provider.Responses { + end = []string{"event: response.completed\ndata: " + `{"type":"response.completed","response":{"id":"r1","status":"completed","output":[],"usage":{"input_tokens":5,"output_tokens":` + strconv.Itoa(out) + `}}}`} + } + last := &chunks[len(chunks)-1] + *last = append(*last, end...) + return chunks +} + +// A reply whose first word came at once and the rest only after a wait, +// all in one burst, wrote nothing over the wait, and tells no speed (John +// on Discord: a Kimi Code reply to OpenCode read 1,367 tok/s): before, +// its 1,000 tokens over the 600 ms from its first word to its end read +// about 1,600 tok/s. One that streamed steadily keeps its speed, over +// nearly all of its time from its first word. +func TestBurstTellsNoSpeed(t *testing.T) { + const gap = 600 * time.Millisecond + for _, c := range []struct { + name string + proto provider.Protocol + path string + body string + }{ + {"chat", provider.Chat, "/v1/chat/completions", `{"model":"fake/m1","stream":true,"messages":[{"role":"user","content":"hi"}]}`}, + {"responses from chat", provider.Chat, "/v1/responses", `{"model":"fake/m1","stream":true,"input":"hi"}`}, + } { + for _, steady := range []bool{false, true} { + name := c.name + " burst" + chunks, g := burstChunks(c.proto, 200, 199, 1000), gap + if steady { + name = c.name + " steady" + chunks, g = burstChunks(c.proto, 11, 1, 200), gap/10 + } + t.Run(name, func(t *testing.T) { + fresh(t) + up := httptest.NewServer(&bursty{chunks: chunks, gap: g}) + t.Cleanup(up.Close) + p := provider.Provider{ID: "fake", Name: "Fake", Key: "k", Models: []string{"m1"}, Chat: up.URL + "/v1"} + if err := provider.Save(p); err != nil { + t.Fatal(err) + } + s := New() + rec := httptest.NewRecorder() + s.Handler().ServeHTTP(rec, httptest.NewRequest("POST", c.path, strings.NewReader(c.body))) + if rec.Code != 200 || !strings.Contains(rec.Body.String(), "word") { + t.Fatalf("%d %s", rec.Code, rec.Body) + } + u := lastUsage(t) + span := u.Millis - u.TTFT + if u.TTFT <= 0 || span < gap.Milliseconds() { + t.Fatalf("usage ttft %d, ms %d; want the end %v after the first word", u.TTFT, u.Millis, gap) + } + r := lastRoute(s) + if len(r.Tries) != 1 || r.Flow != u.Flow || r.Tries[0].Flow != u.Flow { + t.Errorf("flow: usage %d, route %d, tries %+v", u.Flow, r.Flow, r.Tries) + } + n, w := u.Decode() + if !steady { + if n != 0 || w != 0 { + t.Errorf("burst: %d tokens in %d ms (flow %d, %d ms from the first word) read %d tok/s; want no speed", n, w, u.Flow, span, int64(n)*1000/max(w, 1)) + } + return + } + if n != 200 || w < span*7/10 || w > span { + t.Errorf("steady: %d tokens in %d ms (flow %d); want 200 over most of the %d ms from the first word", n, w, u.Flow, span) + } + }) + } + } +} + +// Codex's own models, relayed to the ChatGPT backend as they came, tell no +// speed for a burst either. +func TestCodexBackendBurstTellsNoSpeed(t *testing.T) { + setup(t, provider.Chat, &fake{t: t}) + b := &bursty{chunks: burstChunks(provider.Responses, 200, 199, 1000), gap: 600 * time.Millisecond, header: true} + chatgpt(t, b.ServeHTTP) + if code, body := codexPost(t, `{"model":"gpt-5.5","stream":true,"input":"hi"}`); code != 200 { + t.Fatalf("%d %s", code, body) + } + u := lastUsage(t) + if u.Millis-u.TTFT < 600 || u.Output != 1000 { + t.Fatalf("usage %+v", u) + } + if n, w := u.Decode(); n != 0 || w != 0 { + t.Errorf("burst: %d tokens in %d ms (flow %d) read a speed; want none", n, w, u.Flow) + } +} diff --git a/internal/gui/assets/app.js b/internal/gui/assets/app.js index c7bf4404..af7d10f6 100644 --- a/internal/gui/assets/app.js +++ b/internal/gui/assets/app.js @@ -15583,11 +15583,13 @@ const LED_COLS = [ // the tokens a reply was seen to write and the ms it took, as // usage.DecodeOf and routing.js's decodeOf tell them: one that reasoned // counts its answer from its first text, its reasoning written before -// the stream showed any (tony on Discord); null when it tells no speed +// the stream showed any (tony on Discord); null when it tells no speed. +// Its window is no longer than its content took to come (flow_ms): a +// reply held back and sent in one burst tells none (John on Discord) const ledDecode = (r) => { if (ledFailed_(r) || !(r.ttft_ms > 0)) return null; const think = r.reasoning > 0, n = think ? r.out - r.reasoning : r.out, from = think ? r.first_text_ms : r.ttft_ms; - const w = r.ms - from; + const w = r.flow_ms > 0 ? Math.min(r.ms - from, r.flow_ms) : r.ms - from; return n > 0 && from > 0 && w >= 100 && n * 1000 <= 10000 * w ? { n, w } : null; }; // how fast a reply wrote, in tokens a second: 0 when it can't tell diff --git a/internal/gui/assets/routing.js b/internal/gui/assets/routing.js index 1ccc7154..a13fcd6d 100644 --- a/internal/gui/assets/routing.js +++ b/internal/gui/assets/routing.js @@ -257,14 +257,16 @@ // the reasoning was written before the stream showed any, OpenAI's // encrypted and its summary sent when done, and counting it read // gpt-6.1-sol at 163 tok/s. The reply's reasoning is its served try's. - function decodeOf(r, ms, ttft, firstText) { + // The window is no longer than its content took to come (flow): a reply + // held back and sent in one burst tells no speed (John on Discord). + function decodeOf(r, ms, ttft, firstText, flow = r.flow) { const think = reasoningOf(r), n = think > 0 ? r.out - think : r.out, from = think > 0 ? firstText : ttft; - const w = ms - from; + const w = flow > 0 ? Math.min(ms - from, flow) : ms - from; return n > 0 && ttft > 0 && from > 0 && w >= 100 && n * 1000 <= 10000 * w ? { n, w } : null; } const reasoningOf = (r) => r.reasoning ?? (r.usage?.length ? r.usage[r.usage.length - 1].reasoning || 0 : 0); - const speedOf = (r, ms = r.ms, ttft = r.ttft, firstText = r.firstText) => { - const d = decodeOf(r, ms, ttft, firstText); + const speedOf = (r, ms = r.ms, ttft = r.ttft, firstText = r.firstText, flow = r.flow) => { + const d = decodeOf(r, ms, ttft, firstText, flow); return d ? d.n / (d.w / 1000) : 0; }; function promptOf(r) { @@ -326,7 +328,7 @@ function firstNote(r, tr) { let s = tr.ttft ? " · " + t("first token in {ms}", { ms: took(tr.ttft) }) : ""; if (tr.ttft && tr.firstText > tr.ttft) s += " · " + t("first text in {ms}", { ms: took(tr.firstText) }); - const v = speedOf(r, tr.ms, tr.ttft, tr.firstText); + const v = speedOf(r, tr.ms, tr.ttft, tr.firstText, tr.flow); if (v) s += " · " + t("{n} tok/s", { n: Math.round(v) }); const { prompt, read } = promptOf(r); if (prompt) s += " · " + t("request cache hit rate {p}", { p: pct(100 * read / prompt) }); diff --git a/internal/gui/tests/routing-burst-speed.test.cjs b/internal/gui/tests/routing-burst-speed.test.cjs index 4c222de6..e7dd4b53 100644 --- a/internal/gui/tests/routing-burst-speed.test.cjs +++ b/internal/gui/tests/routing-burst-speed.test.cjs @@ -17,11 +17,14 @@ const at = (i) => new Date(now.getTime() - (i + 1) * 60e3).toISOString(); const key = { id: "antigravity", provider: "antigravity", name: "Antigravity", kind: "provider", model: "gemini-3.8-flash" }; // newest first: the report's burst, a 500 ms burst, then ordinary replies // (100 tok/s and 50 tok/s; a few rows, so the stage above holds still) -const timing = [[8264, 24360, 24359], [8264, 1500, 1000], [1200, 15000, 3000], [100, 3000, 1000], [100, 3000, 1000], [100, 3000, 1000]]; -const routes = timing.map(([out, ms, ttft], i) => ({ +// and last, a reply whose first words came at 5.1 s and the rest held back +// and sent at once (John on Discord: Kimi Code, 1,367 tok/s): its flow says +// its content took 2 ms to come +const timing = [[8264, 24360, 24359], [8264, 1500, 1000], [1200, 15000, 3000], [100, 3000, 1000], [100, 3000, 1000], [100, 3000, 1000], [2187, 6700, 5100, 2]]; +const routes = timing.map(([out, ms, ttft, flow], i) => ({ id: 100 - i, seq: 100 - i, time: at(i), agent: "codex", model: "antigravity/gemini-3.8-flash", provider: "antigravity", - order: [key], tries: [{ id: key.id, model: key.model, start: at(i), done: true, status: 200, ms, ttft }], - done: true, status: 200, ms, ttft, tokens: out + 4087, out, + order: [key], tries: [{ id: key.id, model: key.model, start: at(i), done: true, status: 200, ms, ttft, flow }], + done: true, status: 200, ms, ttft, flow, tokens: out + 4087, out, })); function serve(lang) { @@ -97,6 +100,10 @@ for (const engine of (process.env.BROWSER ? [process.env.BROWSER] : ["chromium", // ordinary replies keep theirs assert.match(await story(2), want[lang].normal); assert.match(await story(3), want[lang].slow); + // a burst after its first words: no speed, not 1,367 tok/s + const held = await story(timing.length - 1); + assert.doesNotMatch(held, want[lang].speed); + assert.doesNotMatch(held, /1367|1,367/); assert.deepEqual(errors, []); }); } diff --git a/internal/gui/tests/usage-speed.test.cjs b/internal/gui/tests/usage-speed.test.cjs index e0752d89..1f8e9f7e 100644 --- a/internal/gui/tests/usage-speed.test.cjs +++ b/internal/gui/tests/usage-speed.test.cjs @@ -30,6 +30,9 @@ const row = (i, model, out, ttft) => ({ const ROWS = [row(0, "glm-fast", 135, 600), row(1, "kimi-slow", 20, 600), row(2, "glm-fast", 135, 600)]; ROWS.push({ ...row(3, "glm-fast", 135, 0), ttft_ms: undefined, ms: 3000 }); for (let i = 4; i < 30; i++) ROWS.push(row(i, "glm-fast", 135, 600)); +// one whose content came in a burst (flow_ms, John on Discord: a Kimi Code +// reply read 1,367 tok/s): its tokens over the wait tell no speed +ROWS[4].flow_ms = 2; const FAST = { id: "glm-fast", calls: 28, input: 1, output: 3780, cache_read: 0, cache_write: 0, cost: 0.3, timed: 27, ttft_ms: 27 * 600, decode_ms: 27 * 1800, decode_out: 27 * 135 }; const SLOW = { id: "kimi-slow", calls: 1, input: 1, output: 20, cache_read: 0, cache_write: 0, cost: 0.01, timed: 1, ttft_ms: 600, decode_ms: 2000, decode_out: 20 }; @@ -119,6 +122,7 @@ for (const engine of (process.env.BROWSER ? [process.env.BROWSER] : ["chromium", assert.equal((await cell(0).textContent()).trim(), w.fast); assert.equal((await cell(1).textContent()).trim(), w.slow); assert.equal((await cell(3).textContent()).trim(), "—"); + assert.equal((await cell(4).textContent()).trim(), "—"); // the table fits the window; what it leaves out is in the details const wrap = p.locator("#ledWrap"); diff --git a/internal/usage/request_page.go b/internal/usage/request_page.go index df42ad31..3eb887bd 100644 --- a/internal/usage/request_page.go +++ b/internal/usage/request_page.go @@ -59,7 +59,7 @@ type packedRow struct { Text [28]uint32 Tokens [6]int64 Millis, TTFT, FirstText, Order int64 - Sent int64 + Sent, Flow int64 RouteID int64 Cost float64 Status int32 @@ -101,7 +101,7 @@ func (c *rowChunk) add(r Row, msg string, order int64, failed bool) { c.Bytes += int64(len(s) + 48) return id } - p := packedRow{Time: r.Time, Tokens: [6]int64{int64(r.Input), int64(r.Output), int64(r.CacheRead), int64(r.CacheWrite), int64(r.Reasoning), int64(r.CacheWrite1h)}, Millis: r.Millis, TTFT: r.TTFT, FirstText: r.FirstText, Sent: r.Sent, Order: order, RouteID: r.RouteID, Cost: r.Cost, Status: int32(r.Status)} + p := packedRow{Time: r.Time, Tokens: [6]int64{int64(r.Input), int64(r.Output), int64(r.CacheRead), int64(r.CacheWrite), int64(r.Reasoning), int64(r.CacheWrite1h)}, Millis: r.Millis, TTFT: r.TTFT, FirstText: r.FirstText, Sent: r.Sent, Flow: r.Flow, Order: order, RouteID: r.RouteID, Cost: r.Cost, Status: int32(r.Status)} for i, s := range rowText(&r) { p.Text[i] = intern(*s) } @@ -133,7 +133,7 @@ func (c *rowChunk) add(r Row, msg string, order int64, failed bool) { } func (c *rowChunk) row(i int) Row { p := &c.Rows[i] - r := Row{Record: Record{RouteID: p.RouteID, Time: p.Time, Input: int(p.Tokens[0]), Output: int(p.Tokens[1]), CacheRead: int(p.Tokens[2]), CacheWrite: int(p.Tokens[3]), CacheWrite1h: int(p.Tokens[5]), Reasoning: int(p.Tokens[4]), Millis: p.Millis, TTFT: p.TTFT, FirstText: p.FirstText, Sent: p.Sent, Status: int(p.Status), Rejected: p.Flags&4 != 0, SessionOfficialLogin: p.Flags&8 != 0}, Cost: p.Cost, Priced: p.Flags&1 != 0, Swapped: p.Flags&2 != 0, Routed: p.Flags&32 != 0} + r := Row{Record: Record{RouteID: p.RouteID, Time: p.Time, Input: int(p.Tokens[0]), Output: int(p.Tokens[1]), CacheRead: int(p.Tokens[2]), CacheWrite: int(p.Tokens[3]), CacheWrite1h: int(p.Tokens[5]), Reasoning: int(p.Tokens[4]), Millis: p.Millis, TTFT: p.TTFT, FirstText: p.FirstText, Sent: p.Sent, Flow: p.Flow, Status: int(p.Status), Rejected: p.Flags&4 != 0, SessionOfficialLogin: p.Flags&8 != 0}, Cost: p.Cost, Priced: p.Flags&1 != 0, Swapped: p.Flags&2 != 0, Routed: p.Flags&32 != 0} for i, s := range rowText(&r) { *s = c.Strings[p.Text[i]] } diff --git a/internal/usage/request_snapshot.go b/internal/usage/request_snapshot.go index e3c8bda3..72ca1df9 100644 --- a/internal/usage/request_snapshot.go +++ b/internal/usage/request_snapshot.go @@ -50,7 +50,7 @@ func (c *rowChunk) pack() *rowChunk { for _, n := range p.Tokens { put(uint64(n)) } - for _, n := range []int64{p.Millis, p.TTFT, p.FirstText, p.Order, p.RouteID, p.Sent} { + for _, n := range []int64{p.Millis, p.TTFT, p.FirstText, p.Order, p.RouteID, p.Sent, p.Flow} { put(uint64(n)) } put(math.Float64bits(p.Cost)) @@ -131,7 +131,7 @@ func (c *rowChunk) unpack() *rowChunk { } p.Millis, p.TTFT, p.FirstText, p.Order = int64(take()), int64(take()), int64(take()), int64(take()) p.RouteID = int64(take()) - p.Sent = int64(take()) + p.Sent, p.Flow = int64(take()), int64(take()) p.Cost, p.Status, p.Flags = math.Float64frombits(take()), int32(take()), uint8(take()) } if !valid || len(data) != 0 { diff --git a/internal/usage/usage.go b/internal/usage/usage.go index ab3d5ff2..aa498961 100644 --- a/internal/usage/usage.go +++ b/internal/usage/usage.go @@ -82,7 +82,14 @@ type Record struct { // the tries that failed first are before it, and TTFT-Sent is how // long the vendor took to its first content. 0 where it isn't known // (a reply not streamed, or a vendor magpie doesn't reach over HTTP). - Sent int64 `json:"sent_ms,omitempty"` + Sent int64 `json:"sent_ms,omitempty"` + // Flow: the ms the reply's content took to come, from where its speed + // is counted (DecodeOf): from a tenth of its content to nine tenths, + // spread over the whole. A reply a vendor held and let go in one burst + // has a flow of next to none, however long its end came after its + // first content, and so tells no speed (John on Discord: a Kimi Code + // reply read 1,367 tok/s). 0 where it isn't known. + Flow int64 `json:"flow_ms,omitempty"` Status int `json:"status"` // Error is why a call failed, in the vendor's words and cut short; // ErrType what its body called the error (rate_limit_error, @@ -345,8 +352,12 @@ func DecodeWindow(out int, ms, ttft int64) int64 { // hold, and counting them read a 2,000-token think and a 200-token answer // in 1.5 s as 1,467 tok/s. Its time goes to the wait before the answer. // One with reasoning and no text (all tool calls) tells no speed. The -// window is DecodeWindow's: 0 for a reply that tells none. -func DecodeOf(out, reasoning int, ms, ttft, firstText int64) (tokens int, w int64) { +// window is DecodeWindow's: 0 for a reply that tells none. flow, when +// known (Record.Flow), is how long its content took to come, and the +// window is no longer than it: a reply whose first words came at once +// and the rest held back and sent in a burst at its end wrote nothing +// in between, and its tokens over the wait read an impossible speed. +func DecodeOf(out, reasoning int, ms, ttft, firstText, flow int64) (tokens int, w int64) { start := ttft if reasoning > 0 { out, start = out-reasoning, firstText @@ -354,6 +365,9 @@ func DecodeOf(out, reasoning int, ms, ttft, firstText int64) (tokens int, w int6 if ttft <= 0 { return 0, 0 } + if flow > 0 && start > 0 && start+flow < ms { + ms = start + flow + } if w = DecodeWindow(out, ms, start); w == 0 { return 0, 0 } @@ -362,7 +376,7 @@ func DecodeOf(out, reasoning int, ms, ttft, firstText int64) (tokens int, w int6 // Decode is the record's DecodeOf. func (r Record) Decode() (tokens int, w int64) { - return DecodeOf(r.Output, r.Reasoning, r.Millis, r.TTFT, r.FirstText) + return DecodeOf(r.Output, r.Reasoning, r.Millis, r.TTFT, r.FirstText, r.Flow) } // FormatCost renders an effective-price cost, kept in USD everywhere it's