Reusing answers: a response cache per project, exact match, opt-in with a TTL

llmgw keys cache --project P --ttl 1h [--any] | --off, and Cache answers… on
the console's project menu. The same request — project, endpoint, canonical
model and the body without stream, stream_options, user, metadata, store and
service_tier, keys sorted — is answered from the gateway's memory for the
TTL. Only reproducible answers (temperature 0 or a seed; embeddings, rerank,
score) unless --any; only a whole 200 from the model asked for, not a
fallback's, a canary's, or one a guardrail changed.

Looked up after the limits and the input guardrails; the output policy
judges a reused answer; no_store and stored conversations bypass it.
Streamed answers are kept put back together and streamed back as chunks.
x-llmgw-cache: hit|miss|bypass, Age on a hit, Cache-Control: no-cache or
x-llmgw-cache: no to bypass. A hit is recorded with no tokens and no cost,
marked cache_hit (migration 0027) and shown by ctl request. An LRU bounded
by router.cache.max_bytes and max_entry_bytes; llmgw_cache_requests_total,
llmgw_cache_bytes, llmgw_cache_entries.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

deploy/examples/server.yaml

@@ -159,6 +159,12 @@ router:
   # back to a client that did not ask for it.
   inject_usage: true
   max_body_bytes: 33554432
+  # Memory for the answers of projects that reuse them (llmgw keys cache),
+  # all together; the least recently used go first. Nothing is reused until a
+  # project turns it on.
+  # cache:
+  #   max_bytes: 134217728     # 128 MB
+  #   max_entry_bytes: 1048576 # 1 MB
 
   timeouts:
     dial: 5s

docs/console.md

@@ -195,7 +195,9 @@ row: add a team or project inside it, rename, move, limits for the level and
 for each of its members, its capacity quota, and for a project how long
 conversations and agent notes are kept, and whether its requests are
 (**Keep requests…**: for how long, and what share), and whether they stay in
-the cluster (**Requests stay in the cluster…**: never an outside provider). **Longest key lifetime…** caps how
+the cluster (**Requests stay in the cluster…**: never an outside provider),
+and whether the same request is answered again from memory (**Cache
+answers…**: for how long, and whether only reproducible answers or any). **Longest key lifetime…** caps how
 long a key made there may live, and is the expiry of one made without any
 (`llmgw keys ttl`; `auth.max_key_ttl` in server.yaml caps every key).
 **Remove** takes away only something empty.

docs/data.md

@@ -67,10 +67,20 @@ until deleted (`DELETE /v1/files/{id}`) or, for what a batch wrote, until
 them can read them, and a project under a `no_store` policy cannot make
 them.
 
+**Reused answers** — only for a project that asked (`llmgw keys cache`):
+each answer that may be reused, as the client got it, and a hash of the
+request it answers, in the gateway process's memory — never in the store,
+never on disk — for the project's TTL or until pushed out by newer ones
+(`router.cache.max_bytes`), and gone with a restart. Never under a `no_store`
+policy. Forgetting a person does not reach into it; the answers age out by
+the TTL. The store keeps only the setting, and a usage record's mark that it
+was answered from the cache.
+
 ## What is not stored
 
 Prompts. Completions. Anything the model was shown or produced, unless the
 project turned on conversations, notes or kept requests, or sent a batch.
+(Reused answers are held in memory only, as above.)
 
 The chat page in llmgw's own web interface keeps the conversation in a
 JavaScript variable in the browser. Refresh it and the conversation is gone,

docs/install.md

@@ -1208,6 +1208,68 @@ llmgw keys requests --project Acme/ml/train --off
   requests past their hours, or of a project turned off, are removed hourly.
   The console has it as **Keep requests…** on a project.
 
+## Reusing answers
+
+A project that asks the same thing again and again — an evaluation run over
+the same prompts, embeddings of documents that do not change, an agent that
+retries — can have the same request answered from memory rather than by the
+model:
+
+```
+llmgw keys cache --project Acme/ml/train --ttl 1h          # reproducible answers
+llmgw keys cache --project Acme/ml/eval --ttl 24h --any    # every answer
+llmgw keys cache --project Acme/ml/train                   # how it stands
+llmgw keys cache --project Acme/ml/train --off
+```
+
+- **What is the same request:** the same project, the same endpoint, the same
+  model (by its catalog name, whichever alias was used) and the same body,
+  with `stream`, `stream_options`, `user`, `metadata`, `store` and
+  `service_tier` left out and the keys in any order. Keys of another project
+  never see the answer, whatever they send.
+- **Which answers are reused:** a chat or completion asked at `temperature`
+  0 or with a `seed`, and every embedding, rerank and score, which sample
+  nothing. A request sampled above temperature 0 with no seed asks for a new
+  answer, and is asked anew — unless the project says `--any`. Only a whole
+  200 from the model that was asked for is kept: not an error, not a
+  fallback model's or a canary's answer, not one a guardrail changed or
+  stopped, not one longer than `router.cache.max_entry_bytes` (1 MB).
+- **Streams:** an answer given whole is streamed to a client that asks for a
+  stream, as one chunk per choice with the usage chunk if it asked for that;
+  a streamed answer is kept put back together, and served whole to a client
+  that does not stream. Through `/v1/messages` and `/v1/responses` too.
+- **The client decides too:** `Cache-Control: no-cache` (or `no-store`), or
+  `x-llmgw-cache: no`, is always asked anew. Every answer of a project that
+  reuses answers says `x-llmgw-cache: hit`, `miss` or `bypass` (not one that
+  can be reused), and a hit says its `Age` in seconds.
+- **Policies still apply:** the limits are checked and the guardrails judge
+  the request before the cache is looked in, and the output policy judges a
+  reused answer as it would a new one. A request continuing a stored
+  conversation, or under a `no_store` policy, is never answered from the
+  cache nor kept in it.
+- **What a hit costs:** nothing. It is recorded as a request — status 200,
+  the model, the key, the rate limit's count — with no tokens and no cost,
+  marked as a cache hit (`llmgw ctl request <id>` says so), so it spends no
+  budget and no token limit. The answer itself carries the usage the model
+  reported when it was first given.
+- **Where it lives:** in each gateway process's memory, at most
+  `router.cache.max_bytes` (128 MB) of answers for all projects together,
+  those used longest ago dropped first; nothing is written to the store, and
+  a restart begins it again. With two control planes behind a balancer each
+  keeps its own. A change of setting is seen within 30 seconds; a shorter TTL
+  applies to what is already kept.
+
+```yaml
+router:
+  cache:
+    max_bytes: 268435456     # 256 MB
+    max_entry_bytes: 2097152 # 2 MB
+```
+
+`llmgw_cache_requests_total{model,result}` counts hits, misses and
+bypasses, `llmgw_cache_bytes` and `llmgw_cache_entries` what is held. The
+console has it as **Cache answers…** on a project.
+
 ## Batches
 
 Work that can wait — a night's classification, embeddings for a document

docs/reference.md

@@ -47,6 +47,8 @@ rather than repeated.
 | `max_lines` | `int` | [Batch](#batch) |
 | `concurrency` | `int` | [Batch](#batch) |
 | `retain_days` | `int` | [Batch](#batch) |
+| `max_bytes` | `int64` | [Cache](#cache) |
+| `max_entry_bytes` | `int64` | [Cache](#cache) |
 | `model` | `string` | [Canary](#canary) |
 | `percent` | `int` | [Canary](#canary) |
 | `min_requests` | `int` | [Canary](#canary) |
@@ -204,6 +206,7 @@ rather than repeated.
 | `tag_labels` | `[]string` | [Router](#router) |
 | `max_body_bytes` | `int64` | [Router](#router) |
 | `batch` | `Batch` | [Router](#router) |
+| `cache` | `Cache` | [Router](#router) |
 | `trusted_proxies` | `[]string` | [Router](#router) |
 | `timeouts` | `Timeouts` | [Router](#router) |
 | `scoring` | `Scoring` | [Router](#router) |
@@ -570,6 +573,22 @@ every interactive request.
 RetainDays is how long a batch's answer and error files are kept;
 30 unless set. Uploaded files are kept until deleted.
 
+## Cache
+
+### `max_bytes`
+
+`int64`
+
+MaxBytes bounds every project's answers together; 128 MB unless set.
+Past it, the answers used longest ago go first.
+
+### `max_entry_bytes`
+
+`int64`
+
+MaxEntryBytes bounds one answer; a longer one is not kept. 1 MB
+unless set.
+
 ## Canary
 
 ### `model`
@@ -1692,6 +1711,16 @@ another. Needs the store.
 
 Takes a [Batch](#batch) block.
 
+### `cache`
+
+`Cache`
+
+Cache bounds the answers kept for the projects that reuse them
+(llmgw keys cache). Which projects do, and for how long, is theirs
+to set; this is only how much memory it may take.
+
+Takes a [Cache](#cache) block.
+
 ### `trusted_proxies`
 
 `[]string`

internal/cli/answercache.go

@@ -0,0 +1,85 @@
+package cli
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/spf13/cobra"
+)
+
+func newAnswerCacheCmd(open opener) *cobra.Command {
+	var (
+		project   string
+		ttl       time.Duration
+		anyAnswer bool
+		off       bool
+	)
+	cmd := &cobra.Command{
+		Use:   "cache",
+		Short: "Reuse a project's answers: the same request is answered again without the model",
+		Long: "With this on, a request of the project that is the same as one answered within\n" +
+			"--ttl, to the same model, gets that answer again, from the gateway's memory,\n" +
+			"rather than asking the model. It is answered with x-llmgw-cache: hit, recorded\n" +
+			"as a request with no tokens, and costs nothing against a budget.\n\n" +
+			"Only answers that would come out the same are reused: a chat or completion\n" +
+			"asked at temperature 0 or with a seed, and embeddings, rerank and score.\n" +
+			"--any reuses every answer, sampled or not. A client sending\n" +
+			"Cache-Control: no-cache or x-llmgw-cache: no is always asked afresh, and so is\n" +
+			"a request continuing a stored conversation or under a guardrails no_store\n" +
+			"policy. The guardrails judge a reused answer as they judge any other.\n\n" +
+			"Off for every project until this is run. Without --ttl or --off, says how\n" +
+			"it stands. The answers are held in memory, router.cache bounds how much.",
+		Example: "  llmgw keys cache --project Acme/ml/train --ttl 1h\n" +
+			"  llmgw keys cache --project Acme/ml/eval --ttl 24h --any\n" +
+			"  llmgw keys cache --project Acme/ml/train --off",
+		Args: cobra.NoArgs,
+		RunE: func(cmd *cobra.Command, _ []string) error {
+			switch {
+			case off && (ttl != 0 || anyAnswer):
+				return fmt.Errorf("--off, or --ttl with --any: not both")
+			case ttl < 0:
+				return fmt.Errorf("--ttl is how long an answer is reused, like 1h")
+			case ttl > 0 && ttl < time.Second:
+				return fmt.Errorf("--ttl is at least a second")
+			case anyAnswer && ttl == 0:
+				return fmt.Errorf("--any goes with --ttl: say how long answers are reused")
+			}
+			s, err := open(cmd.Context())
+			if err != nil {
+				return err
+			}
+			defer s.Close()
+			p, err := resolveProject(cmd.Context(), s, project)
+			if err != nil {
+				return err
+			}
+			if off || ttl > 0 {
+				if err := s.SetAnswerCache(cmd.Context(), p.ID, ttl, anyAnswer); err != nil {
+					return err
+				}
+			}
+			c, err := s.AnswerCacheOf(cmd.Context(), p.ID)
+			if err != nil {
+				return err
+			}
+			switch {
+			case c.TTL == 0:
+				fmt.Fprintf(cmd.OutOrStdout(), "%s reuses no answers.\n", p.Path)
+			case c.Any:
+				fmt.Fprintf(cmd.OutOrStdout(), "%s reuses every answer for %s, sampled or not.\n", p.Path, c.TTL)
+			default:
+				fmt.Fprintf(cmd.OutOrStdout(), "%s reuses answers for %s: those asked at temperature 0 or with a seed, "+
+					"and embeddings, rerank and score.\n", p.Path, c.TTL)
+			}
+			if off || ttl > 0 {
+				fmt.Fprintln(cmd.OutOrStdout(), "Gateways see the change within 30 seconds.")
+			}
+			return nil
+		},
+	}
+	cmd.Flags().StringVar(&project, "project", "", "project: id, path or unique name")
+	cmd.Flags().DurationVar(&ttl, "ttl", 0, "how long an answer is reused, like 1h")
+	cmd.Flags().BoolVar(&anyAnswer, "any", false, "reuse sampled answers too, not only reproducible ones")
+	cmd.Flags().BoolVar(&off, "off", false, "stop reusing answers for this project")
+	return cmd
+}

internal/cli/answercache_test.go

@@ -0,0 +1,32 @@
+package cli
+
+import (
+	"path/filepath"
+	"strings"
+	"testing"
+)
+
+func TestAProjectIsToldToReuseItsAnswersAndToStop(t *testing.T) {
+	db := filepath.Join(t.TempDir(), "s.db")
+	mustRun(t, db, "org", "add", "org", "Acme")
+	mustRun(t, db, "org", "add", "team", "ml", "--in", "Acme")
+	mustRun(t, db, "org", "add", "project", "train", "--in", "Acme/ml")
+
+	if out := mustRun(t, db, "keys", "cache", "--project", "Acme/ml/train"); !strings.Contains(out, "reuses no answers") {
+		t.Errorf("before: %s", out)
+	}
+	if out := mustRun(t, db, "keys", "cache", "--project", "Acme/ml/train", "--ttl", "1h"); !strings.Contains(out, "for 1h0m0s: those asked at temperature 0") {
+		t.Errorf("on: %s", out)
+	}
+	if out := mustRun(t, db, "keys", "cache", "--project", "train", "--ttl", "24h", "--any"); !strings.Contains(out, "every answer for 24h0m0s") {
+		t.Errorf("any: %s", out)
+	}
+	for _, bad := range [][]string{{"--off", "--ttl", "1h"}, {"--any"}, {"--ttl", "10ms"}} {
+		if _, err := run(t, db, append([]string{"keys", "cache", "--project", "train"}, bad...)...); err == nil {
+			t.Errorf("%v was accepted", bad)
+		}
+	}
+	if out := mustRun(t, db, "keys", "cache", "--project", "train", "--off"); !strings.Contains(out, "reuses no answers") {
+		t.Errorf("off: %s", out)
+	}
+}

internal/cli/ctl_request.go

@@ -56,6 +56,9 @@ func newRequestCmd(client func() *ctlClient, asJSON *bool) *cobra.Command {
 			fmt.Fprintf(out, "status    %d\n", v.Status)
 			fmt.Fprintf(out, "time      %d ms total: %d queued, %d to the first token\n", v.LatencyMs, v.QueueMs, v.TTFTMs)
 			fmt.Fprintf(out, "tokens    %d in, %d out\n", v.PromptTokens, v.CompletionTokens)
+			if v.CacheHit {
+				fmt.Fprintf(out, "cache     answered from the project's answer cache: nothing generated, nothing charged\n")
+			}
 			fmt.Fprintf(out, "who       key %s, project %s%s\n", v.KeyID, v.ProjectID, prefixed(", user ", v.UserID))
 			if v.EndUser != "" {
 				fmt.Fprintf(out, "end user  %s (the client's own name for them)\n", v.EndUser)

internal/cli/keys.go

@@ -44,6 +44,7 @@ func newTenancyCmd() *cobra.Command {
 		newThreadsCmd(open),
 		newRequestLogCmd(open),
 		newResidencyCmd(open),
+		newAnswerCacheCmd(open),
 		newMemoryCmd(open),
 		newMoveStoreCmd(),
 	)

internal/cli/tui_people.go

@@ -72,6 +72,8 @@ type peopleData struct {
 	notes   map[string]int // project id → hours kept, agent notes
 	// requests is how each project keeps its requests and answers.
 	requests map[string]store.RequestLog
+	// cache is how each project reuses answers.
+	cache map[string]store.AnswerCache
 	// guardPolicies are server.yaml's guardrail policies, guardDefault those
 	// applied everywhere, and guardAt which are attached where.
 	guardPolicies []string
@@ -125,6 +127,7 @@ func (m consoleModel) fetchPeople() tea.Cmd {
 			d.guardAt, _ = s.GuardAttachments(ctx)
 		}
 		d.threads, d.notes, d.requests = map[string]int{}, map[string]int{}, map[string]store.RequestLog{}
+		d.cache = map[string]store.AnswerCache{}
 		for _, r := range d.orgs {
 			if r.level != "project" {
 				continue
@@ -138,6 +141,9 @@ func (m consoleModel) fetchPeople() tea.Cmd {
 			if l, err := s.RequestLogOf(ctx, r.id); err == nil {
 				d.requests[r.id] = l
 			}
+			if c, err := s.AnswerCacheOf(ctx, r.id); err == nil {
+				d.cache[r.id] = c
+			}
 		}
 		return consolePeopleMsg(d)
 	}
@@ -552,6 +558,39 @@ func (m consoleModel) orgMenu(r orgRow) *menuDialog {
 						return m.runCLI("Keep requests", append(args, "--retain-hours="+v[0], "--sample="+v[1])...), nil
 					}})
 				return nil
+			}},
+			menuItem{"Cache answers…", func(m *consoleModel) tea.Cmd {
+				c := m.people.cache[r.id]
+				ttl, which := "off", "reproducible"
+				if c.TTL > 0 {
+					ttl = c.TTL.String()
+				}
+				if c.Any {
+					which = "any"
+				}
+				m.openForm(&formDialog{title: "Cache answers · " + r.path,
+					note: "Off by default. On, the same request to the same model within the time below is\n" +
+						"answered again from the gateway's memory, without the model, and costs nothing.\n" +
+						"Reproducible: only temperature 0 or a seed, and embeddings. Any: sampled answers too.",
+					fields: []formField{
+						textF("reuse for", ttl, "like 1h; off reuses none"),
+						choiceF("answers", []string{"reproducible", "any"}, which),
+					},
+					submit: func(m *consoleModel, v []string) (tea.Cmd, error) {
+						args := []string{"keys", "cache", "--project=" + r.id}
+						if v[0] == "off" || v[0] == "0" {
+							return m.runCLI("Cache answers", append(args, "--off")...), nil
+						}
+						if d, err := time.ParseDuration(v[0]); err != nil || d < time.Second {
+							return nil, fmt.Errorf("a duration like 1h, or off")
+						}
+						args = append(args, "--ttl="+v[0])
+						if v[1] == "any" {
+							args = append(args, "--any")
+						}
+						return m.runCLI("Cache answers", args...), nil
+					}})
+				return nil
 			}})
 	}
 	items = append(items, menuItem{"Remove…", func(m *consoleModel) tea.Cmd {

internal/config/config.go

@@ -476,6 +476,11 @@ type Router struct {
 	// another. Needs the store.
 	Batch Batch `koanf:"batch"`
 
+	// Cache bounds the answers kept for the projects that reuse them
+	// (llmgw keys cache). Which projects do, and for how long, is theirs
+	// to set; this is only how much memory it may take.
+	Cache Cache `koanf:"cache"`
+
 	// TrustedProxies are the reverse proxies in front of the client port,
 	// as addresses or CIDR prefixes. Their X-Forwarded-For, -Proto and -Host
 	// are believed; anyone else's are not, since they decide the address
@@ -609,6 +614,18 @@ type Batch struct {
 	RetainDays int `koanf:"retain_days"`
 }
 
+// Cache is how much of this process's memory reused answers may take. The
+// answers are held here and nowhere else: each gateway process keeps its
+// own, and a restart begins them again.
+type Cache struct {
+	// MaxBytes bounds every project's answers together; 128 MB unless set.
+	// Past it, the answers used longest ago go first.
+	MaxBytes int64 `koanf:"max_bytes"`
+	// MaxEntryBytes bounds one answer; a longer one is not kept. 1 MB
+	// unless set.
+	MaxEntryBytes int64 `koanf:"max_entry_bytes"`
+}
+
 // Default returns the configuration used when nothing is specified.
 func Default() Server {
 	return Server{

internal/control/adminapi/api.go

@@ -1419,6 +1419,8 @@ type RequestView struct {
 	// What the client said about it: OpenAI's user field, and its tags.
 	EndUser string            `json:"end_user,omitempty"`
 	Tags    map[string]string `json:"tags,omitempty"`
+	// CacheHit is an answer reused from the project's answer cache.
+	CacheHit bool `json:"cache_hit,omitempty"`
 }
 
 func (a *API) request(w http.ResponseWriter, r *http.Request) {
@@ -1442,7 +1444,7 @@ func (a *API) request(w http.ResponseWriter, r *http.Request) {
 		LatencyMs: u.LatencyMs, QueueMs: u.QueueMs, TTFTMs: u.TTFTMs,
 		PromptTokens: u.PromptTokens, CompletionTokens: u.CompletionTokens, CostMicros: u.CostMicros,
 		GuardAction: u.GuardAction, GuardPolicy: u.GuardPolicy, GuardCategory: u.GuardCategory, GuardMs: u.GuardMs,
-		EndUser: u.EndUser, Tags: u.Tags,
+		EndUser: u.EndUser, Tags: u.Tags, CacheHit: u.CacheHit,
 	})
 }
 

internal/docgen/reference_gen.go

@@ -44,6 +44,10 @@ var Reference = []Block{
 		{Key: "concurrency", Type: "int", Doc: "Concurrency is how many of one batch's requests are in flight at\nonce; 4 unless set. They queue in the lowest admission band, behind\nevery interactive request.", Nested: ""},
 		{Key: "retain_days", Type: "int", Doc: "RetainDays is how long a batch's answer and error files are kept;\n30 unless set. Uploaded files are kept until deleted.", Nested: ""},
 	}},
+	{Package: "config", Name: "Cache", Title: "Cache", Doc: "", Fields: []Field{
+		{Key: "max_bytes", Type: "int64", Doc: "MaxBytes bounds every project's answers together; 128 MB unless set.\nPast it, the answers used longest ago go first.", Nested: ""},
+		{Key: "max_entry_bytes", Type: "int64", Doc: "MaxEntryBytes bounds one answer; a longer one is not kept. 1 MB\nunless set.", Nested: ""},
+	}},
 	{Package: "catalog", Name: "Canary", Title: "Canary", Doc: "", Fields: []Field{
 		{Key: "model", Type: "string", Doc: "Model is the candidate: a catalog model, usually this one at newer\nweights, which clients cannot name themselves.", Nested: ""},
 		{Key: "percent", Type: "int", Doc: "Percent of requests go to it, 0 to 100.", Nested: ""},
@@ -248,6 +252,7 @@ var Reference = []Block{
 		{Key: "tag_labels", Type: "[]string", Doc: "TagLabels are the request tags (metadata, x-llmgw-tags) counted in\nllmgw_tagged_requests_total and llmgw_tagged_tokens_total, by value.\nOnly these: a tag is the client's to set, and every value of one is\na series. The first 100 values of each are counted apart, the rest as\n\"other\". Empty counts none; the usage records keep every tag anyway.", Nested: ""},
 		{Key: "max_body_bytes", Type: "int64", Doc: "MaxBodyBytes bounds the request body the router will read.", Nested: ""},
 		{Key: "batch", Type: "Batch", Doc: "Batch is the Files and Batch APIs (/v1/files, /v1/batches): a file of\nrequests served at the lowest priority, its answers written to\nanother. Needs the store.", Nested: "config.Batch"},
+		{Key: "cache", Type: "Cache", Doc: "Cache bounds the answers kept for the projects that reuse them\n(llmgw keys cache). Which projects do, and for how long, is theirs\nto set; this is only how much memory it may take.", Nested: "config.Cache"},
 		{Key: "trusted_proxies", Type: "[]string", Doc: "TrustedProxies are the reverse proxies in front of the client port,\nas addresses or CIDR prefixes. Their X-Forwarded-For, -Proto and -Host\nare believed; anyone else's are not, since they decide the address\nwritten into the setup scripts and whose attempts are counted where.", Nested: ""},
 		{Key: "timeouts", Type: "Timeouts", Doc: "", Nested: "config.Timeouts"},
 		{Key: "scoring", Type: "Scoring", Doc: "", Nested: "config.Scoring"},

internal/router/cache.go

@@ -0,0 +1,525 @@
+package router
+
+import (
+	"bytes"
+	"container/list"
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"io"
+	"net/http"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+
+	"gitea.mynodes.xyz/ilker/llmgw/internal/guard"
+	"gitea.mynodes.xyz/ilker/llmgw/internal/router/proxy"
+	storepkg "gitea.mynodes.xyz/ilker/llmgw/internal/store"
+)
+
+// Reusing answers, for a project that asks: the same request to the same
+// model, from the same project and within its TTL, is answered with what
+// the model said the first time rather than asking it again.
+//
+// Only an answer that would come out the same is reused, unless the project
+// says otherwise: a chat or a completion asked at temperature 0 or with a
+// seed, and an embedding, a rerank or a score, which have no sampling at
+// all. A request sampled above temperature 0 is asked for a new answer, and
+// handing it an old one would quietly turn variety into repetition.
+//
+// The answers are held in this process's memory and nowhere else. A shared
+// cache in the store would write every answer of every project that turned
+// this on to the database — the very thing the request log is opt-in for —
+// and add a round trip to each request to save one; with two control planes
+// behind a balancer each fills its own, which costs some hits and no
+// correctness.
+
+// cacheHeader says what the cache did for a request, hit, miss or bypass,
+// and is how a client asks it to stay out of the way: "no".
+const cacheHeader = "x-llmgw-cache"
+
+// Defaults for router.cache.
+const (
+	defaultCacheBytes      = 128 << 20
+	defaultCacheEntryBytes = 1 << 20
+)
+
+// cacheEndpoints are those whose answers can be reused. The chat and
+// completion endpoints only when the answer is reproducible; the others
+// always, as they sample nothing. Audio is a form, not JSON, and is not
+// among them.
+var cacheEndpoints = map[string]bool{
+	"/v1/chat/completions": true,
+	"/v1/completions":      true,
+	"/v1/embeddings":       true,
+	"/v1/rerank":           true,
+	"/v1/score":            true,
+}
+
+// sampledEndpoints are those whose answer depends on sampling.
+var sampledEndpoints = map[string]bool{
+	"/v1/chat/completions": true,
+	"/v1/completions":      true,
+}
+
+// cacheState is where a request's answer is kept, for one whose answer
+// may be reused and was not found.
+type cacheState struct {
+	// key is the answer's place in the cache; empty when it is not kept.
+	key string
+	// project is whose answer it is, and ttl how long it is reused.
+	project string
+	ttl     time.Duration
+}
+
+// cachedAnswer is one reusable answer.
+type cachedAnswer struct {
+	key, project string
+	// model is the model that answered, by its catalog name.
+	model string
+	path  string
+	// body is the answer as one JSON object — a chat.completion, a
+	// text_completion, a list of embeddings — even when it was streamed.
+	body []byte
+	at   time.Time
+	// expires is when the TTL in force when it was kept runs out. The TTL
+	// in force when it is asked for is checked too, so shortening a
+	// project's TTL shortens the life of what it already has.
+	expires time.Time
+	size    int64
+}
+
+// answerCache is a least-recently-used cache of answers, bounded in bytes.
+type answerCache struct {
+	maxBytes, maxEntry int64
+
+	mu    sync.Mutex
+	lru   *list.List // of *cachedAnswer, most recently used at the front
+	by    map[string]*list.Element
+	bytes int64
+	// swept is when expired answers were last cleared out.
+	swept time.Time
+	// size reports bytes and answers held, after each change.
+	size func(bytes int64, n int)
+}
+
+func newAnswerCache(maxBytes, maxEntry int64, size func(int64, int)) *answerCache {
+	if maxBytes <= 0 {
+		maxBytes = defaultCacheBytes
+	}
+	if maxEntry <= 0 {
+		maxEntry = defaultCacheEntryBytes
+	}
+	maxEntry = min(maxEntry, maxBytes)
+	if size == nil {
+		size = func(int64, int) {}
+	}
+	return &answerCache{maxBytes: maxBytes, maxEntry: maxEntry, lru: list.New(),
+		by: map[string]*list.Element{}, size: size}
+}
+
+// get is the answer kept under key, when it is younger than ttl and has
+// not expired.
+func (c *answerCache) get(key string, ttl time.Duration, now time.Time) *cachedAnswer {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	el, ok := c.by[key]
+	if !ok {
+		return nil
+	}
+	a := el.Value.(*cachedAnswer)
+	if now.After(a.expires) || now.Sub(a.at) >= ttl {
+		c.remove(el)
+		c.size(c.bytes, len(c.by))
+		return nil
+	}
+	c.lru.MoveToFront(el)
+	return a
+}
+
+// put keeps an answer, making room for it by dropping those used longest
+// ago. One larger than an entry may be is not kept.
+func (c *answerCache) put(a *cachedAnswer, now time.Time) bool {
+	// What the entry costs besides its body: the key twice, the names, and
+	// the bookkeeping, roughly.
+	a.size = int64(len(a.body) + 2*len(a.key) + len(a.project) + len(a.model) + len(a.path) + 160)
+	if a.size > c.maxEntry {
+		return false
+	}
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if el, ok := c.by[a.key]; ok {
+		c.remove(el)
+	}
+	// Now and then, those past their time, so a quiet project's answers
+	// do not sit in memory until pushed out by others.
+	if now.Sub(c.swept) > time.Minute {
+		c.swept = now
+		for el := c.lru.Back(); el != nil; {
+			prev := el.Prev()
+			if now.After(el.Value.(*cachedAnswer).expires) {
+				c.remove(el)
+			}
+			el = prev
+		}
+	}
+	for c.bytes+a.size > c.maxBytes && c.lru.Len() > 0 {
+		c.remove(c.lru.Back())
+	}
+	c.by[a.key] = c.lru.PushFront(a)
+	c.bytes += a.size
+	c.size(c.bytes, len(c.by))
+	return true
+}
+
+func (c *answerCache) remove(el *list.Element) {
+	a := el.Value.(*cachedAnswer)
+	c.lru.Remove(el)
+	delete(c.by, a.key)
+	c.bytes -= a.size
+}
+
+// answerCacheOf is a project's setting, read at most every requestLogTTL
+// with its other settings. Unreadable, it reuses nothing: asking the model
+// again is the safe way to be wrong.
+func (r *Router) answerCacheOf(ctx context.Context, project string) storepkg.AnswerCache {
+	if r.store == nil || project == "" {
+		return storepkg.AnswerCache{}
+	}
+	r.reqLogs.mu.Lock()
+	c, ok := r.reqLogs.by[project]
+	r.reqLogs.mu.Unlock()
+	if ok && time.Since(c.at) < requestLogTTL {
+		return c.cache
+	}
+	c, err := r.readProjectPolicy(ctx, project)
+	if err != nil {
+		r.log.Warn("could not read whether a project reuses answers", "project", project, "err", err)
+	}
+	return c.cache
+}
+
+// cacheLookup decides whether a request's answer may come from its
+// project's cache, says what it decided in x-llmgw-cache, and returns the
+// answer when there is one. Run after the guardrails have judged the
+// request, so a cached answer reaches only a request that would have been
+// sent to the model, and the request it is looked up by is the one the
+// model would have been sent.
+func (r *Router) cacheLookup(ctx context.Context, w http.ResponseWriter, req *parsedRequest) *cachedAnswer {
+	if r.answers == nil || req.internal || req.ProjectID == "" {
+		return nil
+	}
+	set := r.answerCacheOf(ctx, req.ProjectID)
+	if set.TTL <= 0 {
+		// Says nothing: there is no cache here to have bypassed.
+		return nil
+	}
+	result := "bypass"
+	defer func() {
+		w.Header().Set(cacheHeader, result)
+		r.m.cacheRequests.WithLabelValues(r.metricModel(req.Model), result).Inc()
+	}()
+	if !reusable(req, set.Any) {
+		return nil
+	}
+	model, _ := r.resolve(req.Model)
+	key, ok := answerKey(req.ProjectID, req.Path, model, req.UpstreamBody)
+	if !ok {
+		return nil
+	}
+	now := time.Now()
+	if a := r.answers.get(key, set.TTL, now); a != nil && a.project == req.ProjectID {
+		result = "hit"
+		// Age is HTTP's own way of saying how old a reused answer is.
+		w.Header().Set("Age", strconv.Itoa(int(now.Sub(a.at).Seconds())))
+		return a
+	}
+	result = "miss"
+	req.cache = cacheState{key: key, project: req.ProjectID, ttl: set.TTL}
+	return nil
+}
+
+// reusable says whether a request's answer may be reused at all: not one
+// the client asked to be answered afresh, not one continuing a stored
+// conversation (each turn is new, and is saved), not one under a policy
+// that keeps nothing — a cached answer is a kept one — and, for chat and
+// completions, only one asked at temperature 0 or with a seed unless the
+// project reuses any answer.
+func reusable(req *parsedRequest, anyAnswer bool) bool {
+	if req.noStore || req.ThreadID != "" || req.ContentType != "" || !cacheEndpoints[req.Path] {
+		return false
+	}
+	if v := strings.ToLower(strings.TrimSpace(req.Header.Get(cacheHeader))); v == "no" || v == "off" || v == "bypass" {
+		return false
+	}
+	for _, v := range req.Header.Values("Cache-Control") {
+		v = strings.ToLower(v)
+		if strings.Contains(v, "no-cache") || strings.Contains(v, "no-store") {
+			return false
+		}
+	}
+	if !sampledEndpoints[req.Path] || anyAnswer {
+		return true
+	}
+	var b struct {
+		Temperature *float64        `json:"temperature"`
+		Seed        json.RawMessage `json:"seed"`
+	}
+	if json.Unmarshal(req.UpstreamBody, &b) != nil {
+		return false
+	}
+	seeded := len(b.Seed) > 0 && string(b.Seed) != "null"
+	return seeded || b.Temperature != nil && *b.Temperature == 0
+}
+
+// answerFields are taken out of a request before it is looked up: they
+// change how an answer is delivered or recorded, never what it says.
+var answerFields = []string{"model", "stream", "stream_options", "user", "metadata", "store", "service_tier"}
+
+// answerKey is where a request's answer is kept: the project, the endpoint,
+// the model by its catalog name, and the request with what does not change
+// the answer taken out and its keys sorted. The project is spelled out
+// rather than only hashed, so no answer is ever looked up across projects.
+func answerKey(project, path, model string, body []byte) (string, bool) {
+	d := json.NewDecoder(bytes.NewReader(body))
+	d.UseNumber()
+	var m map[string]any
+	if err := d.Decode(&m); err != nil || m == nil {
+		return "", false
+	}
+	for _, f := range answerFields {
+		delete(m, f)
+	}
+	// encoding/json writes map keys sorted, at every depth.
+	norm, err := json.Marshal(canonicalNumbers(m))
+	if err != nil {
+		return "", false
+	}
+	h := sha256.New()
+	for _, part := range [][]byte{[]byte(path), []byte(model), norm} {
+		h.Write(part)
+		h.Write([]byte{0})
+	}
+	return project + "\x00" + hex.EncodeToString(h.Sum(nil)), true
+}
+
+// canonicalNumbers writes each number one way, so 0, 0.0 and 0e0 are the
+// same temperature. Whole numbers are left as written: a seed above 2^53
+// would lose its last digits as a float, and two seeds become one.
+func canonicalNumbers(v any) any {
+	switch x := v.(type) {
+	case map[string]any:
+		for k, e := range x {
+			x[k] = canonicalNumbers(e)
+		}
+	case []any:
+		for i, e := range x {
+			x[i] = canonicalNumbers(e)
+		}
+	case json.Number:
+		s := string(x)
+		if _, err := strconv.ParseInt(s, 10, 64); err == nil {
+			return x
+		}
+		if f, err := strconv.ParseFloat(s, 64); err == nil {
+			if f == float64(int64(f)) && f > -1<<53 && f < 1<<53 {
+				return json.Number(strconv.FormatInt(int64(f), 10))
+			}
+			return json.Number(strconv.FormatFloat(f, 'g', -1, 64))
+		}
+	}
+	return v
+}
+
+// cacheWriter keeps a copy of what is written to the client, up to a
+// bound, so an answer can be kept once it is known to be a good one.
+type cacheWriter struct {
+	http.ResponseWriter
+	status int
+	buf    bytes.Buffer
+	limit  int
+	over   bool
+}
+
+func (c *cacheWriter) WriteHeader(code int) {
+	if c.status == 0 {
+		c.status = code
+	}
+	c.ResponseWriter.WriteHeader(code)
+}
+
+func (c *cacheWriter) Write(p []byte) (int, error) {
+	if c.status == 0 {
+		c.status = http.StatusOK
+	}
+	if !c.over {
+		if c.buf.Len()+len(p) > c.limit {
+			c.over = true
+			c.buf = bytes.Buffer{}
+		} else {
+			c.buf.Write(p)
+		}
+	}
+	return c.ResponseWriter.Write(p)
+}
+
+// Flush passes through, which the proxy needs to stream.
+func (c *cacheWriter) Flush() {
+	if f, ok := c.ResponseWriter.(http.Flusher); ok {
+		f.Flush()
+	}
+}
+
+func (c *cacheWriter) Unwrap() http.ResponseWriter { return c.ResponseWriter }
+
+// cacheWriterFor wraps w when a request's answer may be kept.
+func (r *Router) cacheWriterFor(w http.ResponseWriter, req *parsedRequest) (http.ResponseWriter, *cacheWriter) {
+	if req.cache.key == "" {
+		return w, nil
+	}
+	cw := &cacheWriter{ResponseWriter: w, limit: int(r.answers.maxEntry)}
+	return cw, cw
+}
+
+// cacheKeep keeps an answer that may be reused: a whole 200 from the model
+// asked for — not a fallback, which is another model, nor a canary, which
+// is weights on trial — that no guard changed or stopped.
+func (r *Router) cacheKeep(req *parsedRequest, res dispatchResult, status int, cw *cacheWriter) {
+	if cw == nil || cw.over || status != http.StatusOK || cw.status != http.StatusOK || res.err != nil ||
+		res.outcome.Status != http.StatusOK || res.outcome.Stopped || res.fallback || res.onCanary {
+		return
+	}
+	if res.judged && res.guard.Action != guard.Allow && !res.guard.Shadow {
+		return
+	}
+	var body []byte
+	captured := bytes.TrimSpace(cw.buf.Bytes())
+	if strings.Contains(cw.Header().Get("Content-Type"), "text/event-stream") ||
+		bytes.HasPrefix(captured, []byte("data:")) {
+		b, ok := assembleStream(req.Path, captured, res.outcome.Usage)
+		if !ok {
+			return
+		}
+		body = b
+	} else {
+		if len(captured) == 0 || captured[0] != '{' || !json.Valid(captured) {
+			return
+		}
+		body = bytes.Clone(captured)
+	}
+	model := res.model
+	if model == "" {
+		model, _ = r.resolve(req.Model)
+	}
+	now := time.Now()
+	r.answers.put(&cachedAnswer{key: req.cache.key, project: req.cache.project, model: model,
+		path: req.Path, body: body, at: now, expires: now.Add(req.cache.ttl)}, now)
+}
+
+// serveCached answers a request from the cache, as the model would have:
+// as one JSON object, or as a stream of chunks to a client that asked for
+// one. The guardrails' output policy judges it as it judges any answer —
+// the policy in force now, not the one when it was first given — and the
+// usage is recorded as a request that generated nothing: no tokens, no
+// cost, marked as a cache hit.
+func (r *Router) serveCached(ctx context.Context, w http.ResponseWriter, req *parsedRequest, a *cachedAnswer, start time.Time) {
+	body := a.body
+	res := dispatchResult{servedModel: servedModel{model: a.model}, cached: true}
+	if r.guard != nil {
+		if f := r.guard.Output(ctx, r.subject(req, a.model)); f != nil {
+			// A copy: the kept answer is shared by every request reading it,
+			// and a filter is free to rewrite what it is given.
+			body = f.Buffered(bytes.Clone(body))
+			res.guard, res.judged = f.Verdict(), true
+			if res.guard.Stricter(req.guardIn) {
+				setGuardHeaders(w.Header(), res.guard)
+			}
+		}
+	}
+	var werr error
+	if req.Stream && sampledEndpoints[req.Path] {
+		werr = writeCachedStream(w, req, body)
+	} else {
+		w.Header().Set("Content-Type", "application/json")
+		w.WriteHeader(http.StatusOK)
+		_, werr = w.Write(body)
+	}
+	status := http.StatusOK
+	if werr != nil {
+		status = 499
+	}
+	res.outcome = proxy.Outcome{Status: http.StatusOK, Wrote: true, Usage: &proxy.Usage{},
+		Reply: replyText(body)}
+	took := time.Since(start)
+	r.m.observeRequest(r.metricModel(req.Model), "cache", status, took, nil, 0)
+	r.account(req, res, status, took)
+	r.accessLog(req, res, status, took)
+	r.observeOutput(req, res, res.judged)
+	if req.logRetain > 0 {
+		go r.keepRequest(context.WithoutCancel(ctx), req, res, status)
+	}
+}
+
+// writeCachedStream sends a kept answer as the chunks a model streams, with
+// the usage chunk only to a client that asked for one.
+func writeCachedStream(w http.ResponseWriter, req *parsedRequest, body []byte) error {
+	events, err := streamFromAnswer(req.Path, body, clientWantsUsage(req))
+	if err != nil {
+		// Not an answer that can be streamed — a guard's replacement, say.
+		// Sent whole, which a client reads as a model that does not stream.
+		w.Header().Set("Content-Type", "application/json")
+		w.WriteHeader(http.StatusOK)
+		_, err = w.Write(body)
+		return err
+	}
+	w.Header().Set("Content-Type", "text/event-stream")
+	w.Header().Set("Cache-Control", "no-cache")
+	w.WriteHeader(http.StatusOK)
+	for _, e := range append(events, "[DONE]") {
+		if _, err := io.WriteString(w, "data: "+e+"\n\n"); err != nil {
+			return err
+		}
+	}
+	if f, ok := w.(http.Flusher); ok {
+		f.Flush()
+	}
+	return nil
+}
+
+// clientWantsUsage says whether the client asked for the usage chunk
+// itself, rather than the gateway asking for it on its behalf.
+func clientWantsUsage(req *parsedRequest) bool {
+	if req.SuppressUsageChunk {
+		return false
+	}
+	var b struct {
+		StreamOptions *struct {
+			IncludeUsage bool `json:"include_usage"`
+		} `json:"stream_options"`
+	}
+	return json.Unmarshal(req.UpstreamBody, &b) == nil && b.StreamOptions != nil && b.StreamOptions.IncludeUsage
+}
+
+// replyText is an answer's first choice as text, for the request log.
+func replyText(body []byte) string {
+	var c struct {
+		Choices []struct {
+			Message struct {
+				Content *string `json:"content"`
+			} `json:"message"`
+			Text *string `json:"text"`
+		} `json:"choices"`
+	}
+	if json.Unmarshal(body, &c) != nil || len(c.Choices) == 0 {
+		return ""
+	}
+	if s := c.Choices[0].Message.Content; s != nil {
+		return *s
+	}
+	if s := c.Choices[0].Text; s != nil {
+		return *s
+	}
+	return ""
+}

internal/router/cache_test.go

@@ -0,0 +1,615 @@
+package router
+
+import (
+	"bytes"
+	"context"
+	"crypto/sha256"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus/testutil"
+
+	"gitea.mynodes.xyz/ilker/llmgw/internal/guard"
+	"gitea.mynodes.xyz/ilker/llmgw/internal/router/proxy"
+	"gitea.mynodes.xyz/ilker/llmgw/internal/store"
+)
+
+// cachingRouter is pricedRouter with its project reusing answers for an
+// hour, reproducible ones only unless anyAnswer.
+func cachingRouter(t *testing.T, engineURL string, anyAnswer bool) (*Router, string) {
+	t.Helper()
+	rt, secret := pricedRouter(t, engineURL)
+	if err := rt.store.SetAnswerCache(t.Context(), mustProjectOf(t, rt, "key-1"), time.Hour, anyAnswer); err != nil {
+		t.Fatal(err)
+	}
+	return rt, secret
+}
+
+func ask(temperature any, text string) map[string]any {
+	b := map[string]any{"model": "q14", "messages": []any{map[string]any{"role": "user", "content": text}}}
+	if temperature != nil {
+		b["temperature"] = temperature
+	}
+	return b
+}
+
+// usageOf waits for the record of one request.
+func usageOf(t *testing.T, rt *Router, id string) store.Usage {
+	t.Helper()
+	deadline := time.Now().Add(5 * time.Second)
+	for time.Now().Before(deadline) {
+		if u, ok, err := rt.store.UsageByRequest(t.Context(), id); err == nil && ok {
+			return u
+		}
+		time.Sleep(20 * time.Millisecond)
+	}
+	t.Fatalf("no usage was recorded for %s", id)
+	return store.Usage{}
+}
+
+func TestAProjectThatReusesAnswersAnswersTheSameRequestWithoutTheModel(t *testing.T) {
+	e := newFakeEngine(t, "qwen2.5-14b-instruct")
+	rt, secret := cachingRouter(t, e.srv.URL, false)
+	h := rt.Handler()
+	auth := func(id string) map[string]string {
+		return map[string]string{"Authorization": "Bearer " + secret, "X-Request-Id": id}
+	}
+
+	first := post(t, h, ask(0, "what is 2+2"), auth("first"))
+	if first.Code != 200 || first.Header().Get(cacheHeader) != "miss" {
+		t.Fatalf("first: %d %q %s", first.Code, first.Header().Get(cacheHeader), first.Body.String())
+	}
+	// The same request, said differently: keys in another order, another
+	// end user, the temperature written as a float, by the canonical name.
+	again := map[string]any{"messages": []any{map[string]any{"content": "what is 2+2", "role": "user"}},
+		"temperature": 0.0, "model": "qwen2.5-14b-instruct", "user": "someone-else"}
+	second := post(t, h, again, auth("second"))
+	if second.Code != 200 || second.Header().Get(cacheHeader) != "hit" {
+		t.Fatalf("second: %d %q %s", second.Code, second.Header().Get(cacheHeader), second.Body.String())
+	}
+	if e.calls.Load() != 1 {
+		t.Errorf("the engine was asked %d times; the second answer should have been reused", e.calls.Load())
+	}
+	if second.Body.String() != first.Body.String() {
+		t.Errorf("reused %s, first was %s", second.Body.String(), first.Body.String())
+	}
+	if second.Header().Get("Age") == "" {
+		t.Error("a reused answer did not say its age")
+	}
+
+	// Recorded, as a request that generated nothing.
+	miss, hit := usageOf(t, rt, "first"), usageOf(t, rt, "second")
+	if miss.CacheHit || miss.CostMicros == 0 || miss.PromptTokens == 0 {
+		t.Errorf("the answer the model gave: %+v", miss)
+	}
+	if !hit.CacheHit || hit.CostMicros != 0 || hit.PromptTokens != 0 || hit.CompletionTokens != 0 ||
+		hit.Status != 200 || hit.ServedModel != "qwen2.5-14b-instruct" {
+		t.Errorf("the reused answer: %+v", hit)
+	}
+
+	// A sampled answer is asked for each time, and says so.
+	for i := range 2 {
+		rec := post(t, h, ask(0.7, "write a poem"), auth(fmt.Sprintf("poem-%d", i)))
+		if rec.Header().Get(cacheHeader) != "bypass" {
+			t.Errorf("a sampled request: %q", rec.Header().Get(cacheHeader))
+		}
+	}
+	if e.calls.Load() != 3 {
+		t.Errorf("the engine was asked %d times, expected each sampled request to reach it", e.calls.Load())
+	}
+	// A seed makes it reproducible.
+	seeded := ask(0.7, "write a poem")
+	seeded["seed"] = 42
+	post(t, h, seeded, auth("seed-1"))
+	if rec := post(t, h, seeded, auth("seed-2")); rec.Header().Get(cacheHeader) != "hit" {
+		t.Errorf("a seeded request: %q", rec.Header().Get(cacheHeader))
+	}
+	// A client that asks for a fresh answer gets one.
+	hdr := auth("fresh")
+	hdr["Cache-Control"] = "no-cache"
+	if rec := post(t, h, ask(0, "what is 2+2"), hdr); rec.Header().Get(cacheHeader) != "bypass" {
+		t.Errorf("Cache-Control: no-cache: %q", rec.Header().Get(cacheHeader))
+	}
+	hdr = auth("fresh-2")
+	hdr[cacheHeader] = "no"
+	if rec := post(t, h, ask(0, "what is 2+2"), hdr); rec.Header().Get(cacheHeader) != "bypass" {
+		t.Errorf("x-llmgw-cache: no: %q", rec.Header().Get(cacheHeader))
+	}
+
+	if n := testutil.ToFloat64(rt.m.cacheRequests.WithLabelValues("qwen2.5-14b-instruct", "hit")); n != 2 {
+		t.Errorf("%v hits counted", n)
+	}
+	if n := testutil.ToFloat64(rt.m.cacheRequests.WithLabelValues("qwen2.5-14b-instruct", "miss")); n != 2 {
+		t.Errorf("%v misses counted", n)
+	}
+	if n := testutil.ToFloat64(rt.m.cacheBytes); n <= 0 {
+		t.Errorf("the cache holds %v bytes", n)
+	}
+
+	// Turned off, nothing is reused, and nothing is said about it.
+	_ = rt.store.SetAnswerCache(t.Context(), mustProjectOf(t, rt, "key-1"), 0, false)
+	rt.reqLogs.by = nil
+	calls := e.calls.Load()
+	if rec := post(t, h, ask(0, "what is 2+2"), auth("off")); rec.Header().Get(cacheHeader) != "" || e.calls.Load() != calls+1 {
+		t.Errorf("a project that reuses nothing: %q, %d calls", rec.Header().Get(cacheHeader), e.calls.Load()-calls)
+	}
+}
+
+func TestAProjectThatReusesAnyAnswerReusesSampledOnes(t *testing.T) {
+	e := newFakeEngine(t, "qwen2.5-14b-instruct")
+	rt, secret := cachingRouter(t, e.srv.URL, true)
+	auth := map[string]string{"Authorization": "Bearer " + secret}
+	post(t, rt.Handler(), ask(nil, "write a poem"), auth)
+	if rec := post(t, rt.Handler(), ask(nil, "write a poem"), auth); rec.Header().Get(cacheHeader) != "hit" {
+		t.Errorf("%q", rec.Header().Get(cacheHeader))
+	}
+}
+
+func TestKeysOfOneProjectNeverSeeAnotherProjectsAnswers(t *testing.T) {
+	e := newFakeEngine(t, "qwen2.5-14b-instruct")
+	rt, secret := cachingRouter(t, e.srv.URL, false)
+	team, err := rt.store.CreateTeam(t.Context(), mustOrgOf(t, rt), "other")
+	if err != nil {
+		t.Fatal(err)
+	}
+	other, err := rt.store.CreateProject(t.Context(), team, "elsewhere")
+	if err != nil {
+		t.Fatal(err)
+	}
+	_ = rt.store.SetAnswerCache(t.Context(), other, time.Hour, true)
+	secret2 := store.KeyPrefix + strings.Repeat("b", 43)
+	sum := sha256.Sum256([]byte(secret2))
+	if err := rt.store.CreateKey(t.Context(), store.Key{ID: "key-2", Hash: sum[:], Prefix: "bbbb", Name: "other",
+		ProjectID: other, CreatedAt: time.Now()}); err != nil {
+		t.Fatal(err)
+	}
+
+	post(t, rt.Handler(), ask(0, "the secret plan"), map[string]string{"Authorization": "Bearer " + secret})
+	rec := post(t, rt.Handler(), ask(0, "the secret plan"), map[string]string{"Authorization": "Bearer " + secret2})
+	if rec.Header().Get(cacheHeader) != "miss" || e.calls.Load() != 2 {
+		t.Errorf("another project's key: %q, the engine asked %d times", rec.Header().Get(cacheHeader), e.calls.Load())
+	}
+}
+
+func mustOrgOf(t *testing.T, rt *Router) string {
+	t.Helper()
+	orgs, err := rt.store.Tree(t.Context())
+	if err != nil || len(orgs) == 0 {
+		t.Fatalf("no organization: %v", err)
+	}
+	return orgs[0].ID
+}
+
+func TestAFailedAnswerIsNotReused(t *testing.T) {
+	e := newFakeEngine(t, "qwen2.5-14b-instruct")
+	rt, secret := cachingRouter(t, e.srv.URL, false)
+	auth := map[string]string{"Authorization": "Bearer " + secret}
+	e.status.Store(http.StatusBadRequest)
+	if rec := post(t, rt.Handler(), ask(0, "hello"), auth); rec.Code == 200 {
+		t.Fatalf("%d", rec.Code)
+	}
+	e.status.Store(0)
+	if rec := post(t, rt.Handler(), ask(0, "hello"), auth); rec.Code != 200 || rec.Header().Get(cacheHeader) != "miss" {
+		t.Errorf("after a failure: %d %q", rec.Code, rec.Header().Get(cacheHeader))
+	}
+}
+
+// switchGuard blocks every request, or puts every one under no_store, when
+// told to.
+type switchGuard struct {
+	block, noStore atomic.Bool
+	// answers, when set, stops answers saying "hi".
+	answers atomic.Bool
+}
+
+func (g *switchGuard) Input(context.Context, guard.Subject, []byte) (guard.Verdict, error) {
+	if g.block.Load() {
+		return guard.Verdict{Action: guard.Block, Policy: "nothing"}, nil
+	}
+	return guard.Verdict{}, nil
+}
+
+func (g *switchGuard) Output(context.Context, guard.Subject) guard.OutputFilter {
+	if !g.answers.Load() {
+		return nil
+	}
+	return &stopHi{}
+}
+
+func (g *switchGuard) Directives(guard.Subject) guard.Directives {
+	return guard.Directives{NoStore: g.noStore.Load()}
+}
+
+func TestTheGuardrailsStillDecideAboutAnAnswerFromTheCache(t *testing.T) {
+	e := newFakeEngine(t, "qwen2.5-14b-instruct")
+	rt, secret := cachingRouter(t, e.srv.URL, false)
+	g := &switchGuard{}
+	rt.SetGuard(g)
+	auth := map[string]string{"Authorization": "Bearer " + secret}
+	if rec := post(t, rt.Handler(), ask(0, "hello"), auth); rec.Header().Get(cacheHeader) != "miss" {
+		t.Fatalf("%q", rec.Header().Get(cacheHeader))
+	}
+
+	// A policy that now refuses the request refuses it, answer kept or not.
+	g.block.Store(true)
+	if rec := post(t, rt.Handler(), ask(0, "hello"), auth); rec.Code != http.StatusBadRequest || rec.Header().Get(cacheHeader) != "" {
+		t.Errorf("a blocked request: %d %q %s", rec.Code, rec.Header().Get(cacheHeader), rec.Body.String())
+	}
+	g.block.Store(false)
+
+	// A policy that keeps nothing neither reads the cache nor fills it.
+	g.noStore.Store(true)
+	if rec := post(t, rt.Handler(), ask(0, "hello"), auth); rec.Code != 200 || rec.Header().Get(cacheHeader) != "bypass" {
+		t.Errorf("under no_store: %d %q", rec.Code, rec.Header().Get(cacheHeader))
+	}
+	if rec := post(t, rt.Handler(), ask(0, "under no_store"), auth); rec.Header().Get(cacheHeader) != "bypass" {
+		t.Errorf("under no_store: %q", rec.Header().Get(cacheHeader))
+	}
+	g.noStore.Store(false)
+	if rec := post(t, rt.Handler(), ask(0, "under no_store"), auth); rec.Header().Get(cacheHeader) != "miss" {
+		t.Errorf("an answer given under no_store was kept: %q", rec.Header().Get(cacheHeader))
+	}
+
+	// An output policy added since judges the reused answer too.
+	g.answers.Store(true)
+	rec := post(t, rt.Handler(), ask(0, "hello"), auth)
+	if rec.Header().Get(cacheHeader) != "hit" || !strings.Contains(rec.Body.String(), "content_filter") ||
+		rec.Header().Get(guardHeader) != "blocked" {
+		t.Errorf("an answer the output policy stops: %q %q %s", rec.Header().Get(cacheHeader),
+			rec.Header().Get(guardHeader), rec.Body.String())
+	}
+	// And an answer it stopped is not kept.
+	if rec := post(t, rt.Handler(), ask(0, "new question"), auth); rec.Header().Get(cacheHeader) != "miss" {
+		t.Fatalf("%q", rec.Header().Get(cacheHeader))
+	}
+	g.answers.Store(false)
+	if rec := post(t, rt.Handler(), ask(0, "new question"), auth); rec.Header().Get(cacheHeader) != "miss" {
+		t.Errorf("an answer a guard stopped was kept: %q", rec.Header().Get(cacheHeader))
+	}
+}
+
+// greetingEngine streams "Hello world" to a request that asks for a
+// stream, with usage when asked, and answers whole otherwise.
+func greetingEngine(t *testing.T) (*httptest.Server, *atomic.Int64) {
+	t.Helper()
+	var calls atomic.Int64
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path == "/metrics" {
+			return
+		}
+		calls.Add(1)
+		var b struct {
+			Stream        bool `json:"stream"`
+			StreamOptions struct {
+				IncludeUsage bool `json:"include_usage"`
+			} `json:"stream_options"`
+		}
+		body, _ := io.ReadAll(r.Body)
+		_ = json.Unmarshal(body, &b)
+		if !b.Stream {
+			w.Header().Set("Content-Type", "application/json")
+			_, _ = io.WriteString(w, `{"id":"c1","object":"chat.completion","created":1,"model":"qwen2.5-14b-instruct",`+
+				`"choices":[{"index":0,"message":{"role":"assistant","content":"Hello world"},"finish_reason":"stop"}],`+
+				`"usage":{"prompt_tokens":5,"completion_tokens":2,"total_tokens":7}}`)
+			return
+		}
+		w.Header().Set("Content-Type", "text/event-stream")
+		head := `{"id":"c1","object":"chat.completion.chunk","created":1,"model":"qwen2.5-14b-instruct","choices":`
+		for _, c := range []string{
+			`[{"index":0,"delta":{"role":"assistant","content":""},"finish_reason":null}]}`,
+			`[{"index":0,"delta":{"content":"Hello"},"finish_reason":null}]}`,
+			`[{"index":0,"delta":{"content":" world"},"finish_reason":null}]}`,
+			`[{"index":0,"delta":{},"finish_reason":"stop"}]}`,
+		} {
+			fmt.Fprintf(w, "data: %s%s\n\n", head, c)
+		}
+		if b.StreamOptions.IncludeUsage {
+			fmt.Fprintf(w, "data: %s[],\"usage\":{\"prompt_tokens\":5,\"completion_tokens\":2,\"total_tokens\":7}}\n\n", head)
+		}
+		_, _ = io.WriteString(w, "data: [DONE]\n\n")
+	}))
+	t.Cleanup(srv.Close)
+	return srv, &calls
+}
+
+// streamedText is the content of a streamed chat answer, and whether it
+// carried usage and ended with [DONE].
+func streamedText(t *testing.T, sse string) (text string, usage, done bool) {
+	t.Helper()
+	for _, line := range strings.Split(sse, "\n") {
+		data, ok := strings.CutPrefix(line, "data: ")
+		if !ok {
+			continue
+		}
+		if data == "[DONE]" {
+			done = true
+			continue
+		}
+		var c struct {
+			Object  string `json:"object"`
+			Choices []struct {
+				Delta struct {
+					Content string `json:"content"`
+				} `json:"delta"`
+			} `json:"choices"`
+			Usage *struct {
+				CompletionTokens int `json:"completion_tokens"`
+			} `json:"usage"`
+		}
+		if err := json.Unmarshal([]byte(data), &c); err != nil || c.Object != "chat.completion.chunk" {
+			t.Fatalf("not a chunk: %q", data)
+		}
+		for _, ch := range c.Choices {
+			text += ch.Delta.Content
+		}
+		if c.Usage != nil && c.Usage.CompletionTokens == 2 {
+			usage = true
+		}
+	}
+	return text, usage, done
+}
+
+func TestAStreamedAnswerIsReusedForClientsThatStreamAndThatDoNot(t *testing.T) {
+	srv, calls := greetingEngine(t)
+	rt, secret := cachingRouter(t, srv.URL, false)
+	auth := map[string]string{"Authorization": "Bearer " + secret}
+	streamed := func(includeUsage bool) map[string]any {
+		b := ask(0, "greet me")
+		b["stream"] = true
+		if includeUsage {
+			b["stream_options"] = map[string]any{"include_usage": true}
+		}
+		return b
+	}
+
+	first := post(t, rt.Handler(), streamed(false), auth)
+	if first.Header().Get(cacheHeader) != "miss" {
+		t.Fatalf("%q %s", first.Header().Get(cacheHeader), first.Body.String())
+	}
+	if text, _, _ := streamedText(t, first.Body.String()); text != "Hello world" {
+		t.Fatalf("streamed %q", text)
+	}
+
+	// Whole, to a client that does not stream.
+	whole := post(t, rt.Handler(), ask(0, "greet me"), auth)
+	var c struct {
+		Object  string `json:"object"`
+		Choices []struct {
+			Message struct {
+				Role, Content string
+			} `json:"message"`
+			FinishReason string `json:"finish_reason"`
+		} `json:"choices"`
+		Usage struct {
+			PromptTokens, CompletionTokens int
+		} `json:"usage"`
+	}
+	if err := json.Unmarshal(whole.Body.Bytes(), &c); err != nil || whole.Header().Get(cacheHeader) != "hit" {
+		t.Fatalf("%q %s %v", whole.Header().Get(cacheHeader), whole.Body.String(), err)
+	}
+	if c.Object != "chat.completion" || len(c.Choices) != 1 || c.Choices[0].Message.Content != "Hello world" ||
+		c.Choices[0].Message.Role != "assistant" || c.Choices[0].FinishReason != "stop" {
+		t.Errorf("put back together as %s", whole.Body.String())
+	}
+
+	// Streamed again, with the usage only to a client that asks for it.
+	for _, withUsage := range []bool{false, true} {
+		rec := post(t, rt.Handler(), streamed(withUsage), auth)
+		text, usage, done := streamedText(t, rec.Body.String())
+		if rec.Header().Get(cacheHeader) != "hit" || text != "Hello world" || !done || usage != withUsage ||
+			!strings.HasPrefix(rec.Header().Get("Content-Type"), "text/event-stream") {
+			t.Errorf("streamed from the cache (usage %v): %q %q usage=%v done=%v\n%s", withUsage,
+				rec.Header().Get(cacheHeader), text, usage, done, rec.Body.String())
+		}
+	}
+	if calls.Load() != 1 {
+		t.Errorf("the engine was asked %d times", calls.Load())
+	}
+}
+
+func TestManyClientsReadingAndFillingTheCacheAtOnce(t *testing.T) {
+	e := newFakeEngine(t, "qwen2.5-14b-instruct")
+	rt, secret := cachingRouter(t, e.srv.URL, false)
+	h := rt.Handler()
+	auth := map[string]string{"Authorization": "Bearer " + secret}
+	var wg sync.WaitGroup
+	for i := range 40 {
+		wg.Go(func() {
+			rec := post(t, h, ask(0, fmt.Sprintf("question %d", i%5)), auth)
+			if rec.Code != 200 {
+				t.Errorf("%d %s", rec.Code, rec.Body.String())
+			}
+		})
+	}
+	wg.Wait()
+	for i := range 5 {
+		if rec := post(t, h, ask(0, fmt.Sprintf("question %d", i)), auth); rec.Header().Get(cacheHeader) != "hit" {
+			t.Errorf("question %d: %q", i, rec.Header().Get(cacheHeader))
+		}
+	}
+}
+
+func TestTheAnswerCacheDropsWhatWasUsedLongestAgoPastItsBudget(t *testing.T) {
+	var held int64
+	c := newAnswerCache(1000, 400, func(b int64, _ int) { held = b })
+	now := time.Now()
+	put := func(key string) bool {
+		return c.put(&cachedAnswer{key: key, project: "p", body: bytes.Repeat([]byte("x"), 100),
+			at: now, expires: now.Add(time.Hour)}, now)
+	}
+	for _, k := range []string{"a", "b", "c"} {
+		put(k)
+	}
+	c.get("a", time.Hour, now) // a is now the most recently used
+	put("d")
+	put("e")
+	if c.get("b", time.Hour, now) != nil {
+		t.Error("b, used longest ago, was kept past the budget")
+	}
+	if c.get("a", time.Hour, now) == nil {
+		t.Error("a, used recently, was dropped")
+	}
+	if held > 1000 {
+		t.Errorf("holding %d bytes of 1000", held)
+	}
+	if c.put(&cachedAnswer{key: "big", body: bytes.Repeat([]byte("x"), 500)}, now) {
+		t.Error("an answer larger than an entry may be was kept")
+	}
+	// Past its TTL, or past a TTL shortened since, it is gone.
+	if c.get("a", time.Hour, now.Add(2*time.Hour)) != nil {
+		t.Error("an expired answer was returned")
+	}
+	if c.get("e", time.Minute, now.Add(2*time.Minute)) != nil {
+		t.Error("an answer older than the project's TTL now was returned")
+	}
+}
+
+func TestAnAnswerIsLookedUpByWhatChangesIt(t *testing.T) {
+	key := func(project, model, body string) string {
+		k, ok := answerKey(project, "/v1/chat/completions", model, []byte(body))
+		if !ok {
+			t.Fatalf("no key for %s", body)
+		}
+		return k
+	}
+	base := key("p", "m", `{"model":"m","messages":[{"role":"user","content":"hi"}],"temperature":0}`)
+	same := []string{
+		`{"messages":[{"content":"hi","role":"user"}],"temperature":0.0,"model":"alias"}`,
+		`{"model":"m","messages":[{"role":"user","content":"hi"}],"temperature":0,"stream":true,"stream_options":{"include_usage":true}}`,
+		`{"model":"m","messages":[{"role":"user","content":"hi"}],"temperature":0e0,"user":"u","metadata":{"a":"b"}}`,
+	}
+	for _, b := range same {
+		if key("p", "m", b) != base {
+			t.Errorf("%s was looked up apart", b)
+		}
+	}
+	different := []string{
+		`{"model":"m","messages":[{"role":"user","content":"hi!"}],"temperature":0}`,
+		`{"model":"m","messages":[{"role":"user","content":"hi"}],"temperature":0,"max_tokens":5}`,
+		`{"model":"m","messages":[{"role":"user","content":"hi"}],"temperature":0,"seed":9007199254740993}`,
+	}
+	for _, b := range different {
+		if key("p", "m", b) == base {
+			t.Errorf("%s was looked up as the same", b)
+		}
+	}
+	if key("p", "m", `{"seed":9007199254740993}`) == key("p", "m", `{"seed":9007199254740992}`) {
+		t.Error("two large seeds were taken for one")
+	}
+	if key("other", "m", `{"model":"m","messages":[{"role":"user","content":"hi"}],"temperature":0}`) == base {
+		t.Error("another project's request was looked up as the same")
+	}
+	if key("p", "other", `{"model":"m","messages":[{"role":"user","content":"hi"}],"temperature":0}`) == base {
+		t.Error("another model's request was looked up as the same")
+	}
+	if _, ok := answerKey("p", "/v1/chat/completions", "m", []byte(`not json`)); ok {
+		t.Error("a body that is not JSON was given a key")
+	}
+}
+
+func TestAStreamedAnswerIsPutBackTogetherAndTakenApartAgain(t *testing.T) {
+	sse := strings.Join([]string{
+		`data: {"id":"c","object":"chat.completion.chunk","created":7,"model":"m","choices":[{"index":0,"delta":{"role":"assistant"},"finish_reason":null}]}`,
+		`data: {"id":"c","object":"chat.completion.chunk","created":7,"model":"m","choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"id":"call_1","type":"function","function":{"name":"get_weather","arguments":""}}]},"finish_reason":null}]}`,
+		`data: {"id":"c","object":"chat.completion.chunk","created":7,"model":"m","choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"function":{"arguments":"{\"city\":"}}]},"finish_reason":null}]}`,
+		`data: {"id":"c","object":"chat.completion.chunk","created":7,"model":"m","choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"function":{"arguments":"\"Paris\"}"}}]},"finish_reason":null}]}`,
+		`data: {"id":"c","object":"chat.completion.chunk","created":7,"model":"m","choices":[{"index":0,"delta":{},"finish_reason":"tool_calls"}]}`,
+		`data: [DONE]`,
+	}, "\n\n") + "\n\n"
+	body, ok := assembleStream("/v1/chat/completions", []byte(sse), &proxy.Usage{PromptTokens: 3, CompletionTokens: 4})
+	if !ok {
+		t.Fatal("not put back together")
+	}
+	var c struct {
+		Choices []struct {
+			Message struct {
+				Content   *string    `json:"content"`
+				ToolCalls []toolCall `json:"tool_calls"`
+			} `json:"message"`
+			FinishReason string `json:"finish_reason"`
+		} `json:"choices"`
+		Usage struct {
+			TotalTokens int `json:"total_tokens"`
+		} `json:"usage"`
+	}
+	if err := json.Unmarshal(body, &c); err != nil {
+		t.Fatal(err)
+	}
+	if len(c.Choices) != 1 || c.Choices[0].Message.Content != nil || len(c.Choices[0].Message.ToolCalls) != 1 ||
+		c.Choices[0].Message.ToolCalls[0].Function.Arguments != `{"city":"Paris"}` ||
+		c.Choices[0].Message.ToolCalls[0].ID != "call_1" || c.Choices[0].FinishReason != "tool_calls" ||
+		c.Usage.TotalTokens != 7 {
+		t.Errorf("put back together as %s", body)
+	}
+
+	// Taken apart, and put back together again, it is the same answer.
+	events, err := streamFromAnswer("/v1/chat/completions", body, false)
+	if err != nil {
+		t.Fatal(err)
+	}
+	again, ok := assembleStream("/v1/chat/completions",
+		[]byte("data: "+strings.Join(append(events, "[DONE]"), "\n\ndata: ")+"\n\n"), &proxy.Usage{PromptTokens: 3, CompletionTokens: 4})
+	if !ok || !bytes.Equal(again, body) {
+		t.Errorf("round trip:\n%s\n%s", body, again)
+	}
+
+	// What it does not understand, and a stream cut short, are not kept.
+	for _, bad := range []string{
+		strings.Replace(sse, `"delta":{}`, `"delta":{"audio":{"id":"a"}}`, 1),
+		strings.Replace(sse, `"finish_reason":null}]}`, `"finish_reason":null,"logprobs":{"content":[]}}]}`, 1),
+		strings.Replace(sse, "data: [DONE]", "", 1),
+	} {
+		if _, ok := assembleStream("/v1/chat/completions", []byte(bad), nil); ok {
+			t.Errorf("kept: %s", bad)
+		}
+	}
+
+	// A completion, too.
+	text := `data: {"id":"t","object":"text_completion","created":1,"model":"m","choices":[{"index":0,"text":"Hel","finish_reason":null}]}` + "\n\n" +
+		`data: {"id":"t","object":"text_completion","created":1,"model":"m","choices":[{"index":0,"text":"lo","finish_reason":"length"}]}` + "\n\n" +
+		"data: [DONE]\n\n"
+	tb, ok := assembleStream("/v1/completions", []byte(text), nil)
+	if !ok || !strings.Contains(string(tb), `"text":"Hello"`) || !strings.Contains(string(tb), `"object":"text_completion"`) {
+		t.Errorf("a completion: %s", tb)
+	}
+}
+
+func TestAnAnswerFromTheCacheReachesAnthropicClientsInTheirShape(t *testing.T) {
+	srv, calls := greetingEngine(t)
+	rt, secret := cachingRouter(t, srv.URL, false)
+	send := func(stream bool) *httptest.ResponseRecorder {
+		b, _ := json.Marshal(map[string]any{"model": "q14", "max_tokens": 50, "temperature": 0, "stream": stream,
+			"messages": []any{map[string]any{"role": "user", "content": "greet me"}}})
+		req := httptest.NewRequest(http.MethodPost, "/v1/messages", bytes.NewReader(b))
+		req.Header.Set("Content-Type", "application/json")
+		req.Header.Set("x-api-key", secret)
+		rec := httptest.NewRecorder()
+		rt.Handler().ServeHTTP(rec, req)
+		return rec
+	}
+	first := send(true)
+	if first.Code != 200 || !strings.Contains(first.Body.String(), "Hello") {
+		t.Fatalf("%d %s", first.Code, first.Body.String())
+	}
+	whole := send(false)
+	if whole.Code != 200 || !strings.Contains(whole.Body.String(), `"text":"Hello world"`) {
+		t.Errorf("whole: %d %s", whole.Code, whole.Body.String())
+	}
+	streamed := send(true)
+	if !strings.Contains(streamed.Body.String(), "content_block_delta") ||
+		!strings.Contains(streamed.Body.String(), "Hello world") ||
+		!strings.Contains(streamed.Body.String(), "message_stop") {
+		t.Errorf("streamed: %s", streamed.Body.String())
+	}
+	if calls.Load() != 1 {
+		t.Errorf("the engine was asked %d times", calls.Load())
+	}
+}

internal/router/cachestream.go

@@ -0,0 +1,364 @@
+package router
+
+import (
+	"bufio"
+	"bytes"
+	"encoding/json"
+	"errors"
+	"strings"
+
+	"gitea.mynodes.xyz/ilker/llmgw/internal/router/proxy"
+)
+
+// A cached answer is kept as one JSON object whichever way it was asked
+// for, so a client that streams and one that does not share it. A streamed
+// answer is put back together into the object a model would have sent
+// whole, and an answer is taken apart into chunks for a client that
+// streams.
+
+// streamChunk is one streamed chunk of a chat or a completion.
+type streamChunk struct {
+	ID                string `json:"id"`
+	Object            string `json:"object"`
+	Created           int64  `json:"created"`
+	Model             string `json:"model"`
+	SystemFingerprint string `json:"system_fingerprint"`
+	Choices           []struct {
+		Index int `json:"index"`
+		// Delta is a chat chunk's, as fields: those not understood here
+		// leave the answer unkept rather than kept without them.
+		Delta        map[string]json.RawMessage `json:"delta"`
+		Text         *string                    `json:"text"`
+		FinishReason *string                    `json:"finish_reason"`
+		Logprobs     json.RawMessage            `json:"logprobs"`
+	} `json:"choices"`
+}
+
+// toolCallDelta is part of one tool call, as streamed.
+type toolCallDelta struct {
+	Index    int    `json:"index"`
+	ID       string `json:"id"`
+	Type     string `json:"type"`
+	Function struct {
+		Name      string `json:"name"`
+		Arguments string `json:"arguments"`
+	} `json:"function"`
+}
+
+type toolCall struct {
+	ID       string `json:"id"`
+	Type     string `json:"type"`
+	Function struct {
+		Name      string `json:"name"`
+		Arguments string `json:"arguments"`
+	} `json:"function"`
+}
+
+// assembledChoice is one choice being put back together.
+type assembledChoice struct {
+	role                     string
+	content, reasoning, text strings.Builder
+	hasContent, hasReasoning bool
+	tools                    []*toolCall
+	finish                   *string
+}
+
+// assembleStream puts a streamed chat or completion back together as the
+// object the model would have answered without streaming, with usage as the
+// engine reported it. ok is false for anything it does not fully
+// understand — logprobs, audio, a chunk that is not JSON — which is then
+// not kept: an answer kept with something missing would be served short.
+func assembleStream(path string, sse []byte, usage *proxy.Usage) ([]byte, bool) {
+	chat := path == "/v1/chat/completions"
+	if !chat && path != "/v1/completions" {
+		return nil, false
+	}
+	var head streamChunk
+	choices := map[int]*assembledChoice{}
+	maxIndex := -1
+	done := false
+	sc := bufio.NewScanner(bytes.NewReader(sse))
+	sc.Buffer(make([]byte, 64<<10), len(sse)+1)
+	for sc.Scan() {
+		data, ok := strings.CutPrefix(sc.Text(), "data:")
+		if !ok {
+			continue
+		}
+		data = strings.TrimSpace(data)
+		if data == "[DONE]" {
+			done = true
+			continue
+		}
+		var c streamChunk
+		if json.Unmarshal([]byte(data), &c) != nil {
+			return nil, false
+		}
+		if head.ID == "" {
+			head.ID, head.Created, head.Model = c.ID, c.Created, c.Model
+		}
+		if c.SystemFingerprint != "" {
+			head.SystemFingerprint = c.SystemFingerprint
+		}
+		for _, ch := range c.Choices {
+			if len(ch.Logprobs) > 0 && string(ch.Logprobs) != "null" {
+				return nil, false
+			}
+			a := choices[ch.Index]
+			if a == nil {
+				a = &assembledChoice{}
+				choices[ch.Index] = a
+				maxIndex = max(maxIndex, ch.Index)
+			}
+			if ch.FinishReason != nil {
+				a.finish = ch.FinishReason
+			}
+			if ch.Text != nil {
+				a.text.WriteString(*ch.Text)
+			}
+			if !addDelta(a, ch.Delta) {
+				return nil, false
+			}
+		}
+	}
+	if sc.Err() != nil || !done || len(choices) == 0 || len(choices) != maxIndex+1 {
+		// A stream that never said it was done may have been cut short.
+		return nil, false
+	}
+	out := map[string]any{"id": head.ID, "created": head.Created, "model": head.Model}
+	if head.SystemFingerprint != "" {
+		out["system_fingerprint"] = head.SystemFingerprint
+	}
+	list := make([]map[string]any, 0, len(choices))
+	for i := 0; i <= maxIndex; i++ {
+		a := choices[i]
+		if a.finish == nil {
+			return nil, false
+		}
+		c := map[string]any{"index": i, "finish_reason": *a.finish, "logprobs": nil}
+		if chat {
+			role := a.role
+			if role == "" {
+				role = "assistant"
+			}
+			msg := map[string]any{"role": role, "content": nil}
+			if a.hasContent || len(a.tools) == 0 {
+				msg["content"] = a.content.String()
+			}
+			if a.hasReasoning {
+				msg["reasoning_content"] = a.reasoning.String()
+			}
+			if len(a.tools) > 0 {
+				msg["tool_calls"] = a.tools
+			}
+			c["message"] = msg
+		} else {
+			c["text"] = a.text.String()
+		}
+		list = append(list, c)
+	}
+	out["choices"] = list
+	if chat {
+		out["object"] = "chat.completion"
+	} else {
+		out["object"] = "text_completion"
+	}
+	if usage != nil {
+		out["usage"] = map[string]int{"prompt_tokens": usage.PromptTokens,
+			"completion_tokens": usage.CompletionTokens,
+			"total_tokens":      usage.PromptTokens + usage.CompletionTokens}
+	}
+	b, err := json.Marshal(out)
+	return b, err == nil
+}
+
+// addDelta adds one chat chunk's delta to its choice, and says whether it
+// understood every field of it.
+func addDelta(a *assembledChoice, delta map[string]json.RawMessage) bool {
+	for k, raw := range delta {
+		if string(raw) == "null" {
+			continue
+		}
+		switch k {
+		case "role":
+			if json.Unmarshal(raw, &a.role) != nil {
+				return false
+			}
+		case "content", "reasoning_content":
+			var s string
+			if json.Unmarshal(raw, &s) != nil {
+				return false
+			}
+			// Empty says nothing: the first chunk carries the role with an
+			// empty content even when the answer is a tool call.
+			if s == "" {
+				continue
+			}
+			if k == "content" {
+				a.content.WriteString(s)
+				a.hasContent = true
+			} else {
+				a.reasoning.WriteString(s)
+				a.hasReasoning = true
+			}
+		case "refusal":
+			// An empty refusal is a field some engines always send.
+			var s string
+			if json.Unmarshal(raw, &s) != nil || s != "" {
+				return false
+			}
+		case "tool_calls":
+			var parts []toolCallDelta
+			if json.Unmarshal(raw, &parts) != nil {
+				return false
+			}
+			for _, p := range parts {
+				for len(a.tools) <= p.Index {
+					a.tools = append(a.tools, &toolCall{})
+				}
+				t := a.tools[p.Index]
+				if p.ID != "" {
+					t.ID = p.ID
+				}
+				if p.Type != "" {
+					t.Type = p.Type
+				}
+				t.Function.Name += p.Function.Name
+				t.Function.Arguments += p.Function.Arguments
+			}
+		default:
+			return false
+		}
+	}
+	return true
+}
+
+// streamFromAnswer is a kept chat or completion as the chunks a model would
+// have streamed: per choice, its whole text in one chunk and then its
+// finish reason, and the usage last for a client that asks for it.
+func streamFromAnswer(path string, body []byte, withUsage bool) ([]string, error) {
+	var a struct {
+		ID                string          `json:"id"`
+		Created           int64           `json:"created"`
+		Model             string          `json:"model"`
+		SystemFingerprint string          `json:"system_fingerprint"`
+		Usage             json.RawMessage `json:"usage"`
+		Choices           []struct {
+			Index        int             `json:"index"`
+			FinishReason json.RawMessage `json:"finish_reason"`
+			Logprobs     json.RawMessage `json:"logprobs"`
+			Text         *string         `json:"text"`
+			Message      *struct {
+				Role             string            `json:"role"`
+				Content          *string           `json:"content"`
+				ReasoningContent *string           `json:"reasoning_content"`
+				ToolCalls        []json.RawMessage `json:"tool_calls"`
+			} `json:"message"`
+		} `json:"choices"`
+	}
+	if err := json.Unmarshal(body, &a); err != nil {
+		return nil, err
+	}
+	if len(a.Choices) == 0 {
+		return nil, errors.New("an answer with no choices")
+	}
+	chat := path == "/v1/chat/completions"
+	object := "text_completion"
+	if chat {
+		object = "chat.completion.chunk"
+	}
+	chunk := func(choices []map[string]any) (string, error) {
+		c := map[string]any{"id": a.ID, "object": object, "created": a.Created, "model": a.Model,
+			"choices": choices}
+		if a.SystemFingerprint != "" {
+			c["system_fingerprint"] = a.SystemFingerprint
+		}
+		b, err := json.Marshal(c)
+		return string(b), err
+	}
+	var events []string
+	add := func(choices []map[string]any) error {
+		e, err := chunk(choices)
+		events = append(events, e)
+		return err
+	}
+	for _, ch := range a.Choices {
+		var logprobs any
+		if len(ch.Logprobs) > 0 {
+			logprobs = ch.Logprobs
+		}
+		finish := any(json.RawMessage("null"))
+		if len(ch.FinishReason) > 0 {
+			finish = ch.FinishReason
+		}
+		if !chat {
+			if ch.Text == nil {
+				return nil, errors.New("a completion with no text")
+			}
+			if err := add([]map[string]any{{"index": ch.Index, "text": *ch.Text, "logprobs": logprobs,
+				"finish_reason": nil}}); err != nil {
+				return nil, err
+			}
+			if err := add([]map[string]any{{"index": ch.Index, "text": "", "finish_reason": finish}}); err != nil {
+				return nil, err
+			}
+			continue
+		}
+		m := ch.Message
+		if m == nil {
+			return nil, errors.New("a chat answer with no message")
+		}
+		role := m.Role
+		if role == "" {
+			role = "assistant"
+		}
+		if err := add([]map[string]any{{"index": ch.Index, "delta": map[string]any{"role": role, "content": ""},
+			"finish_reason": nil}}); err != nil {
+			return nil, err
+		}
+		if m.ReasoningContent != nil && *m.ReasoningContent != "" {
+			if err := add([]map[string]any{{"index": ch.Index,
+				"delta": map[string]any{"reasoning_content": *m.ReasoningContent}, "finish_reason": nil}}); err != nil {
+				return nil, err
+			}
+		}
+		if m.Content != nil && *m.Content != "" || logprobs != nil {
+			content := ""
+			if m.Content != nil {
+				content = *m.Content
+			}
+			if err := add([]map[string]any{{"index": ch.Index, "delta": map[string]any{"content": content},
+				"logprobs": logprobs, "finish_reason": nil}}); err != nil {
+				return nil, err
+			}
+		}
+		if len(m.ToolCalls) > 0 {
+			calls := make([]map[string]any, 0, len(m.ToolCalls))
+			for i, raw := range m.ToolCalls {
+				var tc map[string]any
+				if err := json.Unmarshal(raw, &tc); err != nil {
+					return nil, err
+				}
+				tc["index"] = i
+				calls = append(calls, tc)
+			}
+			if err := add([]map[string]any{{"index": ch.Index, "delta": map[string]any{"tool_calls": calls},
+				"finish_reason": nil}}); err != nil {
+				return nil, err
+			}
+		}
+		if err := add([]map[string]any{{"index": ch.Index, "delta": map[string]any{},
+			"finish_reason": finish}}); err != nil {
+			return nil, err
+		}
+	}
+	if withUsage && len(a.Usage) > 0 && string(a.Usage) != "null" {
+		c := map[string]any{"id": a.ID, "object": object, "created": a.Created, "model": a.Model,
+			"choices": []any{}, "usage": a.Usage}
+		b, err := json.Marshal(c)
+		if err != nil {
+			return nil, err
+		}
+		events = append(events, string(b))
+	}
+	return events, nil
+}

internal/router/handlers.go

@@ -366,6 +366,9 @@ type parsedRequest struct {
 	// internal is a request the gateway makes of itself — a judge's — which
 	// no guard filters.
 	internal bool
+	// cache is where its answer is kept, when its project reuses answers
+	// and this one may be.
+	cache cacheState
 }
 
 // inferenceBody covers the fields the router reads across all inference
@@ -531,8 +534,17 @@ func (r *Router) handleInference(path string) http.HandlerFunc {
 		}
 
 		r.decideRequestLog(req.Context(), parsed)
-		res := r.dispatch(req.Context(), w, parsed)
+		// After the guardrails, so a policy still decides whether a request is
+		// answered, and the request looked up is the one the model would be
+		// sent.
+		if a := r.cacheLookup(req.Context(), w, parsed); a != nil {
+			r.serveCached(req.Context(), w, parsed, a, start)
+			return
+		}
+		dw, cw := r.cacheWriterFor(w, parsed)
+		res := r.dispatch(req.Context(), dw, parsed)
 		status := r.finish(w, parsed, res, start)
+		r.cacheKeep(parsed, res, status, cw)
 		r.observeCanary(parsed, res, status, time.Since(start))
 		if parsed.logRetain > 0 {
 			go r.keepRequest(context.WithoutCancel(req.Context()), parsed, res, status)
@@ -645,6 +657,9 @@ func (r *Router) account(req *parsedRequest, res dispatchResult, status int, too
 		EndUser: req.EndUser, Tags: req.Tags,
 	}
 	guardFields(&u, req.guardIn, res.guard)
+	// An answer reused generates nothing: its usage is the zero the hit
+	// carries, and it costs nothing against a budget or a token limit.
+	u.CacheHit = res.cached
 	if u.ServedModel == "" {
 		u.ServedModel = req.Model
 	}
@@ -1169,6 +1184,9 @@ func (r *Router) accessLog(req *parsedRequest, res dispatchResult, status int, t
 	if res.outcome.Stopped {
 		attrs = append(attrs, "stopped_by_guard", true)
 	}
+	if res.cached {
+		attrs = append(attrs, "cache", "hit")
+	}
 	r.log.Info("request", attrs...)
 }
 

internal/router/metrics.go

@@ -31,6 +31,12 @@ type metrics struct {
 
 	prefixDecisions *prometheus.CounterVec
 
+	// cacheRequests counts what the answer cache did, cacheBytes and
+	// cacheEntries what it holds.
+	cacheRequests *prometheus.CounterVec
+	cacheBytes    prometheus.Gauge
+	cacheEntries  prometheus.Gauge
+
 	tenant *tenantMetrics
 	guard  *guardMetrics
 	token  *tokenMetrics
@@ -91,6 +97,20 @@ func newMetrics() *metrics {
 			Name: "llmgw_admission_rejected_total",
 			Help: "Requests refused by flow control, by reason.",
 		}, []string{"band", "reason"}),
+		cacheRequests: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Name: "llmgw_cache_requests_total",
+			Help: "Requests of projects that reuse answers, by what the answer cache did: " +
+				"hit (answered from it), miss (asked the model, and kept the answer if it could), " +
+				"bypass (not one whose answer is reused).",
+		}, []string{"model", "result"}),
+		cacheBytes: prometheus.NewGauge(prometheus.GaugeOpts{
+			Name: "llmgw_cache_bytes",
+			Help: "Memory the answers kept for reuse take in this process, roughly.",
+		}),
+		cacheEntries: prometheus.NewGauge(prometheus.GaugeOpts{
+			Name: "llmgw_cache_entries",
+			Help: "Answers kept for reuse in this process.",
+		}),
 		tenant: newTenantMetrics(),
 		guard:  newGuardMetrics(),
 		token:  newTokenMetrics(),
@@ -103,6 +123,7 @@ func (m *metrics) collectors() []prometheus.Collector {
 		m.requests, m.duration, m.ttft, m.tokens,
 		m.scorerValue, m.filtered, m.considered,
 		m.admissionWait, m.admissionRejected, m.prefixDecisions,
+		m.cacheRequests, m.cacheBytes, m.cacheEntries,
 	}, slices.Concat(m.tenant.collectors(), m.guard.collectors(), m.token.collectors())...)
 }
 

internal/router/requestlog.go

@@ -28,7 +28,9 @@ type requestLogs struct {
 type cachedLog struct {
 	l         storepkg.RequestLog
 	inCluster bool
-	at        time.Time
+	// cache is how the project reuses answers.
+	cache storepkg.AnswerCache
+	at    time.Time
 }
 
 // requestLogOf is a project's setting, read at most every requestLogTTL.
@@ -49,7 +51,8 @@ func (r *Router) requestLogOf(ctx context.Context, project string) storepkg.Requ
 	return c.l
 }
 
-// readProjectPolicy reads a project's keeping and residency, and caches them.
+// readProjectPolicy reads a project's keeping, residency and reuse of
+// answers, and caches them.
 func (r *Router) readProjectPolicy(ctx context.Context, project string) (cachedLog, error) {
 	l, err := r.store.RequestLogOf(ctx, project)
 	if err != nil {
@@ -59,7 +62,11 @@ func (r *Router) readProjectPolicy(ctx context.Context, project string) (cachedL
 	if err != nil {
 		return cachedLog{}, err
 	}
-	c := cachedLog{l: l, inCluster: in, at: time.Now()}
+	ac, err := r.store.AnswerCacheOf(ctx, project)
+	if err != nil {
+		return cachedLog{}, err
+	}
+	c := cachedLog{l: l, inCluster: in, cache: ac, at: time.Now()}
 	r.reqLogs.mu.Lock()
 	if r.reqLogs.by == nil {
 		r.reqLogs.by = map[string]cachedLog{}

internal/router/router.go

@@ -74,6 +74,8 @@ type Router struct {
 	batchWake chan struct{}
 	// reqLogs is each project's keeping of requests, as last read.
 	reqLogs requestLogs
+	// answers are the answers projects reuse; nil reuses none.
+	answers *answerCache
 
 	// extraMetrics are collectors other components publish on /metrics.
 	extraMetrics []prometheus.Collector
@@ -264,6 +266,11 @@ func New(cfg config.Server, log *slog.Logger) (*Router, error) {
 			MaxResponseBytes:      cfg.Router.MaxBodyBytes,
 		}),
 	}
+	r.answers = newAnswerCache(cfg.Router.Cache.MaxBytes, cfg.Router.Cache.MaxEntryBytes,
+		func(bytes int64, n int) {
+			metrics.cacheBytes.Set(float64(bytes))
+			metrics.cacheEntries.Set(float64(n))
+		})
 	r.poller = scrape.NewPoller(reg, scrape.Config{
 		Interval: cfg.Router.ScrapeInterval,
 		Timeout:  cfg.Router.ScrapeTimeout,
@@ -765,6 +772,9 @@ type dispatchResult struct {
 	// the candidate, and onCanary whether the candidate served it.
 	canaryOf, canary string
 	onCanary         bool
+	// cached is an answer from the project's answer cache, which no model
+	// was asked for.
+	cached bool
 }
 
 type servedModel struct {

internal/store/answercache.go

@@ -0,0 +1,59 @@
+package store
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"fmt"
+	"time"
+)
+
+// AnswerCache is how a project reuses answers: for how long one is served
+// again to the same request, and whether that is any request's answer or
+// only one that would come out the same every time. Zero TTL reuses none.
+//
+// Only the setting lives in the store. The answers are held in each
+// gateway's memory, so turning this on writes no text anywhere.
+type AnswerCache struct {
+	TTL time.Duration
+	// Any reuses answers to requests sampled at a temperature above 0
+	// with no seed, which would otherwise each be asked again.
+	Any bool
+}
+
+// SetAnswerCache turns reusing answers on for a project, or off with ttl 0.
+// The TTL is kept in whole seconds, at least one.
+func (s *DB) SetAnswerCache(ctx context.Context, projectID string, ttl time.Duration, anyAnswer bool) error {
+	var secs any
+	a := 0
+	if ttl > 0 {
+		secs = max(int64(ttl/time.Second), 1)
+		if anyAnswer {
+			a = 1
+		}
+	}
+	res, err := s.db.ExecContext(ctx,
+		`UPDATE projects SET cache_ttl_seconds = ?, cache_any = ? WHERE id = ?`, secs, a, projectID)
+	if err != nil {
+		return err
+	}
+	if n, _ := res.RowsAffected(); n == 0 {
+		return fmt.Errorf("store: no project %q", projectID)
+	}
+	return nil
+}
+
+// AnswerCacheOf is a project's setting; zero when it reuses none.
+func (s *DB) AnswerCacheOf(ctx context.Context, projectID string) (AnswerCache, error) {
+	var secs sql.NullInt64
+	var a int
+	err := s.db.QueryRowContext(ctx,
+		`SELECT cache_ttl_seconds, cache_any FROM projects WHERE id = ?`, projectID).Scan(&secs, &a)
+	if errors.Is(err, sql.ErrNoRows) || err == nil && (!secs.Valid || secs.Int64 <= 0) {
+		return AnswerCache{}, nil
+	}
+	if err != nil {
+		return AnswerCache{}, err
+	}
+	return AnswerCache{TTL: time.Duration(secs.Int64) * time.Second, Any: a == 1}, nil
+}

internal/store/answercache_test.go

@@ -0,0 +1,57 @@
+package store
+
+import (
+	"testing"
+	"time"
+)
+
+// answerCacheRoundTrip is the same on both backends: a project's setting
+// and a hit's mark on its usage record.
+func answerCacheRoundTrip(t *testing.T, s *DB) {
+	t.Helper()
+	ctx := t.Context()
+	proj := tree(t, s)
+	_, k := mustKey(t, s, "k1", proj)
+
+	if c, err := s.AnswerCacheOf(ctx, proj); err != nil || c.TTL != 0 || c.Any {
+		t.Fatalf("a new project reuses answers: %+v %v", c, err)
+	}
+	if err := s.SetAnswerCache(ctx, proj, 90*time.Minute, true); err != nil {
+		t.Fatal(err)
+	}
+	if c, _ := s.AnswerCacheOf(ctx, proj); c.TTL != 90*time.Minute || !c.Any {
+		t.Errorf("set: %+v", c)
+	}
+	if err := s.SetAnswerCache(ctx, proj, 0, true); err != nil {
+		t.Fatal(err)
+	}
+	if c, _ := s.AnswerCacheOf(ctx, proj); c.TTL != 0 || c.Any {
+		t.Errorf("off: %+v", c)
+	}
+	if err := s.SetAnswerCache(ctx, "no-such-project", time.Hour, false); err == nil {
+		t.Error("a project that does not exist was set")
+	}
+
+	now := time.Now()
+	if err := s.WriteUsage(ctx, []Usage{
+		{At: now, KeyID: k.ID, ProjectID: proj, Model: "m", ServedModel: "m", Status: 200, RequestID: "hit", CacheHit: true},
+		{At: now, KeyID: k.ID, ProjectID: proj, Model: "m", ServedModel: "m", Status: 200, RequestID: "miss",
+			PromptTokens: 3, CompletionTokens: 4, CostMicros: 10},
+	}); err != nil {
+		t.Fatal(err)
+	}
+	if u, ok, err := s.UsageByRequest(ctx, "hit"); err != nil || !ok || !u.CacheHit {
+		t.Errorf("a hit: %+v %v", u, err)
+	}
+	if u, ok, err := s.UsageByRequest(ctx, "miss"); err != nil || !ok || u.CacheHit {
+		t.Errorf("a miss: %+v %v", u, err)
+	}
+}
+
+func TestAProjectsReuseOfAnswersIsKeptAndAHitIsMarked(t *testing.T) {
+	answerCacheRoundTrip(t, capacityStore(t))
+}
+
+func TestPostgresKeepsAProjectsReuseOfAnswers(t *testing.T) {
+	answerCacheRoundTrip(t, openPG(t))
+}

internal/store/migrations/postgres/0027_answer_cache.sql

@@ -0,0 +1,11 @@
+-- Ported from the SQLite migration of the same number.
+-- Reusing answers, for a project that asks: how long one is reused, and
+-- whether any request's answer is, or only those that come out the same
+-- every time (temperature 0, or a seed). Off for every project until set.
+-- The answers themselves are held in each gateway's memory, never here.
+ALTER TABLE projects ADD COLUMN cache_ttl_seconds BIGINT;
+ALTER TABLE projects ADD COLUMN cache_any BIGINT NOT NULL DEFAULT 0;
+
+-- A request answered from that cache: nothing was generated for it, so it
+-- carries no tokens and costs nothing, and this says why.
+ALTER TABLE usage ADD COLUMN cache_hit BIGINT NOT NULL DEFAULT 0;

internal/store/migrations/sqlite/0027_answer_cache.sql

@@ -0,0 +1,10 @@
+-- Reusing answers, for a project that asks: how long one is reused, and
+-- whether any request's answer is, or only those that come out the same
+-- every time (temperature 0, or a seed). Off for every project until set.
+-- The answers themselves are held in each gateway's memory, never here.
+ALTER TABLE projects ADD COLUMN cache_ttl_seconds INTEGER;
+ALTER TABLE projects ADD COLUMN cache_any INTEGER NOT NULL DEFAULT 0;
+
+-- A request answered from that cache: nothing was generated for it, so it
+-- carries no tokens and costs nothing, and this says why.
+ALTER TABLE usage ADD COLUMN cache_hit INTEGER NOT NULL DEFAULT 0;

internal/store/sqlite.go

@@ -435,18 +435,22 @@ func (s *DB) WriteUsage(ctx context.Context, records []Usage) error {
 			`INSERT INTO usage (ts, key_id, project_id, model, served_model,
 			   prompt_tokens, completion_tokens, cost_micros, status, latency_ms, user_id,
 			   request_id, replica, queue_ms, ttft_ms,
-			   guard_action, guard_policy, guard_category, guard_ms, end_user, tags)
-			 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`)
+			   guard_action, guard_policy, guard_category, guard_ms, end_user, tags, cache_hit)
+			 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`)
 		if err != nil {
 			return err
 		}
 		defer stmt.Close()
 		for _, r := range records {
+			hit := 0
+			if r.CacheHit {
+				hit = 1
+			}
 			if _, err := stmt.ExecContext(ctx, millis(r.At), r.KeyID, r.ProjectID,
 				r.Model, r.ServedModel, r.PromptTokens, r.CompletionTokens,
 				r.CostMicros, r.Status, r.LatencyMs, r.UserID,
 				r.RequestID, r.Replica, r.QueueMs, r.TTFTMs,
-				r.GuardAction, r.GuardPolicy, r.GuardCategory, r.GuardMs, r.EndUser, tagsJSON(r.Tags)); err != nil {
+				r.GuardAction, r.GuardPolicy, r.GuardCategory, r.GuardMs, r.EndUser, tagsJSON(r.Tags), hit); err != nil {
 				return err
 			}
 		}
@@ -856,15 +860,16 @@ func (s *DB) UsageByRequest(ctx context.Context, requestID string) (Usage, bool,
 	var u Usage
 	var ts int64
 	var tags string
+	var hit int
 	err := s.db.QueryRowContext(ctx,
 		`SELECT ts, key_id, project_id, COALESCE(user_id, ''), model, served_model, prompt_tokens,
 		        completion_tokens, cost_micros, status, latency_ms, request_id, replica, queue_ms, ttft_ms,
-		        guard_action, guard_policy, guard_category, guard_ms, end_user, tags
+		        guard_action, guard_policy, guard_category, guard_ms, end_user, tags, cache_hit
 		 FROM usage WHERE request_id = ? ORDER BY ts DESC LIMIT 1`, requestID).
 		Scan(&ts, &u.KeyID, &u.ProjectID, &u.UserID, &u.Model, &u.ServedModel, &u.PromptTokens,
 			&u.CompletionTokens, &u.CostMicros, &u.Status, &u.LatencyMs, &u.RequestID, &u.Replica,
 			&u.QueueMs, &u.TTFTMs, &u.GuardAction, &u.GuardPolicy, &u.GuardCategory, &u.GuardMs,
-			&u.EndUser, &tags)
+			&u.EndUser, &tags, &hit)
 	if errors.Is(err, sql.ErrNoRows) {
 		return Usage{}, false, nil
 	}
@@ -873,5 +878,6 @@ func (s *DB) UsageByRequest(ctx context.Context, requestID string) (Usage, bool,
 	}
 	u.At = time.UnixMilli(ts)
 	u.Tags = parseTags(tags)
+	u.CacheHit = hit == 1
 	return u, true, nil
 }

internal/store/store.go

@@ -269,6 +269,10 @@ type Usage struct {
 	// and x-llmgw-tags. Both for attributing spend inside a project.
 	EndUser string
 	Tags    map[string]string
+
+	// CacheHit says the answer came from the project's answer cache:
+	// nothing was generated, so the record carries no tokens and no cost.
+	CacheHit bool
 }
 
 // tagsJSON is how tags are kept: a JSON object, keys sorted, or empty.

notes/c17.md

@@ -0,0 +1,55 @@
+## CHANGELOG
+
+### Reusing answers
+
+- **A response cache, per project, off until asked for.** `llmgw keys cache
+  --project P --ttl 1h` has the same request — same project, endpoint,
+  model by its catalog name, and body with `stream`, `stream_options`,
+  `user`, `metadata`, `store` and `service_tier` left out and its keys in
+  any order — answered from the gateway's memory instead of the model, for
+  the TTL. `--off` stops it; without either it says how it stands. The
+  console has it as **Cache answers…** on a project.
+- **Only answers that come out the same are reused:** chat and completions
+  asked at temperature 0 or with a seed, and embeddings, rerank and score.
+  `--any` reuses sampled answers too. Only a whole 200 from the model asked
+  for is kept — not an error, a fallback's or a canary's answer, one a
+  guardrail changed or stopped, or one over `router.cache.max_entry_bytes`.
+- **Streams both ways:** a streamed answer is kept put back together into
+  one completion (tool calls and reasoning included), and a kept answer is
+  streamed as chunks — with the usage chunk if the client asked — to a
+  client that streams. `/v1/messages` and `/v1/responses` get it too.
+- **Headers:** `x-llmgw-cache: hit|miss|bypass` on every answer of a
+  project that reuses answers, `Age` on a hit. `Cache-Control: no-cache`
+  or `no-store`, or `x-llmgw-cache: no`, asks the model anew.
+- **Policies still apply:** limits and the input guardrails run before the
+  cache is looked in, the output policy judges a reused answer as it would
+  a new one, and a request under `no_store` or continuing a stored
+  conversation is never served from it nor kept in it. A project's answers
+  are never looked up by another project's keys.
+- **A hit costs nothing:** recorded as a 200 with no tokens and no cost,
+  marked `cache_hit` on the usage record (migration 0027), and shown by
+  `llmgw ctl request`; it counts towards the requests-per-minute limit only.
+- **In memory only:** `router.cache.max_bytes` (128 MB) for all projects
+  together in each gateway process, least recently used dropped first;
+  nothing is written to the store. `llmgw_cache_requests_total{model,result}`,
+  `llmgw_cache_bytes` and `llmgw_cache_entries`.
+
+## BACKLOG
+
+**Done (branch `features-c2`):** exact-match answer cache per project
+(`keys cache --ttl/--any/--off`, console **Cache answers…**, `projects`
+columns, 30-second setting cache), reproducible answers only unless
+`--any`, LRU in memory bounded by `router.cache.max_bytes` and
+`max_entry_bytes`, streamed answers kept and replayed as chunks,
+`x-llmgw-cache` and `Cache-Control: no-cache`, guardrails before a hit and
+the output policy on it, `no_store` and threads bypass, a hit recorded as
+`cache_hit` with no tokens and no cost (migration 0027), metrics, docs.
+Left: semantic (embedding-similarity) caching through the embeddings
+models; a cache shared across control planes (each process keeps its own,
+deliberately: a shared one would write every answer to the store); a hit
+still waits its turn in admission, because the guardrails judge the request
+after admission and the key is the request they let through; logprobs in a
+streamed answer leave it unkept; `users forget` does not reach into the
+cache (answers age out by the TTL); no listing or flushing of what is
+cached (`--off` stops it being served at once, the memory is freed by TTL
+and LRU).