feat: Basic user level rate limits + redact known secret formats + basic model fallback mechanism

cmd/pathway/main.go

@@ -95,7 +95,7 @@ func run() error {
 		return err
 	}
 
-	meter := metering.New(store.Usage, store.Health, log.Process)
+	meter := metering.New(store.Usage, store.Limits, store.Health, log.Process)
 	if !cfg.IsIngress() {
 		// What the store refuses while it is down waits on disk and is replayed once it answers.
 		meter.Spool = metering.NewSpool(store.Usage, log.Process,
@@ -105,7 +105,7 @@ func run() error {
 	}
 
 	server := gatewayhttp.New(cfg, gatewayhttp.Services{
-		Admission: admission.New(store.Keys, store.Users, store.Groups),
+		Admission: admission.New(store.Keys, store.Users, store.Groups, store.Limits),
 		Routing: routing.New(store, log.Process, routing.Options{
 			Region:       cfg.Region,
 			SyntheticTTL: func() time.Duration { return live.Get().SyntheticSessionTTL },

go.mod

@@ -1,6 +1,6 @@
 module github.com/phot0n/pathway
 
-go 1.23
+go 1.24
 
 require (
 	github.com/cloudflare/tableflip v1.2.3

internal/config/dynamic.go

@@ -81,7 +81,7 @@ type Resolved struct {
 func Defaults() Resolved {
 	return Resolved{
 		LogLevel:            slog.LevelInfo,
-		Transforms:          []string{"modelmap", "streamusage", "servicetier", "maxtokens"},
+		Transforms:          []string{"modelmap", "streamusage", "servicetier", "vendorfields"},
 		SyntheticSessionTTL: 0,
 		MaxBodyBytes:        32 << 20,
 		UpstreamReadTimeout: 600 * time.Second,

internal/domain/access.go

@@ -57,6 +57,8 @@ type UserRecord struct {
 	Prepaid bool
 	Budget  int64
 	Spent   int64
+	// Limits caps the holder's requests and tokens per reset window; none = uncapped.
+	Limits []Limit
 
 	// Set only by SynthUser off a pre-group key, where the control plane had already resolved
 	// access down to one model set. Nothing read from Redis sets it.

internal/domain/denial.go

@@ -1,11 +1,14 @@
 package domain
 
-// Denial carries the status to answer with — 401 credential, 403 grant, 429 budget or full engine,
-// 503 model or store — so a handler translates one type instead of each service's own errors. A
-// store failure is a 503 Denial too: to the caller it is the same answer, and the why goes in the log.
+// Denial carries the status to answer with — 401 credential, 402 credit, 403 grant, 429 rate limit
+// or full engine, 503 model or store — so a handler translates one type instead of each service's
+// own errors. A store failure is a 503 Denial too: to the caller it is the same answer, and the why
+// goes in the log.
 type Denial struct {
 	Status int
 	Reason string
+	// RetryAfter is the seconds until the same request could pass; 0 names none.
+	RetryAfter int
 }
 
 func (d Denial) Error() string { return d.Reason }

internal/domain/endpoint.go

@@ -1,43 +1,104 @@
 package domain
 
-import "strings"
-
-var endpointModalities = map[string]map[string]bool{
-	"/v1/chat/completions":     {"text": true, "multimodal": true},
-	"/v1/messages":             {"text": true, "multimodal": true},
-	"/v1/embeddings":           {"embedding": true},
-	"/v1/audio/transcriptions": {"audio": true},
-	"/v1/audio/translations":   {"audio": true},
-}
+import (
+	"encoding/json"
+	"slices"
+	"strings"
+)
 
-// knownModalities is what this build understands. A value outside it comes from a control plane
-// newer than this binary, and is treated as unrestricted — a fleet mid-upgrade must not start
-// refusing traffic because one side learned a word first.
-var knownModalities = map[string]bool{
-	"text": true, "multimodal": true, "embedding": true, "audio": true,
+// endpointOutputs is the output a model must give to answer on a path. Only the paths one kind of
+// model serves are here; any other is nobody's to refuse.
+var endpointOutputs = map[string]string{
+	"/v1/chat/completions":     "text",
+	"/v1/messages":             "text",
+	"/v1/embeddings":           "embeddings",
+	"/v1/audio/transcriptions": "transcription",
+	"/v1/audio/translations":   "transcription",
 }
 
-// ServesRoute reports whether this route answers on this surface and path — the model's modality
+// ServesRoute reports whether this route answers on this surface and path — what the model gives
 // when it is an engine of ours, and the pushed dialect when it is a vendor. A provider is a closed
 // set where an engine is not: anything but its dialect's own paths is a 404 from it, after a
 // round trip we paid for — so it is our 404 instead.
 func ServesRoute(r Route, dialect, path string) bool {
 	if r.IsProvider() {
 		return PathDialect(path) == dialect && r.SpeaksDialect(dialect)
 	}
-	return Serves(r.Modality, path)
+	return Serves(r.OutputModalities, path)
 }
 
-// Serves reports whether a model of this modality answers on this path. Pure, and deliberately
-// generous: it refuses only when the path is claimed by a modality this model does not have.
-func Serves(modality, path string) bool {
-	modality = strings.TrimSpace(modality)
-	if modality == "" || !knownModalities[modality] {
-		return true
+// Serves reports whether a model giving `outputs` answers on this path. Pure, and deliberately
+// generous: it refuses only a path claimed by an output the model does not give. A model that
+// declares nothing — a row from before the control plane said — is unrestricted.
+func Serves(outputs []string, path string) bool {
+	need, claimed := endpointOutputs[strings.TrimRight(path, "/")]
+	return !claimed || len(outputs) == 0 || slices.Contains(outputs, need)
+}
+
+// partInputs is the input a content part asks of a model, by the part's `type` on either surface.
+// Text is not here: every model takes it. Images and files are the parts looked for yet; any other
+// is the upstream's to refuse.
+// ponytail: every Anthropic document is a file, one of plain text too; read its source.type if
+// that skips fallbacks that would have served.
+var partInputs = map[string]string{
+	"image_url": "image",
+	"image":     "image",
+	"file":      "file",
+	"document":  "file",
+}
+
+// SentInputs is what a chat or messages body carries that partInputs knows of, in the words a
+// model's inputs are declared in. A body it cannot read sends nothing it knows of: that one is the
+// upstream's to refuse.
+// ponytail: copies each message's content once; a token scan if big bodies fall back often.
+func SentInputs(body []byte) []string {
+	var request struct {
+		Messages []struct {
+			Content json.RawMessage `json:"content"`
+		} `json:"messages"`
+	}
+	_ = json.Unmarshal(body, &request)
+	var sent []string
+	for _, message := range request.Messages {
+		sent = partsInputs(message.Content, sent)
 	}
-	allowed, claimed := endpointModalities[strings.TrimRight(path, "/")]
-	if !claimed {
+	return sent
+}
+
+// partsInputs adds what one content list carries, with the lists inside it: a tool result's own,
+// and the parts an Anthropic document is built of. A tool call's input is the caller's JSON, not a
+// part, and is not looked into.
+func partsInputs(content json.RawMessage, sent []string) []string {
+	var parts []struct {
+		Type    string          `json:"type"`
+		Content json.RawMessage `json:"content"`
+		Source  struct {
+			Content json.RawMessage `json:"content"`
+		} `json:"source"`
+	}
+	// The error is not read: a string is text and decodes to no parts, and a part with a field of
+	// the wrong shape must not hide the ones beside it.
+	_ = json.Unmarshal(content, &parts)
+	for _, part := range parts {
+		if input, known := partInputs[part.Type]; known && !slices.Contains(sent, input) {
+			sent = append(sent, input)
+		}
+		sent = partsInputs(part.Content, sent)
+		sent = partsInputs(part.Source.Content, sent)
+	}
+	return sent
+}
+
+// Takes reports whether a model taking `inputs` can be sent a request carrying `sent`. As generous
+// as Serves: a model that declares nothing is unrestricted.
+func Takes(inputs, sent []string) bool {
+	if len(inputs) == 0 {
 		return true
 	}
-	return allowed[modality]
+	for _, input := range sent {
+		if !slices.Contains(inputs, input) {
+			return false
+		}
+	}
+	return true
 }

internal/domain/endpoint_test.go

@@ -1,54 +1,100 @@
 package domain
 
-import "testing"
+import (
+	"slices"
+	"testing"
+)
 
-// A model is just a name in the route table, so nothing but its modality distinguishes an ASR
+// A model is just a name in the route table, so nothing but what it gives distinguishes an ASR
 // container from a chat engine. These are the rules that keep a transcription off a chat model.
 
 func TestServes(t *testing.T) {
+	chat, pooling, asr := []string{"text"}, []string{"embeddings"}, []string{"transcription"}
 	for _, c := range []struct {
-		modality string
-		path     string
-		want     bool
-		why      string
+		outputs []string
+		path    string
+		want    bool
+		why     string
 	}{
-		{"text", "/v1/chat/completions", true, "the ordinary case"},
-		{"text", "/v1/completions", true, "legacy completions is still text"},
-		{"multimodal", "/v1/chat/completions", true, "images ride inside the chat body"},
-		{"embedding", "/v1/embeddings", true, ""},
-		{"audio", "/v1/audio/transcriptions", true, ""},
-		{"audio", "/v1/audio/translations", true, ""},
-		{"multimodal", "/v1/audio/transcriptions", false, "images ride the chat body; audio in is an ASR box"},
-		{"multimodal", "/v1/audio/translations", false, ""},
-		{"multimodal", "/v1/embeddings", false, "nor is it a pooling model"},
-
-		{"text", "/v1/audio/transcriptions", false, "a chat engine would 404 this after a round trip"},
-		{"audio", "/v1/chat/completions", false, "an ASR box cannot hold a conversation"},
-		{"embedding", "/v1/chat/completions", false, ""},
-		{"text", "/v1/embeddings", false, ""},
+		{chat, "/v1/chat/completions", true, "the ordinary case"},
+		{chat, "/v1/completions", true, "legacy completions is unclaimed"},
+		{[]string{"text", "image"}, "/v1/chat/completions", true, "a model that also draws still talks"},
+		{pooling, "/v1/embeddings", true, ""},
+		{asr, "/v1/audio/transcriptions", true, ""},
+		{asr, "/v1/audio/translations", true, ""},
+
+		{chat, "/v1/audio/transcriptions", false, "a chat engine would 404 this after a round trip"},
+		{chat, "/v1/embeddings", false, "nor is it a pooling model"},
+		{asr, "/v1/chat/completions", false, "an ASR box cannot hold a conversation"},
+		{pooling, "/v1/chat/completions", false, ""},
+		{[]string{"image"}, "/v1/chat/completions", false, "a model that only draws gives no text"},
 
 		// /v1/messages is the same request as /v1/chat/completions in Anthropic's spelling, so it
-		// is claimed by the same modalities. Enforcing one spelling and not the other was an
-		// omission, not a policy.
-		{"text", "/v1/messages", true, ""},
-		{"multimodal", "/v1/messages", true, ""},
-		{"embedding", "/v1/messages", false, "a pooling model cannot hold a conversation either way"},
-		{"audio", "/v1/messages", false, ""},
-
-		// Generosity is deliberate: this refuses only what another modality has claimed.
-		{"text", "/v1/rerank", true, "engines serve more than the OpenAI core"},
-		{"audio", "/tokenize", true, "an unclaimed path is nobody's to refuse"},
-		{"embedding", "/v1/score", true, ""},
-
-		// Never take traffic away over a value this build does not recognise.
-		{"", "/v1/audio/transcriptions", true, "blank = a control plane predating the field"},
-		{"   ", "/v1/audio/transcriptions", true, "blank after trimming is still blank"},
-		{"video", "/v1/chat/completions", true, "newer control plane, older binary — must not refuse"},
-
-		{"audio", "/v1/audio/transcriptions/", true, "a trailing slash is the same endpoint"},
+		// asks for the same output.
+		{chat, "/v1/messages", true, ""},
+		{pooling, "/v1/messages", false, "a pooling model cannot hold a conversation either way"},
+		{asr, "/v1/messages", false, ""},
+
+		// Generosity is deliberate: this refuses only a path some output has claimed.
+		{chat, "/v1/rerank", true, "engines serve more than the OpenAI core"},
+		{asr, "/tokenize", true, "an unclaimed path is nobody's to refuse"},
+		{pooling, "/v1/score", true, ""},
+
+		// Never take traffic away from a model that declares nothing.
+		{nil, "/v1/audio/transcriptions", true, "nothing declared = a control plane predating the lists"},
+		{[]string{}, "/v1/chat/completions", true, "an empty list is nothing declared"},
+
+		{asr, "/v1/audio/transcriptions/", true, "a trailing slash is the same endpoint"},
+	} {
+		if got := Serves(c.outputs, c.path); got != c.want {
+			t.Errorf("Serves(%v, %q) = %v, want %v — %s", c.outputs, c.path, got, c.want, c.why)
+		}
+	}
+}
+
+// What a request carries is read off its content parts, in either surface's spelling, and a model
+// is held to it only when it has said what it takes.
+func TestSentInputsAndTakes(t *testing.T) {
+	for _, c := range []struct {
+		body string
+		want []string
+		why  string
+	}{
+		{`{"messages":[{"content":"hi"}]}`, nil, "a string is text"},
+		{`{"messages":[{"content":[{"type":"text","text":"hi"}]}]}`, nil, "every model takes text"},
+		{`{"messages":[{"content":[{"type":"image_url","image_url":{"url":"u"}}]}]}`, []string{"image"}, "OpenAI's image"},
+		{`{"messages":[{"content":[{"type":"image"},{"type":"image"},{"type":"document"},{"type":"input_audio"}]}]}`,
+			[]string{"image", "file"}, "Anthropic's, each word once; a clip is not looked for yet"},
+		{`{"messages":[{"content":[{"type":"file","file":{"file_data":"data:application/pdf;base64,JVBERi0="}}]}]}`, []string{"file"}, "OpenAI's file"},
+		{`{"messages":[{"content":[{"type":"tool_result","content":[{"type":"image"}]}]}]}`,
+			[]string{"image"}, "a tool result's own parts are sent too"},
+		{`{"messages":[{"content":[{"type":"tool_result","content":[{"type":"document",` +
+			`"source":{"type":"content","content":[{"type":"text"},{"type":"image"}]}}]}]}]}`,
+			[]string{"file", "image"}, "so are the parts an Anthropic document is built of"},
+		{`{"messages":[{"content":[{"type":"text","source":"odd"},{"type":"image_url"}]}]}`,
+			[]string{"image"}, "one odd part does not hide the rest"},
+		{`{"messages":[{"content":[{"type":"tool_use","input":{"type":"image"}}]}]}`, nil, "a tool call's input is not a part"},
+		{`{"input":"hi"}`, nil, "no messages"},
+		{`not json`, nil, "unreadable is the upstream's to refuse"},
+	} {
+		if got := SentInputs([]byte(c.body)); !slices.Equal(got, c.want) {
+			t.Errorf("SentInputs(%s) = %v, want %v — %s", c.body, got, c.want, c.why)
+		}
+	}
+
+	for _, c := range []struct {
+		inputs, sent []string
+		want         bool
+		why          string
+	}{
+		{nil, []string{"image"}, true, "nothing declared = unrestricted"},
+		{[]string{"text"}, nil, true, "a text request goes anywhere"},
+		{[]string{"text", "image"}, []string{"image"}, true, ""},
+		{[]string{"text"}, []string{"image"}, false, "a text-only model sent an image"},
+		{[]string{"text", "image"}, []string{"image", "audio"}, false, "every part must be taken"},
 	} {
-		if got := Serves(c.modality, c.path); got != c.want {
-			t.Errorf("Serves(%q, %q) = %v, want %v — %s", c.modality, c.path, got, c.want, c.why)
+		if got := Takes(c.inputs, c.sent); got != c.want {
+			t.Errorf("Takes(%v, %v) = %v, want %v — %s", c.inputs, c.sent, got, c.want, c.why)
 		}
 	}
 }
@@ -57,10 +103,10 @@ func TestServes(t *testing.T) {
 // rather than a round trip that comes back as theirs. The engine table is generous for the
 // opposite reason — an engine serves more than we have written down.
 func TestServesRoute(t *testing.T) {
-	vendor := Route{Kind: "provider", Modality: "text", Dialect: DialectAnthropic}
-	openaiVendor := Route{Kind: "provider", Modality: "text", Dialect: DialectOpenAI}
-	blankVendor := Route{Kind: "provider", Modality: "text"}
-	engine := Route{Kind: "direct", Modality: "text"}
+	vendor := Route{Kind: "provider", OutputModalities: []string{"text"}, Dialect: DialectAnthropic}
+	openaiVendor := Route{Kind: "provider", OutputModalities: []string{"text"}, Dialect: DialectOpenAI}
+	blankVendor := Route{Kind: "provider", OutputModalities: []string{"text"}}
+	engine := Route{Kind: "direct", OutputModalities: []string{"text"}}
 
 	for _, c := range []struct {
 		name    string
@@ -88,7 +134,7 @@ func TestServesRoute(t *testing.T) {
 		{"engine", engine, DialectOpenAI, "/v1/rerank", true, "engines still serve more than the OpenAI core"},
 		{"engine", engine, DialectOpenAI, "/v1/chat/completions", true, ""},
 		{"engine", engine, DialectAnthropic, "/v1/messages", true, "vLLM answers both dialects natively"},
-		{"engine", engine, DialectOpenAI, "/v1/embeddings", false, "a text engine is still held to its modality"},
+		{"engine", engine, DialectOpenAI, "/v1/embeddings", false, "a text engine is still held to what it gives"},
 	} {
 		if got := ServesRoute(c.route, c.dialect, c.path); got != c.want {
 			t.Errorf("ServesRoute(%s, %s, %q) = %v, want %v — %s", c.name, c.dialect, c.path, got, c.want, c.why)
@@ -98,7 +144,7 @@ func TestServesRoute(t *testing.T) {
 
 // A route pushed before the vendor split has no Kind, and must keep being read as one of ours.
 func TestABlankKindIsStillAnEngine(t *testing.T) {
-	if !ServesRoute(Route{Modality: "text"}, DialectOpenAI, "/v1/rerank") {
+	if !ServesRoute(Route{OutputModalities: []string{"text"}}, DialectOpenAI, "/v1/rerank") {
 		t.Error("a route with no kind was treated as a vendor")
 	}
 }