Outside services read answers redacted and whole; a card's own lanes

From the second review.
- An output service reads the answer with the values restore put back
  taken out again (guard.Subject.Restored), as an input service reads the
  request redacted.
- With n above 1, the service reads the first choice once it is whole,
  not at whichever choice finished first.
- Restored values go into an answer asked for as JSON (response_format)
  escaped for it, as into tool-call arguments.
- The embedding failure is logged once a minute through an outage, not
  once.
- A card's own lane count and fastest generation come from the kernel
  (max_link_width, max_link_speed): the driver's width.max is what card
  and slot allow together, so an x16 card in an x8 slot read x8 of x8 and
  was never said.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

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,

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)
 		}

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 {

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())
+	}
+}

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.

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

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)
+	}
+}

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
 	}

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")
+}

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")
+	}
+}

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
 }

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)
 		}