diff --git a/docs/guardrails.md b/docs/guardrails.md index 42a1662..c51c505 100644 --- a/docs/guardrails.md +++ b/docs/guardrails.md @@ -309,7 +309,10 @@ service can stop a streamed answer only where it is held back it by then, and what the service finds ends the stream with `content_filter`, flags it and is recorded. The rule recorded is `service::`. The service is told the text only, never who -sent it. +sent it. Under `fail: open`, the default, a service that cannot be asked +lets the text through — and a caller can make it fail, sending what it +refuses to read; give `fail: closed` where the service must decide. Of an +answer asked for with `n` above 1, the service reads the first. ## Where a policy applies diff --git a/internal/agent/engines/unsafeenv.go b/internal/agent/engines/unsafeenv.go index 32fad5b..49079f1 100644 --- a/internal/agent/engines/unsafeenv.go +++ b/internal/agent/engines/unsafeenv.go @@ -220,7 +220,8 @@ func hubID(s string) bool { if s == "" || strings.HasPrefix(s, "/") || strings.HasPrefix(s, ".") || strings.Contains(s, "://") { return false } - if strings.Count(s, "/") != 1 { + // org/name:Q4_K_M is a GGUF in a repository, which vLLM reads as one. + if strings.Count(s, "/") != 1 || strings.Contains(s, ":") { return false } _, err := os.Stat(s) diff --git a/internal/guard/policy.go b/internal/guard/policy.go index 7046255..4b557a5 100644 --- a/internal/guard/policy.go +++ b/internal/guard/policy.go @@ -220,10 +220,21 @@ func Merge(names []string, ps []Policy) Policy { out.RequireCapabilities = out.RequireCapabilities || p.RequireCapabilities out.ValidateStructured = out.ValidateStructured || p.ValidateStructured out.Judge = mergeJudge(out.Judge, p.Judge) - // Every service any policy asks, each once. + // Every service any policy asks, each once; one named by two is + // asked the stricter way of the two: both directions either asks, + // blocking where either blocks. for _, sv := range p.Services { - if !slices.ContainsFunc(out.Services, func(e Service) bool { return e.Name == sv.Name }) { + i := slices.IndexFunc(out.Services, func(e Service) bool { return e.Name == sv.Name }) + if i < 0 { out.Services = append(out.Services, sv) + continue + } + e := &out.Services[i] + if e.action() == Flag && sv.action() == Block { + e.Action = "block" + } + if e.onInput() != sv.onInput() || e.onOutput() != sv.onOutput() { + e.On = "both" } } out.MaxOutputTokens = minSet(out.MaxOutputTokens, p.MaxOutputTokens) diff --git a/internal/guard/service.go b/internal/guard/service.go index 7985bec..b517640 100644 --- a/internal/guard/service.go +++ b/internal/guard/service.go @@ -144,7 +144,8 @@ func (s Service) ask(ctx context.Context, text string) (unsafe bool, reason stri return false, "", fmt.Errorf("answer is not JSON: %w", err) } v, ok := at(doc, s.Blocked) - if !ok { + if !ok || v == nil { + // null is no answer: a service that cannot say is not saying safe. return false, "", fmt.Errorf("the answer has no %s", s.Blocked) } if s.Reason != "" { diff --git a/internal/guard/service_test.go b/internal/guard/service_test.go index af404e5..490eb66 100644 --- a/internal/guard/service_test.go +++ b/internal/guard/service_test.go @@ -139,3 +139,12 @@ func TestAStreamedAnswerIsReadByAServiceOnce(t *testing.T) { t.Errorf("the service was asked %d times: %q", len(*seen), *seen) } } + +func TestAServiceNamedByTwoPoliciesIsAskedTheStricterWay(t *testing.T) { + a := Policy{Services: []Service{{Name: "cls", URL: "https://c", Blocked: "x", On: "input", Action: "flag"}}} + b := Policy{Services: []Service{{Name: "cls", URL: "https://c", Blocked: "x", On: "output", Action: "block"}}} + got := Merge([]string{"a", "b"}, []Policy{a, b}).Services + if len(got) != 1 || got[0].action() != Block || !got[0].onInput() || !got[0].onOutput() { + t.Errorf("%+v", got) + } +} diff --git a/internal/router/handlers.go b/internal/router/handlers.go index a4e8ca8..3689369 100644 --- a/internal/router/handlers.go +++ b/internal/router/handlers.go @@ -286,6 +286,8 @@ func (r *Router) Handler() http.Handler { if r.mcpgw == nil { return nil } + r.mcpgw.mu.Lock() + defer r.mcpgw.mu.Unlock() return r.mcpgw.local }) } diff --git a/internal/router/requestlog.go b/internal/router/requestlog.go index a1d253c..d9f7803 100644 --- a/internal/router/requestlog.go +++ b/internal/router/requestlog.go @@ -6,6 +6,7 @@ import ( "fmt" "hash/fnv" "net/http" + "strings" "sync" "time" @@ -132,7 +133,7 @@ func (r *Router) keepRequest(ctx context.Context, req *parsedRequest, res dispat now := time.Now() b := storepkg.RequestBody{RequestID: req.RequestID, ProjectID: req.ProjectID, At: now, ExpiresAt: now.Add(req.logRetain), Path: req.Path, Model: req.Model, ServedModel: served, - Status: status, Request: string(req.UpstreamBody), Response: res.outcome.Reply, + Status: status, Request: string(req.UpstreamBody), Response: withoutRestored(res.outcome.Reply, req.guardIn.Restore), Truncated: res.outcome.ReplyTruncated} if err := r.store.PutRequestBody(ctx, b); err != nil { r.log.Warn("could not keep a request", "request_id", req.RequestID, "err", err) @@ -192,3 +193,15 @@ func (r *Router) Replay(ctx context.Context, path, model string, body []byte) (i } return status, w.buf.Bytes(), took, nil } + +// withoutRestored is an answer with the values its request had redacted +// taken out again: the request is kept as the model read it, redacted, and +// the answer kept beside it must not hold what that took out. +func withoutRestored(reply string, restore map[string]string) string { + for ph, v := range restore { + if v != "" { + reply = strings.ReplaceAll(reply, v, ph) + } + } + return reply +} diff --git a/internal/router/restore_test.go b/internal/router/restore_test.go index 80f5800..def1dd9 100644 --- a/internal/router/restore_test.go +++ b/internal/router/restore_test.go @@ -212,3 +212,9 @@ func TestHeldTextGoesOutWithTheFinishEvent(t *testing.T) { t.Errorf("%s / %s", out1, out2) } } + +func TestAKeptAnswerHoldsNoValueItsRequestHadRedacted(t *testing.T) { + if got := withoutRestored("Sent to alice@example.com.", map[string]string{"[EMAIL_1]": "alice@example.com"}); got != "Sent to [EMAIL_1]." { + t.Errorf("%q", got) + } +} diff --git a/internal/router/router.go b/internal/router/router.go index 045f7f2..84f01f5 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -94,6 +94,8 @@ type Router struct { // cacheEmbedModel embeds the questions of projects that reuse answers // to similar ones; empty, they reuse only the same request's. cacheEmbedModel string + // embedWarned is when that model's failing was last logged. + embedWarned lastWarned // extraMetrics are collectors other components publish on /metrics. extraMetrics []prometheus.Collector @@ -1378,3 +1380,18 @@ func loopbackOnly(listen string) bool { a, err := netip.ParseAddr(host) return err == nil && a.IsLoopback() } + +// lastWarned is when something was last logged, for logging it seldom. +type lastWarned struct { + mu sync.Mutex + at time.Time +} + +// Swap records now and says when it was before. +func (w *lastWarned) Swap(now time.Time) time.Time { + w.mu.Lock() + defer w.mu.Unlock() + was := w.at + w.at = now + return was +} diff --git a/internal/router/similar.go b/internal/router/similar.go index e919604..4e1c011 100644 --- a/internal/router/similar.go +++ b/internal/router/similar.go @@ -53,8 +53,11 @@ func (r *Router) similarLookup(ctx context.Context, req *parsedRequest, model st } vec, err := r.embedQuestion(ctx, question, req.inCluster) if err != nil { - r.log.Warn("a question could not be embedded for the cache; only the same request is reused", - "model", r.cacheEmbedModel, "err", err) + // Once a minute at most: down, the model fails every miss. + if now := time.Now(); now.Sub(r.embedWarned.Swap(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) + } return nil, 0 } if a, sim := r.answers.nearest(group, vec, set.Similar, set.TTL, time.Now()); a != nil && a.project == req.ProjectID {