diff --git a/CHANGELOG.md b/CHANGELOG.md index ff6202cd..f9941ce4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -113,7 +113,8 @@ APIs change between minors without a deprecation period. - **A card in the wrong PCIe slot is said so.** Each agent reads how every card is linked — the fastest generation card and slot allow together, and - the lanes in use — against the card's own most; a card on fewer lanes or + the lanes in use — against the card's own most, from the kernel (the + driver's figure is what card and slot allow together); a card on fewer lanes or an older generation is a node warning, with what it carries and what it could, on the dashboard, the console and in `llmgw doctor`. - **And on a web page, `/admin/mcp`.** For administrators: the same list, diff --git a/internal/agent/discovery/gpu_nvml.go b/internal/agent/discovery/gpu_nvml.go index 65529e2f..086d3f6c 100644 --- a/internal/agent/discovery/gpu_nvml.go +++ b/internal/agent/discovery/gpu_nvml.go @@ -83,6 +83,7 @@ func DiscoverGPUsNVML(_ context.Context) ([]GPU, error) { g.PCIBusID = busIDString(pci.BusId[:]) byBusID[normalizeBusID(g.PCIBusID)] = i } + g.PCIe = withSysfs(g.PCIe, "/sys/bus/pci/devices", g.PCIBusID) if t, ret := dev.GetTemperature(nvml.TEMPERATURE_GPU); ret == nvml.SUCCESS { g.Health.TemperatureC = int(t) } diff --git a/internal/agent/discovery/pcie.go b/internal/agent/discovery/pcie.go index 721e0671..125a617f 100644 --- a/internal/agent/discovery/pcie.go +++ b/internal/agent/discovery/pcie.go @@ -4,7 +4,9 @@ import ( "context" "encoding/csv" "fmt" + "os" "os/exec" + "path/filepath" "strconv" "strings" ) @@ -84,8 +86,71 @@ func readPCIe(ctx context.Context, binary string, cards []GPU) { cards[i].PCIe = l } } - return + break } + for i := range cards { + if cards[i].MIG == nil { + cards[i].PCIe = withSysfs(cards[i].PCIe, "/sys/bus/pci/devices", cards[i].PCIBusID) + } + } +} + +// withSysfs takes the card's own lane count, and its own fastest +// generation, from the kernel: the driver's width.max is what card and slot +// allow together, so a x16 card in a slot wired x8 reads x8 of x8 there, +// while the device's max_link_width is the card's, 16. +func withSysfs(l PCIeLink, root, busID string) PCIeLink { + dev := sysfsBusID(busID) + if dev == "" { + return l + } + read := func(name string) string { + b, err := os.ReadFile(filepath.Join(root, dev, name)) //nolint:gosec // under /sys + if err != nil { + return "" + } + return strings.TrimSpace(string(b)) + } + if n, err := strconv.Atoi(read("max_link_width")); err == nil && n > 0 { + l.WidthMax = n + } + if n, err := strconv.Atoi(read("current_link_width")); err == nil && n > 0 { + l.Width = n + } + if g := genOfSpeed(read("max_link_speed")); g > 0 && l.GPUGen == 0 { + l.GPUGen = g + } + return l +} + +// sysfsBusID is nvidia-smi's 00000000:3B:00.0 as the kernel names it, +// 0000:3b:00.0. +func sysfsBusID(id string) string { + id = strings.ToLower(strings.TrimSpace(id)) + dom, rest, ok := strings.Cut(id, ":") + if !ok || len(dom) < 4 { + return "" + } + return dom[len(dom)-4:] + ":" + rest +} + +// genOfSpeed is the PCIe generation of a link speed as sysfs writes it. +func genOfSpeed(s string) int { + switch { + case strings.HasPrefix(s, "2.5 GT/s"): + return 1 + case strings.HasPrefix(s, "5.0 GT/s"), strings.HasPrefix(s, "5 GT/s"): + return 2 + case strings.HasPrefix(s, "8.0 GT/s"), strings.HasPrefix(s, "8 GT/s"): + return 3 + case strings.HasPrefix(s, "16.0 GT/s"), strings.HasPrefix(s, "16 GT/s"): + return 4 + case strings.HasPrefix(s, "32.0 GT/s"), strings.HasPrefix(s, "32 GT/s"): + return 5 + case strings.HasPrefix(s, "64.0 GT/s"), strings.HasPrefix(s, "64 GT/s"): + return 6 + } + return 0 } func parsePCIe(out string) map[int]PCIeLink { diff --git a/internal/agent/discovery/pcie_test.go b/internal/agent/discovery/pcie_test.go index 6504ddf0..10b3a8f8 100644 --- a/internal/agent/discovery/pcie_test.go +++ b/internal/agent/discovery/pcie_test.go @@ -1,6 +1,8 @@ package discovery import ( + "os" + "path/filepath" "strings" "testing" ) @@ -35,3 +37,18 @@ func TestTheLinkIsReadFromNvidiaSMI(t *testing.T) { t.Errorf("%+v", links) } } + +func TestTheCardsOwnLanesComeFromTheKernel(t *testing.T) { + // The driver says x8 of x8 for a x16 card in a slot wired x8; the + // device's own max_link_width says 16. + root := t.TempDir() + dev := filepath.Join(root, "0000:3b:00.0") + _ = os.MkdirAll(dev, 0o755) + for f, v := range map[string]string{"max_link_width": "16", "current_link_width": "8", "max_link_speed": "16.0 GT/s PCIe"} { + _ = os.WriteFile(filepath.Join(dev, f), []byte(v+"\n"), 0o644) + } + l := withSysfs(PCIeLink{Gen: 4, Width: 8, WidthMax: 8}, root, "00000000:3B:00.0") + if l.WidthMax != 16 || l.Width != 8 || l.GPUGen != 4 || !strings.Contains(l.Problem(), "x8 of its x16") { + t.Errorf("%+v: %q", l, l.Problem()) + } +} diff --git a/internal/guard/guard.go b/internal/guard/guard.go index 163467b9..e0ccc8a9 100644 --- a/internal/guard/guard.go +++ b/internal/guard/guard.go @@ -54,6 +54,10 @@ type Subject struct { Stream bool // Tools and Vision are what the model's catalog entry says it can do. Tools, Vision bool + // Restored are the values the request's redaction took out, by + // placeholder, which restore puts back in the answer: outside services + // reading the answer are not told them. + Restored map[string]string } // Verdict is a judgement on one stage of one request. diff --git a/internal/guard/output.go b/internal/guard/output.go index bf50c365..2cbfa290 100644 --- a/internal/guard/output.go +++ b/internal/guard/output.go @@ -68,7 +68,7 @@ func (g *Gate) Output(ctx context.Context, s Subject) OutputFilter { return nil } memo := g.tools.get(s.RequestID) - return &outFilter{ctx: ctx, g: g, p: p, names: names, shadow: p.Mode == "shadow", + return &outFilter{ctx: ctx, g: g, p: p, names: names, shadow: p.Mode == "shadow", restored: s.Restored, schemas: memo.schemas, answer: memo.answer, user: memo.user, choices: map[int]*choiceState{}} } @@ -76,21 +76,24 @@ func (g *Gate) Output(ctx context.Context, s Subject) OutputFilter { type outFilter struct { // servicesAsked says the outside services have read the answer. servicesAsked bool - ctx context.Context - g *Gate - p Policy - names string - shadow bool - schemas map[string]json.RawMessage - answer json.RawMessage - full strings.Builder // the whole answer, when it must fit a schema - user string // the request's last user turn, for the judge - sofar strings.Builder // the answer so far, for the judge - served strings.Builder // the answer so far, for outside services, asked once at its end - judged int // how much of it the judge has read - v Verdict - took time.Duration - stopped bool + // restored are the request's redacted values, by placeholder, put back + // in the answer before this filter reads it. + restored map[string]string + ctx context.Context + g *Gate + p Policy + names string + shadow bool + schemas map[string]json.RawMessage + answer json.RawMessage + full strings.Builder // the whole answer, when it must fit a schema + user string // the request's last user turn, for the judge + sofar strings.Builder // the answer so far, for the judge + served strings.Builder // the answer so far, for outside services, asked once at its end + judged int // how much of it the judge has read + v Verdict + took time.Duration + stopped bool held []string // events not yet released sent bool // whether a chunk this filter made has gone out @@ -105,6 +108,8 @@ type choiceState struct { // replaced: it reaches the client only in a chunk this filter makes. carried bool calls map[int]*call + // done says the choice has finished: its text is all here. + done bool } type call struct { @@ -214,6 +219,13 @@ func (f *outFilter) askServices(text string) finding { if !slices.ContainsFunc(f.p.Services, Service.onOutput) || text == "" { return finding{} } + // As the model wrote it: with restore, the values put back for the + // client are taken out again, as redact took them out of the request. + for ph, v := range f.restored { + if v != "" { + text = strings.ReplaceAll(text, v, ph) + } + } start := time.Now() fs, err := f.g.askServices(f.ctx, f.p, false, text) f.took += time.Since(start) @@ -381,6 +393,7 @@ func (f *outFilter) OnEvent(data string) ([]string, bool) { } if ch.FinishReason != nil && *ch.FinishReason != "" { finished = true + st.done = true } } if f.shadow { @@ -447,9 +460,10 @@ func (f *outFilter) flush(final bool) (bool, map[int]string, bool) { } if i == 0 { f.served.WriteString(head) - // Once: the finish chunk is final, and so is the end of the - // stream after it, which would ask again of the same answer. - if final && !f.servicesAsked { + // Once, when the first choice is whole — not when another + // choice finishes first — and not again at the end of the + // stream after its finish. + if final && st.done && !f.servicesAsked { f.servicesAsked = true if f.askServices(f.served.String()).action == Block { blocked = true @@ -573,6 +587,10 @@ func withoutContent(data string) string { } func (f *outFilter) OnEnd() []string { + // The stream is over: whatever a choice has is all it will have. + for _, st := range f.choices { + st.done = true + } switch { case f.stopped: return nil diff --git a/internal/guard/service_test.go b/internal/guard/service_test.go index 490eb669..58f54b3d 100644 --- a/internal/guard/service_test.go +++ b/internal/guard/service_test.go @@ -148,3 +148,39 @@ func TestAServiceNamedByTwoPoliciesIsAskedTheStricterWay(t *testing.T) { t.Errorf("%+v", got) } } + +func TestAnOutputServiceReadsTheAnswerWithoutTheRestoredValues(t *testing.T) { + srv, seen := recorder(t) + g := gate(t, lib(map[string]Policy{"p": { + Services: []Service{{Name: "cls", URL: srv.URL, Blocked: "flagged", On: "output"}}, + }}, "p")) + f := g.Output(context.Background(), Subject{Restored: map[string]string{"[EMAIL_1]": "ayse@example.com"}}) + f.Buffered([]byte(`{"choices":[{"index":0,"message":{"role":"assistant","content":"Sent to ayse@example.com."},"finish_reason":"stop"}]}`)) + if len(*seen) != 1 || strings.Contains((*seen)[0], "ayse@") || !strings.Contains((*seen)[0], "[EMAIL_1]") { + t.Errorf("the service read %q", *seen) + } +} + +func TestOfTwoChoicesTheServiceReadsTheFirstWhole(t *testing.T) { + // The second finishes first; the first is still arriving then. + srv, seen := recorder(t) + g := gate(t, lib(map[string]Policy{"p": { + Services: []Service{{Name: "cls", URL: srv.URL, Blocked: "flagged", On: "output"}}, + }}, "p")) + f := g.Output(context.Background(), Subject{Stream: true}) + ev := func(idx int, text string, finish bool) string { + c := map[string]any{"index": idx, "delta": map[string]any{"content": text}, "finish_reason": nil} + if finish { + c["finish_reason"] = "stop" + } + b, _ := json.Marshal(map[string]any{"choices": []any{c}}) + return string(b) + } + for _, e := range []string{ev(0, "The first ", false), ev(1, "Second.", true), ev(0, "answer.", true)} { + f.OnEvent(e) + } + f.OnEnd() + if len(*seen) != 1 || (*seen)[0] != "The first answer." { + t.Errorf("the service read %q", *seen) + } +} diff --git a/internal/router/guard.go b/internal/router/guard.go index 1c4cb2e6..d8d1a241 100644 --- a/internal/router/guard.go +++ b/internal/router/guard.go @@ -35,6 +35,7 @@ func (p *parsedRequest) guardSubject(model string) guard.Subject { // can do. func (r *Router) subject(req *parsedRequest, model string) guard.Subject { s := req.guardSubject(model) + s.Restored = req.guardIn.Restore if _, entry := r.resolve(model); entry.Name != "" { s.Tools, s.Vision = entry.Capabilities.Tools, entry.Capabilities.Vision } diff --git a/internal/router/restore.go b/internal/router/restore.go index 1530dd9c..d7691c95 100644 --- a/internal/router/restore.go +++ b/internal/router/restore.go @@ -22,6 +22,10 @@ import ( type restorer struct { values map[string]string longest int + // jsonContent is an answer asked for as JSON (response_format): its + // content is JSON, and a value goes into it escaped, as into a tool + // call's arguments. + jsonContent bool // held is, per choice and field, the tail of the text so far that may // be the start of a placeholder the next event completes: "[EMA" before // "IL_1]". Keyed by heldKey. @@ -110,7 +114,7 @@ func (r *restorer) walk(v any, key string) any { x[i] = r.walk(e, key) } case string: - return r.text(x, key == "arguments") + return r.text(x, key == "arguments" || (r.jsonContent && key == "content")) } return v } @@ -164,7 +168,7 @@ func (r *restorer) OnEvent(data string) ([]string, bool) { } } if delta != nil { - swap(delta, "content", heldKey{idx, "content", -1}, false) + swap(delta, "content", heldKey{idx, "content", -1}, r.jsonContent) swap(delta, "reasoning_content", heldKey{idx, "reasoning_content", -1}, false) calls, _ := delta["tool_calls"].([]any) for j, tc := range calls { @@ -260,11 +264,12 @@ func (c streamChain) OnEnd() []string { // restoreInto puts the input's placeholders back in this attempt's answer, // ahead of whatever output filter the attempt already has. -func restoreInto(values map[string]string, stream bool, preq *proxy.Request) { +func restoreInto(values map[string]string, stream, jsonContent bool, preq *proxy.Request) { if len(values) == 0 { return } rs := newRestorer(values) + rs.jsonContent = jsonContent if f := preq.Filter; f != nil { preq.Filter = func(b []byte, h http.Header) []byte { return f(rs.body(b), h) } } else { @@ -279,3 +284,14 @@ func restoreInto(values map[string]string, stream bool, preq *proxy.Request) { preq.StreamFilter = rs } } + +// jsonAnswer reports whether a request asks for its answer as JSON. +func jsonAnswer(body []byte) bool { + var b struct { + ResponseFormat *struct { + Type string `json:"type"` + } `json:"response_format"` + } + return json.Unmarshal(body, &b) == nil && b.ResponseFormat != nil && + (b.ResponseFormat.Type == "json_object" || b.ResponseFormat.Type == "json_schema") +} diff --git a/internal/router/restore_test.go b/internal/router/restore_test.go index def1dd9c..c96b03e1 100644 --- a/internal/router/restore_test.go +++ b/internal/router/restore_test.go @@ -10,6 +10,7 @@ import ( "regexp" "strings" "testing" + "time" "gitea.mynodes.xyz/ilker/llmgw/internal/config" "gitea.mynodes.xyz/ilker/llmgw/internal/guard" @@ -218,3 +219,30 @@ func TestAKeptAnswerHoldsNoValueItsRequestHadRedacted(t *testing.T) { t.Errorf("%q", got) } } + +func TestAnAnswerAskedForAsJSONGetsItsValuesEscapedForIt(t *testing.T) { + r := newRestorer(map[string]string{"[NAME_1]": `O"Brien`}) + r.jsonContent = true + body := r.body([]byte(`{"choices":[{"message":{"content":"{\"name\":\"[NAME_1]\"}"}}]}`)) + var whole struct { + Choices []struct{ Message struct{ Content string } } + } + if err := json.Unmarshal(body, &whole); err != nil { + t.Fatal(err) + } + var obj map[string]string + if err := json.Unmarshal([]byte(whole.Choices[0].Message.Content), &obj); err != nil || obj["name"] != `O"Brien` { + t.Errorf("content %q (%v)", whole.Choices[0].Message.Content, err) + } + if !jsonAnswer([]byte(`{"response_format":{"type":"json_schema"}}`)) || jsonAnswer([]byte(`{"model":"m"}`)) { + t.Error("jsonAnswer") + } +} + +func TestAFailureIsWarnedOfOnceAMinuteNotOnce(t *testing.T) { + var w lastWarned + now := time.Now() + if !w.Due(now, time.Minute) || w.Due(now.Add(30*time.Second), time.Minute) || !w.Due(now.Add(61*time.Second), time.Minute) { + t.Error("a failure every 30 s is warned of once a minute") + } +} diff --git a/internal/router/router.go b/internal/router/router.go index 84f01f54..743094fd 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -1256,7 +1256,7 @@ func (r *Router) dispatchOne(ctx context.Context, w http.ResponseWriter, req *pa } // A filter per attempt, for the model this attempt is served by. filter := r.outputFilter(uctx, req, model, &preq) - restoreInto(req.guardIn.Restore, req.Stream, &preq) + restoreInto(req.guardIn.Restore, req.Stream, jsonAnswer(req.UpstreamBody), &preq) var out proxy.Outcome var perr error if dec.Endpoint.Engine == endpoint.EngineComfyUI { @@ -1387,11 +1387,14 @@ type lastWarned struct { at time.Time } -// Swap records now and says when it was before. -func (w *lastWarned) Swap(now time.Time) time.Time { +// Due says whether every has passed since it was last said, and if so +// records now as that time: once a minute through an outage, not once. +func (w *lastWarned) Due(now time.Time, every time.Duration) bool { w.mu.Lock() defer w.mu.Unlock() - was := w.at + if now.Sub(w.at) < every { + return false + } w.at = now - return was + return true } diff --git a/internal/router/similar.go b/internal/router/similar.go index 4e1c011a..47d6238f 100644 --- a/internal/router/similar.go +++ b/internal/router/similar.go @@ -54,7 +54,7 @@ func (r *Router) similarLookup(ctx context.Context, req *parsedRequest, model st vec, err := r.embedQuestion(ctx, question, req.inCluster) if err != nil { // Once a minute at most: down, the model fails every miss. - if now := time.Now(); now.Sub(r.embedWarned.Swap(now)) > time.Minute { + if r.embedWarned.Due(time.Now(), time.Minute) { r.log.Warn("a question could not be embedded for the cache; only the same request is reused", "model", r.cacheEmbedModel, "err", err) }