diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml
new file mode 100644
index 0000000..e172f86
--- /dev/null
+++ b/.github/workflows/tests.yml
@@ -0,0 +1,22 @@
+name: tests
+on:
+  push:
+    tags:
+      - v*
+    branches:
+      - master
+      - patchset
+  pull_request:
+jobs:
+  go-test-race:
+    name: go test -race
+    runs-on: blacksmith-8vcpu-ubuntu-2404
+    steps:
+      - uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4
+        with:
+          persist-credentials: false
+      - uses: actions/setup-go@40f1582b2485089dde7abd97c1529aa768e1baff # v5
+        with:
+          go-version-file: go.mod
+      - name: go test -race
+        run: go test -race ./...
diff --git a/BUILD.bazel b/BUILD.bazel
index 392fdd1..b0b6dfa 100644
--- a/BUILD.bazel
+++ b/BUILD.bazel
@@ -15,6 +15,7 @@ go_library(
     visibility = ["//visibility:private"],
     deps = [
         "//cache/disk:go_default_library",
+        "//cache/lruflush:go_default_library",
         "//config:go_default_library",
         "//ldap:go_default_library",
         "//server:go_default_library",
diff --git a/cache/disk/BUILD.bazel b/cache/disk/BUILD.bazel
index 2ea9a2b..b5c79d6 100644
--- a/cache/disk/BUILD.bazel
+++ b/cache/disk/BUILD.bazel
@@ -40,6 +40,8 @@ go_test(
         "findmissing_test.go",
         "lru_capture_bench_test.go",
         "lru_capture_test.go",
+        "lru_maxentries_test.go",
+        "lru_mem_test.go",
         "lru_test.go",
         "operation_observer_test.go",
     ],
diff --git a/cache/disk/load.go b/cache/disk/load.go
index 30c155b..6e1a61c 100644
--- a/cache/disk/load.go
+++ b/cache/disk/load.go
@@ -668,6 +668,14 @@ func (c *diskCache) loadExistingFiles(maxSizeBytes int64, cc CacheConfig) error
 			bytesToGigaBytes(c.lru.maxSizeHardLimit))
 	}
 
+	if cc.maxEntries > 0 {
+		// Set before the Add loop below, so a cache directory holding
+		// more files than the cap is trimmed (oldest first) during load,
+		// like a restart with a smaller maxSize.
+		c.lru.maxEntries = cc.maxEntries
+		log.Printf("Will evict at max entries: %d", cc.maxEntries)
+	}
+
 	// Start one single goroutine running in background, continuously
 	// waiting for files to be removed and removing them. Benchmarks on
 	// Linux with the XFS file system have surprisingly shown that removal
diff --git a/cache/disk/lru.go b/cache/disk/lru.go
index 2c0f6dd..792398c 100644
--- a/cache/disk/lru.go
+++ b/cache/disk/lru.go
@@ -37,6 +37,16 @@ type SizedLRU struct {
 	// cache below maxSize.
 	maxSize int64
 
+	// When positive, SizedLRU additionally evicts items as needed to keep
+	// the number of resident entries at or below maxEntries. This bounds
+	// the index metadata (key string, entry struct, list node, map slot -
+	// a fixed cost per entry regardless of blob size), which the byte
+	// budget alone does not: zero-byte blobs charge nothing against
+	// maxSize and tiny blobs charge at most one 4 KiB block, so without a
+	// count bound the metadata of a byte-full cache is effectively
+	// unbounded. Zero or negative means no entry-count bound.
+	maxEntries int64
+
 	// Channel containing evicted entries removed from ll, but not yet
 	// removed from the file system.
 	//
@@ -65,6 +75,7 @@ type SizedLRU struct {
 	gaugeCacheLogicalBytes   prometheus.Gauge
 	counterEvictedBytes      prometheus.Counter
 	counterOverwrittenBytes  prometheus.Counter
+	counterMaxEntriesEvicted prometheus.Counter
 
 	summaryCacheItemBytes prometheus.Summary
 
@@ -130,6 +141,10 @@ func NewSizedLRU(maxSize int64, onEvict EvictCallback, initialCapacity int) Size
 			Name: "bazel_remote_disk_cache_overwritten_bytes_total",
 			Help: "The total number of bytes removed from disk backend, due to put of already existing key",
 		}),
+		counterMaxEntriesEvicted: prometheus.NewCounter(prometheus.CounterOpts{
+			Name: "bazel_remote_disk_cache_max_entries_evictions_total",
+			Help: "The total number of entries evicted because the entry-count limit was reached, rather than the byte-size limit",
+		}),
 		summaryCacheItemBytes: prometheus.NewSummary(prometheus.SummaryOpts{
 			Name: "bazel_remote_disk_cache_entry_bytes",
 			Help: "Size of cache entries",
@@ -150,6 +165,7 @@ func (c *SizedLRU) RegisterMetrics() {
 	prometheus.MustRegister(c.gaugeCacheLogicalBytes)
 	prometheus.MustRegister(c.counterEvictedBytes)
 	prometheus.MustRegister(c.counterOverwrittenBytes)
+	prometheus.MustRegister(c.counterMaxEntriesEvicted)
 	prometheus.MustRegister(c.summaryCacheItemBytes)
 
 	// Set gauges to constant configured values to help visualize configured limits
@@ -222,6 +238,20 @@ func (c *SizedLRU) Add(key string, value lruItem) (ok bool) {
 		}
 	}
 
+	// Entry-count eviction (see maxEntries). Only inserts of new keys can
+	// grow the count, and the new entry is at the front of the list, so
+	// the evicted tail is always an older entry.
+	if c.maxEntries > 0 {
+		for int64(c.ll.Len()) > c.maxEntries {
+			ele := c.ll.Back()
+			if ele == nil {
+				break
+			}
+			c.removeElement(ele)
+			c.counterMaxEntriesEvicted.Inc()
+		}
+	}
+
 	c.currentSize += sizeDelta
 	c.uncompressedSize += uncompressedSizeDelta
 
diff --git a/cache/disk/lru_maxentries_test.go b/cache/disk/lru_maxentries_test.go
new file mode 100644
index 0000000..e33eb25
--- /dev/null
+++ b/cache/disk/lru_maxentries_test.go
@@ -0,0 +1,223 @@
+package disk
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"io"
+	"math"
+	"os"
+	"testing"
+
+	"github.com/buchgr/bazel-remote/v2/cache"
+	testutils "github.com/buchgr/bazel-remote/v2/utils"
+)
+
+// Tests for the entry-count bound (SizedLRU.maxEntries / WithMaxEntries),
+// which caps resident index metadata independently of the byte budget.
+
+func TestMaxEntriesEvictsFromTail(t *testing.T) {
+	var evicted []string
+	onEvict := func(key string, value lruItem) {
+		evicted = append(evicted, key)
+	}
+
+	lru := NewSizedLRU(math.MaxInt64, onEvict, 0)
+	lru.maxEntries = 2
+
+	for i := 0; i < 4; i++ {
+		if !lru.Add(fmt.Sprintf("key-%d", i), lruItem{size: 1, sizeOnDisk: 1}) {
+			t.Fatalf("Add of key-%d rejected", i)
+		}
+	}
+
+	if lru.Len() != 2 {
+		t.Fatalf("expected 2 entries, got %d", lru.Len())
+	}
+	for _, key := range []string{"key-2", "key-3"} {
+		if _, ok := lru.Peek(key); !ok {
+			t.Fatalf("expected %s to be resident", key)
+		}
+	}
+
+	// Evicted entries reach onEvict through the queued-eviction channel,
+	// which has no background consumer in LRU-only tests; drain it once
+	// (all queued entries merge into a single slice).
+	lru.performQueuedEvictions()
+	if len(evicted) != 2 || evicted[0] != "key-0" || evicted[1] != "key-1" {
+		t.Fatalf("expected [key-0 key-1] evicted in LRU order, got %v", evicted)
+	}
+
+	// The byte accounting must reflect the evictions: 2 resident single-
+	// block entries.
+	if lru.TotalSize() != 2*BlockSize {
+		t.Fatalf("expected currentSize %d, got %d", 2*BlockSize, lru.TotalSize())
+	}
+}
+
+func TestMaxEntriesBoundsZeroByteEntries(t *testing.T) {
+	// Zero-byte entries charge nothing against the byte budget, so only
+	// the entry-count bound limits them.
+	lru := NewSizedLRU(math.MaxInt64, func(string, lruItem) {}, 0)
+	lru.maxEntries = 10
+
+	for i := 0; i < 100; i++ {
+		if !lru.Add(fmt.Sprintf("zero-%d", i), lruItem{}) {
+			t.Fatalf("Add of zero-%d rejected", i)
+		}
+	}
+
+	if lru.Len() != 10 {
+		t.Fatalf("expected 10 entries, got %d", lru.Len())
+	}
+	if lru.TotalSize() != 0 {
+		t.Fatalf("expected zero currentSize, got %d", lru.TotalSize())
+	}
+}
+
+func TestMaxEntriesOverwriteDoesNotEvict(t *testing.T) {
+	var evicted []string
+	onEvict := func(key string, value lruItem) {
+		evicted = append(evicted, key)
+	}
+
+	lru := NewSizedLRU(math.MaxInt64, onEvict, 0)
+	lru.maxEntries = 2
+
+	if !lru.Add("a", lruItem{size: 1, sizeOnDisk: 1}) {
+		t.Fatal("Add of a rejected")
+	}
+	if !lru.Add("b", lruItem{size: 1, sizeOnDisk: 1}) {
+		t.Fatal("Add of b rejected")
+	}
+
+	// Overwriting a resident key at the cap must not change the count or
+	// evict the other entry. (The overwrite itself queues the replaced
+	// file for removal, which is not a count eviction but does reach
+	// onEvict.)
+	if !lru.Add("a", lruItem{size: 2, sizeOnDisk: 2}) {
+		t.Fatal("overwrite of a rejected")
+	}
+
+	if lru.Len() != 2 {
+		t.Fatalf("expected 2 entries, got %d", lru.Len())
+	}
+	if _, ok := lru.Peek("b"); !ok {
+		t.Fatal("expected b to remain resident after overwrite of a")
+	}
+
+	lru.performQueuedEvictions()
+	if len(evicted) != 1 || evicted[0] != "a" {
+		t.Fatalf("expected only the replaced version of a in the eviction queue, got %v", evicted)
+	}
+}
+
+func TestMaxEntriesDisabledByDefault(t *testing.T) {
+	lru := NewSizedLRU(math.MaxInt64, func(string, lruItem) {}, 0)
+
+	for i := 0; i < 1000; i++ {
+		if !lru.Add(fmt.Sprintf("zero-%d", i), lruItem{}) {
+			t.Fatalf("Add of zero-%d rejected", i)
+		}
+	}
+
+	if lru.Len() != 1000 {
+		t.Fatalf("expected 1000 entries with no count bound, got %d", lru.Len())
+	}
+}
+
+func TestDiskCacheWithMaxEntries(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	cacheDir := tempDir(t)
+	defer func() { _ = os.RemoveAll(cacheDir) }()
+
+	const maxEntries = 4
+	const numBlobs = 10
+	const itemSize = int64(64)
+
+	testCacheI, err := New(cacheDir, math.MaxInt64,
+		WithAccessLogger(testutils.NewSilentLogger()),
+		WithMaxEntries(maxEntries))
+	if err != nil {
+		t.Fatal(err)
+	}
+	testCache := testCacheI.(*diskCache)
+
+	var hashes []string
+	for i := 0; i < numBlobs; i++ {
+		data, hash := testutils.RandomDataAndHash(itemSize)
+		hashes = append(hashes, hash)
+		err = testCache.Put(ctx, cache.CAS, hash, itemSize,
+			io.NopCloser(bytes.NewReader(data)))
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	if testCache.lru.Len() != maxEntries {
+		t.Fatalf("expected %d resident entries, got %d",
+			maxEntries, testCache.lru.Len())
+	}
+
+	// The most recently put blobs are resident, the oldest are not.
+	for _, hash := range hashes[numBlobs-maxEntries:] {
+		if ok, _ := testCache.Contains(ctx, cache.CAS, hash, itemSize); !ok {
+			t.Fatalf("expected recent blob %s to be resident", hash)
+		}
+	}
+	for _, hash := range hashes[:numBlobs-maxEntries] {
+		if ok, _ := testCache.Contains(ctx, cache.CAS, hash, itemSize); ok {
+			t.Fatalf("expected old blob %s to have been evicted", hash)
+		}
+	}
+}
+
+func TestDiskCacheMaxEntriesTrimsOnLoad(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	cacheDir := tempDir(t)
+	defer func() { _ = os.RemoveAll(cacheDir) }()
+
+	const maxEntries = 3
+	const numBlobs = 8
+	const itemSize = int64(64)
+
+	// Fill a cache directory with no entry-count bound.
+	unboundedI, err := New(cacheDir, math.MaxInt64,
+		WithAccessLogger(testutils.NewSilentLogger()))
+	if err != nil {
+		t.Fatal(err)
+	}
+	unbounded := unboundedI.(*diskCache)
+	for i := 0; i < numBlobs; i++ {
+		data, hash := testutils.RandomDataAndHash(itemSize)
+		err = unbounded.Put(ctx, cache.CAS, hash, itemSize,
+			io.NopCloser(bytes.NewReader(data)))
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+	if unbounded.lru.Len() != numBlobs {
+		t.Fatalf("expected %d entries before reload, got %d",
+			numBlobs, unbounded.lru.Len())
+	}
+
+	// Reloading the same directory with a bound trims it during load,
+	// like a restart with a smaller maxSize. New only returns after the
+	// eviction backlog has been removed from disk.
+	boundedI, err := New(cacheDir, math.MaxInt64,
+		WithAccessLogger(testutils.NewSilentLogger()),
+		WithMaxEntries(maxEntries))
+	if err != nil {
+		t.Fatal(err)
+	}
+	bounded := boundedI.(*diskCache)
+
+	if bounded.lru.Len() != maxEntries {
+		t.Fatalf("expected %d entries after bounded reload, got %d",
+			maxEntries, bounded.lru.Len())
+	}
+}
diff --git a/cache/disk/lru_mem_test.go b/cache/disk/lru_mem_test.go
new file mode 100644
index 0000000..9fb0629
--- /dev/null
+++ b/cache/disk/lru_mem_test.go
@@ -0,0 +1,84 @@
+package disk
+
+// Measurement of resident LRU metadata cost per entry: the heap bytes
+// retained by one cache entry independent of blob size (key string, entry
+// struct, list.Element, map slot). This is the constant behind the
+// entry-count cap in the embedder's memory envelope; run with:
+//
+//	go test ./cache/disk/ -run TestLRUEntryMetadataCost -v
+//
+// It asserts only a loose sanity bound so drift in Go runtime internals
+// does not break the build, and logs the measured value for use in sizing.
+
+import (
+	"crypto/rand"
+	"encoding/hex"
+	"fmt"
+	"runtime"
+	"testing"
+)
+
+func heapInUse() uint64 {
+	runtime.GC()
+	runtime.GC()
+	var ms runtime.MemStats
+	runtime.ReadMemStats(&ms)
+	return ms.HeapAlloc
+}
+
+func TestLRUEntryMetadataCost(t *testing.T) {
+	const numEntries = 1_000_000
+
+	// Realistic lookup keys: "cas/<64 hex>" plus an lruItem with the
+	// random filename component populated, matching what the startup scan
+	// and Put paths store.
+	keys := make([]string, numEntries)
+	var buf [32]byte
+	for i := range keys {
+		if _, err := rand.Read(buf[:]); err != nil {
+			t.Fatal(err)
+		}
+		keys[i] = "cas/" + hex.EncodeToString(buf[:])
+	}
+
+	lru := NewSizedLRU(int64(numEntries)*BlockSize, nil, 0)
+
+	// Baseline after key generation so only LRU-internal allocations are
+	// attributed to the per-entry cost... except the keys themselves are
+	// retained by the LRU, so charge them too by measuring before keys
+	// are considered reachable only from the LRU. Simplest correct
+	// accounting: measure baseline before insertion, keep the keys slice
+	// alive, and add the average key size explicitly afterwards.
+	baseline := heapInUse()
+
+	for i, k := range keys {
+		if !lru.Add(k, lruItem{size: 1, sizeOnDisk: 1, random: fmt.Sprintf("%08d", i)}) {
+			t.Fatalf("Add rejected entry %d", i)
+		}
+	}
+
+	after := heapInUse()
+
+	// Keep both the LRU and the keys reachable across the measurement, or
+	// the GC inside heapInUse collects them first.
+	if lru.Len() != numEntries {
+		t.Fatalf("expected %d entries, got %d", numEntries, lru.Len())
+	}
+	runtime.KeepAlive(keys)
+
+	perEntry := float64(after-baseline) / numEntries
+
+	// The key bytes (~68 B) and its string header were allocated during
+	// key generation and are retained by the LRU afterwards; count them
+	// in the reported total.
+	const keyBytes = 4 + 64
+	total := perEntry + keyBytes
+
+	t.Logf("LRU metadata per entry: %.0f B internal + %d B key = %.0f B total (%d entries, heap %d -> %d)",
+		perEntry, keyBytes, total, numEntries, baseline, after)
+
+	// Sanity bounds only; the log line is the deliverable.
+	if total < 100 || total > 1000 {
+		t.Fatalf("implausible per-entry metadata cost: %.0f B", total)
+	}
+}
diff --git a/cache/disk/options.go b/cache/disk/options.go
index ff508ec..5162b8a 100644
--- a/cache/disk/options.go
+++ b/cache/disk/options.go
@@ -17,6 +17,7 @@ type CacheConfig struct {
 	diskCache        *diskCache        // Assumed to be non-nil.
 	metrics          *metricsDecorator // May be nil.
 	maxSizeHardLimit int64
+	maxEntries       int64
 }
 
 func WithStorageMode(mode string) Option {
@@ -42,6 +43,21 @@ func WithZstdImplementation(impl string) Option {
 	}
 }
 
+// WithMaxEntries bounds the number of entries resident in the cache index,
+// evicting least-recently-used entries when the bound is exceeded - the same
+// semantics as the byte budget, but counting entries. Each resident entry
+// costs a fixed ~270 bytes of process memory (key string, entry struct, list
+// node, map slot) regardless of blob size, while charging at most one 4 KiB
+// block against the byte budget (zero-byte blobs charge nothing), so without
+// a count bound the index metadata of a byte-full cache is effectively
+// unbounded. n <= 0 (the default) disables the bound.
+func WithMaxEntries(n int64) Option {
+	return func(c *CacheConfig) error {
+		c.maxEntries = n
+		return nil
+	}
+}
+
 func WithMaxBlobSize(size int64) Option {
 	return func(c *CacheConfig) error {
 		if size <= 0 {
diff --git a/cache/lruflush/BUILD.bazel b/cache/lruflush/BUILD.bazel
new file mode 100644
index 0000000..35d09ce
--- /dev/null
+++ b/cache/lruflush/BUILD.bazel
@@ -0,0 +1,24 @@
+load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["lruflush.go"],
+    importpath = "github.com/buchgr/bazel-remote/v2/cache/lruflush",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//cache:go_default_library",
+        "@com_github_google_uuid//:go_default_library",
+        "@com_github_prometheus_client_golang//prometheus:go_default_library",
+        "@com_github_prometheus_client_golang//prometheus/promauto:go_default_library",
+    ],
+)
+
+go_test(
+    name = "go_default_test",
+    srcs = ["lruflush_test.go"],
+    embed = [":go_default_library"],
+    deps = [
+        "//cache:go_default_library",
+        "@com_github_prometheus_client_golang//prometheus/testutil:go_default_library",
+    ],
+)
diff --git a/cache/lruflush/lruflush.go b/cache/lruflush/lruflush.go
new file mode 100644
index 0000000..080d55e
--- /dev/null
+++ b/cache/lruflush/lruflush.go
@@ -0,0 +1,483 @@
+// Package lruflush buffers AC-access observations (cache.LRUObserver) in
+// memory and periodically flushes them as JSONL artifacts to the S3 backend
+// serving the observed tenant, using the artifact schema and key layout
+// defined in the cache package (the same shape the web-side retention sweep
+// already consumes — no consumer changes).
+//
+// The design is deliberately minimal: one aggregation map, one flush
+// goroutine, direct PUTs. Everything here is advisory — the artifact stream
+// informs retention decisions days out, so the correct response to any
+// pressure (full buffer, slow backend, failed upload) is to drop the
+// observations, count the drop, and move on. The next access to the same
+// entries re-establishes the recency signal. There are no queues, workers,
+// or retries to tune, and by construction nothing here can stall or fail a
+// cache request. Loss bound: a crash or drop loses at most one flush
+// interval of advisory observations — bounded minutes against retention
+// horizons of days.
+package lruflush
+
+import (
+	"bytes"
+	"context"
+	"os"
+	"strconv"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/buchgr/bazel-remote/v2/cache"
+
+	"github.com/google/uuid"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+)
+
+const (
+	// defaultFlushInterval bounds observation staleness. On the L1 the
+	// ticker is the primary flush trigger (no teardown signal), so it also
+	// bounds crash loss.
+	defaultFlushInterval = 5 * time.Minute
+	// uploadTimeout caps a single artifact PUT. Flushing runs on its own
+	// goroutine, so this bounds how long one slow tenant backend can delay
+	// the other tenants' artifacts within a window, not a request.
+	uploadTimeout = 30 * time.Second
+	// defaultPassTimeout is the wall-clock bound on one WHOLE flush pass,
+	// including the shutdown drain. Without it, N prefixes against a
+	// stalled artifact backend cost N x uploadTimeout — and at shutdown no
+	// customer traffic remains to open the data-plane breaker, so a node
+	// roll could wait through timeout after timeout. LRU data is
+	// disposable; the pass is not allowed to hold a roll hostage.
+	defaultPassTimeout = 2 * time.Minute
+
+	// capObjects is the process-wide bound on buffered object REFERENCES
+	// (AC entries + CAS leaf references; a leaf under two ACs counts
+	// twice, because both buffered memory and serialized artifact size are
+	// proportional to references). Observations past the cap are dropped
+	// whole and counted; the periodic flush clears the buffers, so this is
+	// pure OOM defense, expected to fire ~never (per-tenant touch-sets are
+	// kilobytes). Because the flush loop detaches buffers before
+	// uploading, peak live memory is one detached window plus one filling
+	// window: ~2x this cap's worth of references, ~85 B each.
+	capObjects = 250_000
+	// capClosureLeaves drops a single closure whose leaf list alone would
+	// dominate an artifact: the consumer reads one closure per JSONL line
+	// with a 16 MiB line limit (~190k leaves at ~85 B/ref), so 50k keeps a
+	// ~3.5x margin. Dropped WHOLE, never truncated — truncation would break
+	// complete-or-drop (the sweep would keep the AC but evict untracked
+	// leaves, serving broken action results). Advisory recency loss only.
+	capClosureLeaves = 50_000
+)
+
+// Flush triggers (label values for bazel_remote_lru_artifact_flush_total).
+const (
+	triggerPeriodic = "periodic"
+	triggerShutdown = "shutdown"
+)
+
+// Flush results (label values for bazel_remote_lru_artifact_flush_total).
+const (
+	resultSuccess = "success"
+	resultFailure = "failure"
+)
+
+// Drop reasons (label values for
+// bazel_remote_lru_flush_observations_dropped_total).
+const (
+	dropReasonMissingPrefix   = "missing_prefix"
+	dropReasonBufferFull      = "buffer_full"
+	dropReasonClosureTooLarge = "closure_too_large"
+	dropReasonContention      = "contention"
+)
+
+// Loss reasons (label values for
+// bazel_remote_lru_flush_closures_lost_total): why buffered closures were
+// discarded at flush time, counted per closure so observation-loss ratios
+// stay meaningful.
+const (
+	lossReasonUploadFailed    = "upload_failed"
+	lossReasonSerializeFailed = "serialize_failed"
+	lossReasonPassDeadline    = "pass_deadline"
+	lossReasonBackendSuppress = "backend_suppressed"
+)
+
+var (
+	flushTotal = promauto.NewCounterVec(prometheus.CounterOpts{
+		Name: "bazel_remote_lru_artifact_flush_total",
+		Help: "LRU observation artifact flushes by trigger and result (one artifact per tenant prefix per window; failures drop the window's observations).",
+	}, []string{"trigger", "result"})
+	// Distinct from the capture stage's
+	// bazel_remote_lru_observations_dropped_total (cache/disk): that counts
+	// closures never emitted (incomplete capture); this counts emitted
+	// closures the flusher refused to buffer.
+	droppedTotal = promauto.NewCounterVec(prometheus.CounterOpts{
+		Name: "bazel_remote_lru_flush_observations_dropped_total",
+		Help: "Emitted AC-access observations dropped before buffering (advisory recency loss, never a cache error).",
+	}, []string{"reason"})
+	// closuresLost counts buffered AC closures discarded at flush time
+	// (per closure, not per artifact, so loss ratios are computable
+	// against flushEntries).
+	closuresLost = promauto.NewCounterVec(prometheus.CounterOpts{
+		Name: "bazel_remote_lru_flush_closures_lost_total",
+		Help: "Buffered AC closures discarded at flush time (upload failure, pass deadline, or same-backend suppression). Advisory recency loss, never a cache error.",
+	}, []string{"reason"})
+	bufferedObjects = promauto.NewGauge(prometheus.GaugeOpts{
+		Name: "bazel_remote_lru_flush_buffered_objects",
+		Help: "Object references (AC entries + CAS leaves) currently buffered awaiting the next flush.",
+	})
+	// detachedObjects makes memory during a flush pass visible:
+	// buffered_objects resets at detach, but the detached window stays
+	// resident until its artifacts are uploaded or abandoned.
+	detachedObjects = promauto.NewGauge(prometheus.GaugeOpts{
+		Name: "bazel_remote_lru_flush_detached_objects",
+		Help: "Object references detached from the buffers and held by the in-progress flush pass.",
+	})
+	passDuration = promauto.NewHistogram(prometheus.HistogramOpts{
+		Name:    "bazel_remote_lru_flush_pass_duration_seconds",
+		Help:    "Wall-clock duration of one flush pass (all prefixes, serial).",
+		Buckets: prometheus.ExponentialBuckets(0.01, 4, 10),
+	})
+	flushBytes = promauto.NewHistogram(prometheus.HistogramOpts{
+		Name:    "bazel_remote_lru_artifact_flush_bytes",
+		Help:    "Serialized size of successfully flushed LRU artifacts.",
+		Buckets: prometheus.ExponentialBuckets(1024, 4, 10),
+	})
+	flushEntries = promauto.NewHistogram(prometheus.HistogramOpts{
+		Name:    "bazel_remote_lru_artifact_flush_entries",
+		Help:    "AC closure count of successfully flushed LRU artifacts.",
+		Buckets: prometheus.ExponentialBuckets(1, 4, 10),
+	})
+)
+
+// Sink writes one serialized artifact under a fully-composed object key,
+// routing by the request-scoped backend selection on ctx (both s3proxy
+// proxies implement this). This is the single owner of the artifact-write
+// contract. Implementations must not let artifact traffic influence cache
+// request behavior (see s3Cache.PutArtifact for the breaker stance).
+type Sink interface {
+	PutArtifact(ctx context.Context, key string, body []byte) error
+}
+
+// closureBuf accumulates one AC entry within a window. Leaves are a
+// hash→sizeOnDisk map so a closure's shared leaves dedup within the entry
+// (cross-closure dedup is the sweep's job).
+type closureBuf struct {
+	acSizeOnDisk int64
+	lastAccessMs int64
+	leaves       map[string]int64
+}
+
+// prefixBuffer accumulates AC closures for one storage prefix (which already
+// encodes install/repo/generation) since that prefix's last flush.
+type prefixBuffer struct {
+	// selection is the tenant's backend routing pair, captured from the
+	// request context and refreshed on every observation: artifacts must
+	// land next to the objects they describe, or the sweep (which follows
+	// the namespace row's pin) never sees them.
+	selection     cache.S3BackendSelection
+	windowStartMs int64
+	// order preserves first-seen order of AC hashes so artifact lines stay
+	// in access order without a sort.
+	order   []string
+	entries map[string]*closureBuf
+}
+
+// Flusher implements cache.LRUObserver: it buffers AC-access observations per
+// storage prefix, dedups/unions them by AC hash, and serially flushes one
+// JSONL artifact per prefix through the Sink on a periodic ticker and at
+// shutdown.
+type Flusher struct {
+	sink        Sink
+	interval    time.Duration
+	passTimeout time.Duration
+	host        string
+	processID   string
+	uniqSeq     atomic.Uint64
+
+	mu      sync.Mutex
+	buffers map[string]*prefixBuffer
+	objects int // buffered object references across all buffers, vs capObjects
+
+	stop     chan struct{}
+	stopOnce sync.Once
+	loopDone chan struct{}
+}
+
+// Option adjusts test-relevant knobs; production uses the defaults.
+type Option func(*Flusher)
+
+// WithFlushInterval overrides the periodic flush interval.
+func WithFlushInterval(d time.Duration) Option {
+	return func(f *Flusher) { f.interval = d }
+}
+
+// WithPassTimeout overrides the whole-pass wall-clock bound.
+func WithPassTimeout(d time.Duration) Option {
+	return func(f *Flusher) { f.passTimeout = d }
+}
+
+// New returns a Flusher writing through sink. Call Start to begin the
+// periodic flush loop and Drain exactly once at shutdown.
+func New(sink Sink, options ...Option) *Flusher {
+	host, err := os.Hostname()
+	if err != nil {
+		host = ""
+	}
+	f := &Flusher{
+		sink:        sink,
+		interval:    defaultFlushInterval,
+		passTimeout: defaultPassTimeout,
+		host:        host,
+		processID:   uuid.NewString(),
+		buffers:     map[string]*prefixBuffer{},
+		stop:        make(chan struct{}),
+		loopDone:    make(chan struct{}),
+	}
+	for _, opt := range options {
+		opt(f)
+	}
+	return f
+}
+
+// RecordACAccess buffers one AC-access observation. It performs only bounded,
+// round-trip-free in-memory work; serialization and the upload happen on the
+// flush goroutine. Only trusted-mode requests carry a storage prefix, so a
+// missing prefix (HTTP side door, RAW entries) is dropped — those paths have
+// no tenant keyspace for an artifact to describe.
+func (f *Flusher) RecordACAccess(ctx context.Context, closure cache.ACClosure) {
+	if f == nil {
+		return
+	}
+	prefix, ok := cache.StoragePrefixFromContext(ctx)
+	if !ok {
+		droppedTotal.WithLabelValues(dropReasonMissingPrefix).Inc()
+		return
+	}
+	// Whole-closure guard: one closure past this would dominate an artifact
+	// and can exceed the consumer's per-line read limit. Dropped WHOLE,
+	// never truncated (see capClosureLeaves).
+	if len(closure.Leaves) > capClosureLeaves {
+		droppedTotal.WithLabelValues(dropReasonClosureTooLarge).Inc()
+		return
+	}
+	selection, _ := cache.S3BackendFromContext(ctx)
+
+	// Nonblocking admission: this runs synchronously on the cache hit path,
+	// and a concurrent observation may be holding the lock through a wide
+	// (up to capClosureLeaves) merge. Advisory bookkeeping must never queue
+	// a cache request behind it, so contention is a metered drop — the
+	// recency signal re-establishes itself on the next access.
+	if !f.mu.TryLock() {
+		droppedTotal.WithLabelValues(dropReasonContention).Inc()
+		return
+	}
+	defer f.mu.Unlock()
+	// Full-buffer policy is drop, not flush: an emergency flush under
+	// pressure is exactly when the backend is least likely to absorb it,
+	// and the recency signal re-establishes itself on the next access.
+	// The cap is checked against the worst case for this observation (one
+	// AC reference plus all its leaves) so the accounting never runs ahead
+	// of the cap mid-merge.
+	if f.objects+1+len(closure.Leaves) > capObjects {
+		droppedTotal.WithLabelValues(dropReasonBufferFull).Inc()
+		return
+	}
+	buf := f.buffers[prefix]
+	if buf == nil {
+		buf = &prefixBuffer{
+			windowStartMs: closure.TSMillis,
+			entries:       map[string]*closureBuf{},
+		}
+		f.buffers[prefix] = buf
+	}
+	// Refresh the routing pair on every observation so a re-pinned tenant's
+	// later observations steer the whole window to the current backend.
+	if selection.Endpoint != "" {
+		buf.selection = selection
+	}
+	entry := buf.entries[closure.AC.Hash]
+	if entry == nil {
+		entry = &closureBuf{leaves: map[string]int64{}}
+		buf.entries[closure.AC.Hash] = entry
+		buf.order = append(buf.order, closure.AC.Hash)
+		f.objects++ // the AC reference itself
+	}
+	entry.acSizeOnDisk = closure.AC.SizeOnDisk
+	if closure.TSMillis > entry.lastAccessMs {
+		entry.lastAccessMs = closure.TSMillis
+	}
+	if closure.TSMillis != 0 && (buf.windowStartMs == 0 || closure.TSMillis < buf.windowStartMs) {
+		buf.windowStartMs = closure.TSMillis
+	}
+	for _, leaf := range closure.Leaves {
+		if _, exists := entry.leaves[leaf.Hash]; !exists {
+			f.objects++ // newly inserted leaf reference
+		}
+		entry.leaves[leaf.Hash] = leaf.SizeOnDisk
+	}
+	bufferedObjects.Set(float64(f.objects))
+}
+
+// Start launches the flush loop. It must be called at most once.
+func (f *Flusher) Start() {
+	go func() {
+		defer close(f.loopDone)
+		ticker := time.NewTicker(f.interval)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-f.stop:
+				return
+			case <-ticker.C:
+				f.flush(triggerPeriodic)
+			}
+		}
+	}()
+}
+
+// Drain stops the flush loop and synchronously flushes remaining buffers.
+// Call once, after request serving has stopped (no new observations arrive
+// past that point).
+func (f *Flusher) Drain() {
+	if f == nil {
+		return
+	}
+	f.stopOnce.Do(func() { close(f.stop) })
+	<-f.loopDone
+	f.flush(triggerShutdown)
+}
+
+// flush detaches all buffers in O(1) under the lock — RecordACAccess sits on
+// the request path, so no serialization or upload work ever happens while
+// holding f.mu — then serially uploads one artifact per prefix. Serial and
+// direct on purpose: this is the degenerate, sufficient form of a bounded
+// upload pipeline at one small artifact per tenant per window.
+//
+// The pass is bounded two ways, because LRU data is disposable and a flush
+// must never hold anything hostage: a whole-pass wall-clock deadline
+// (remaining prefixes are abandoned, closures counted as lost), and
+// per-backend failure suppression (after one failed PUT to an
+// (endpoint, bucket), remaining prefixes routed to it are skipped this pass
+// rather than paying the same timeout again). Buffered memory stays bounded
+// because new observations accumulate against a fresh map under the same
+// global cap.
+func (f *Flusher) flush(trigger string) {
+	windowEndMs := nowMs()
+	f.mu.Lock()
+	detached := f.buffers
+	detachedCount := f.objects
+	f.buffers = map[string]*prefixBuffer{}
+	f.objects = 0
+	bufferedObjects.Set(0)
+	f.mu.Unlock()
+	if len(detached) == 0 {
+		return
+	}
+	detachedObjects.Set(float64(detachedCount))
+	defer detachedObjects.Set(0)
+	start := time.Now()
+	defer func() { passDuration.Observe(time.Since(start).Seconds()) }()
+
+	passCtx, cancel := context.WithTimeout(context.Background(), f.passTimeout)
+	defer cancel()
+	failedBackends := map[cache.S3BackendSelection]bool{}
+	for prefix, buf := range detached {
+		if len(buf.entries) == 0 {
+			continue
+		}
+		if passCtx.Err() != nil {
+			closuresLost.WithLabelValues(lossReasonPassDeadline).Add(float64(len(buf.entries)))
+			continue
+		}
+		if failedBackends[buf.selection] {
+			closuresLost.WithLabelValues(lossReasonBackendSuppress).Add(float64(len(buf.entries)))
+			continue
+		}
+		if !f.writeArtifact(passCtx, prefix, buf, windowEndMs, trigger) {
+			failedBackends[buf.selection] = true
+		}
+	}
+}
+
+// writeArtifact serializes one prefix's window and uploads it, once. A failed
+// upload drops the window's observations (advisory; counted per closure).
+// Returns false when the backend refused or timed out, so the pass can
+// suppress further artifacts to the same backend.
+func (f *Flusher) writeArtifact(passCtx context.Context, prefix string, buf *prefixBuffer, windowEndMs int64, trigger string) bool {
+	closures := make([]cache.ACClosure, 0, len(buf.entries))
+	for _, acHash := range buf.order {
+		entry := buf.entries[acHash]
+		if entry == nil {
+			continue
+		}
+		leaves := make([]cache.LRUObject, 0, len(entry.leaves))
+		for leafHash, leafSize := range entry.leaves {
+			leaves = append(leaves, cache.LRUObject{Hash: leafHash, SizeOnDisk: leafSize})
+		}
+		closures = append(closures, cache.ACClosure{
+			AC:       cache.LRUObject{Hash: acHash, SizeOnDisk: entry.acSizeOnDisk},
+			Leaves:   leaves,
+			TSMillis: entry.lastAccessMs,
+		})
+	}
+
+	// Generation and InstanceName stay empty on the L1: the header carries
+	// provenance only, and the object key (via the storage prefix) is what
+	// encodes the generation for the sweep. Reconstructing them here would
+	// bake the host's prefix layout into the fork for no consumer benefit.
+	header := cache.LRUArtifactHeader{
+		SchemaVersion: cache.LRUArtifactSchemaVersion,
+		Host:          f.host,
+		ProcessID:     f.processID,
+		WindowStartMs: buf.windowStartMs,
+		WindowEndMs:   windowEndMs,
+		EntryCount:    len(closures),
+	}
+
+	var body bytes.Buffer
+	if err := cache.WriteLRUArtifact(&body, header, closures); err != nil {
+		flushTotal.WithLabelValues(trigger, resultFailure).Inc()
+		closuresLost.WithLabelValues(lossReasonSerializeFailed).Add(float64(len(closures)))
+		// A serialization failure says nothing about the backend.
+		return true
+	}
+	key := cache.LRUArtifactKey(normalizePrefix(prefix), windowEndMs, f.nextUniq())
+	ctx := passCtx
+	if buf.selection.Endpoint != "" {
+		ctx = cache.WithS3Backend(ctx, buf.selection)
+	}
+	// Per-PUT timeout nested inside the pass deadline: whichever is sooner
+	// wins.
+	putCtx, cancel := context.WithTimeout(ctx, uploadTimeout)
+	defer cancel()
+	if err := f.sink.PutArtifact(putCtx, key, body.Bytes()); err != nil {
+		flushTotal.WithLabelValues(trigger, resultFailure).Inc()
+		closuresLost.WithLabelValues(lossReasonUploadFailed).Add(float64(len(closures)))
+		return false
+	}
+	flushTotal.WithLabelValues(trigger, resultSuccess).Inc()
+	flushBytes.Observe(float64(body.Len()))
+	flushEntries.Observe(float64(len(closures)))
+	return true
+}
+
+// nextUniq returns a process-scoped discriminator so two flushes of the same
+// window for the same prefix never collide on the object key.
+func (f *Flusher) nextUniq() string {
+	short := strings.ReplaceAll(f.processID, "-", "")
+	if len(short) > 8 {
+		short = short[:8]
+	}
+	return short + "-" + strconv.FormatUint(f.uniqSeq.Add(1), 10)
+}
+
+// normalizePrefix ensures exactly one trailing slash: LRUArtifactKey
+// concatenates the prefix verbatim, and the trust boundary accepts a prefix
+// with or without one.
+func normalizePrefix(prefix string) string {
+	return strings.TrimSuffix(prefix, "/") + "/"
+}
+
+func nowMs() int64 {
+	return time.Now().UTC().UnixMilli()
+}
diff --git a/cache/lruflush/lruflush_test.go b/cache/lruflush/lruflush_test.go
new file mode 100644
index 0000000..64851c9
--- /dev/null
+++ b/cache/lruflush/lruflush_test.go
@@ -0,0 +1,513 @@
+package lruflush
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/buchgr/bazel-remote/v2/cache"
+
+	"github.com/prometheus/client_golang/prometheus/testutil"
+)
+
+type recordedPut struct {
+	key       string
+	body      []byte
+	selection cache.S3BackendSelection
+	hasSel    bool
+}
+
+type fakeSink struct {
+	mu       sync.Mutex
+	puts     []recordedPut
+	attempts int
+	failures int // fail the first N puts
+}
+
+func (s *fakeSink) PutArtifact(ctx context.Context, key string, body []byte) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.attempts++
+	if s.failures > 0 {
+		s.failures--
+		return errors.New("injected put failure")
+	}
+	sel, ok := cache.S3BackendFromContext(ctx)
+	s.puts = append(s.puts, recordedPut{
+		key:       key,
+		body:      append([]byte(nil), body...),
+		selection: sel,
+		hasSel:    ok,
+	})
+	return nil
+}
+
+func (s *fakeSink) recorded() []recordedPut {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return append([]recordedPut(nil), s.puts...)
+}
+
+func (s *fakeSink) attemptCount() int {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.attempts
+}
+
+func obsCtx(prefix string, sel *cache.S3BackendSelection) context.Context {
+	ctx := context.Background()
+	if prefix != "" {
+		ctx = cache.WithStoragePrefix(ctx, prefix)
+	}
+	if sel != nil {
+		ctx = cache.WithS3Backend(ctx, *sel)
+	}
+	return ctx
+}
+
+func closure(acHash string, ts int64, leaves ...string) cache.ACClosure {
+	c := cache.ACClosure{
+		AC:       cache.LRUObject{Hash: acHash, SizeOnDisk: 100},
+		TSMillis: ts,
+	}
+	for _, leaf := range leaves {
+		c.Leaves = append(c.Leaves, cache.LRUObject{Hash: leaf, SizeOnDisk: 10})
+	}
+	return c
+}
+
+// wideClosure returns a closure contributing exactly `objects` references
+// (one AC + objects-1 distinct leaves).
+func wideClosure(acHash string, ts int64, objects int) cache.ACClosure {
+	c := cache.ACClosure{AC: cache.LRUObject{Hash: acHash, SizeOnDisk: 1}, TSMillis: ts}
+	for j := 0; j < objects-1; j++ {
+		c.Leaves = append(c.Leaves, cache.LRUObject{Hash: fmt.Sprintf("%s-leaf-%d", acHash, j), SizeOnDisk: 1})
+	}
+	return c
+}
+
+// newTestFlusher returns a Flusher whose periodic loop is effectively
+// disabled so tests drive flushes deterministically via flush or Drain.
+func newTestFlusher(sink Sink) *Flusher {
+	f := New(sink, WithFlushInterval(time.Hour))
+	f.Start()
+	return f
+}
+
+func TestDrainFlushesBufferedClosuresInAccessOrder(t *testing.T) {
+	sink := &fakeSink{}
+	f := newTestFlusher(sink)
+	sel := cache.S3BackendSelection{Endpoint: "http://minio-a:9000", Bucket: "tenant-a"}
+	ctx := obsCtx("bazelre/prod/42/987/v7/", &sel)
+
+	f.RecordACAccess(ctx, closure("ac-b", 2000, "leaf-1", "leaf-2"))
+	f.RecordACAccess(ctx, closure("ac-a", 1000, "leaf-1"))
+	// Re-access ac-b: dedups into the same entry, bumps its timestamp,
+	// unions the new leaf.
+	f.RecordACAccess(ctx, closure("ac-b", 3000, "leaf-3"))
+	f.Drain()
+
+	puts := sink.recorded()
+	if len(puts) != 1 {
+		t.Fatalf("expected 1 artifact, got %d", len(puts))
+	}
+	put := puts[0]
+	if !strings.HasPrefix(put.key, "bazelre/prod/42/987/v7/lru/") {
+		t.Fatalf("artifact key %q not under <prefix>lru/", put.key)
+	}
+	if !put.hasSel || put.selection != sel {
+		t.Fatalf("artifact put did not carry the captured backend selection: %+v", put.selection)
+	}
+
+	header, closures, err := cache.ReadLRUArtifact(bytes.NewReader(put.body))
+	if err != nil {
+		t.Fatalf("artifact does not round-trip: %v", err)
+	}
+	if header.SchemaVersion != cache.LRUArtifactSchemaVersion {
+		t.Fatalf("schema version = %d", header.SchemaVersion)
+	}
+	if header.EntryCount != 2 || len(closures) != 2 {
+		t.Fatalf("expected 2 closures, header=%d got=%d", header.EntryCount, len(closures))
+	}
+	if header.WindowStartMs != 1000 {
+		t.Fatalf("window start = %d, want earliest observation 1000", header.WindowStartMs)
+	}
+	// First-seen (access) order: ac-b was observed first.
+	if closures[0].AC.Hash != "ac-b" || closures[1].AC.Hash != "ac-a" {
+		t.Fatalf("closures out of access order: %s, %s", closures[0].AC.Hash, closures[1].AC.Hash)
+	}
+	if closures[0].TSMillis != 3000 {
+		t.Fatalf("ac-b last-access = %d, want 3000", closures[0].TSMillis)
+	}
+	if len(closures[0].Leaves) != 3 {
+		t.Fatalf("ac-b leaves = %d, want union of 3", len(closures[0].Leaves))
+	}
+}
+
+func TestObservationWithoutPrefixIsDropped(t *testing.T) {
+	sink := &fakeSink{}
+	f := newTestFlusher(sink)
+	f.RecordACAccess(obsCtx("", nil), closure("ac-a", 1000, "leaf-1"))
+	f.Drain()
+	if got := len(sink.recorded()); got != 0 {
+		t.Fatalf("expected no artifacts for prefix-less observation, got %d", got)
+	}
+}
+
+func TestOversizedClosureIsDroppedWhole(t *testing.T) {
+	sink := &fakeSink{}
+	f := newTestFlusher(sink)
+	ctx := obsCtx("tenant/v1/", nil)
+
+	f.RecordACAccess(ctx, wideClosure("ac-big", 1, capClosureLeaves+2))
+	f.RecordACAccess(ctx, closure("ac-ok", 2, "leaf-a"))
+	f.Drain()
+
+	puts := sink.recorded()
+	if len(puts) != 1 {
+		t.Fatalf("expected 1 artifact, got %d", len(puts))
+	}
+	_, closures, err := cache.ReadLRUArtifact(bytes.NewReader(puts[0].body))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(closures) != 1 || closures[0].AC.Hash != "ac-ok" {
+		t.Fatalf("oversized closure leaked into artifact: %+v", closures)
+	}
+}
+
+// TestBufferFullDropsInsteadOfFlushing pins the full-buffer policy: past
+// capObjects new observations are dropped (advisory loss, self-correcting on
+// the next access), never queued, never flushed early.
+func TestBufferFullDropsInsteadOfFlushing(t *testing.T) {
+	sink := &fakeSink{}
+	f := newTestFlusher(sink)
+	ctx := obsCtx("tenant/v1/", nil)
+
+	// Fill to exactly capObjects, then one more observation must drop.
+	per := 1000
+	n := capObjects / per
+	for i := 0; i < n; i++ {
+		f.RecordACAccess(ctx, wideClosure(fmt.Sprintf("ac-%d", i), int64(i+1), per))
+	}
+	f.RecordACAccess(ctx, closure("ac-overflow", int64(n+1), "leaf-x"))
+	// No flush may have fired: drop-not-flush under pressure.
+	if got := len(sink.recorded()); got != 0 {
+		t.Fatalf("full buffer triggered %d flushes, want 0", got)
+	}
+	f.Drain()
+
+	puts := sink.recorded()
+	if len(puts) != 1 {
+		t.Fatalf("expected 1 artifact, got %d", len(puts))
+	}
+	_, closures, err := cache.ReadLRUArtifact(bytes.NewReader(puts[0].body))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(closures) != n {
+		t.Fatalf("closures = %d, want %d accepted before the cap", len(closures), n)
+	}
+	for _, cl := range closures {
+		if cl.AC.Hash == "ac-overflow" {
+			t.Fatal("over-cap observation leaked into the artifact")
+		}
+	}
+}
+
+// TestFlushClearsCapForNewObservations pins that a flush resets the global
+// charge: after a window is detached and uploaded, the buffer accepts new
+// observations again.
+func TestFlushClearsCapForNewObservations(t *testing.T) {
+	sink := &fakeSink{}
+	f := newTestFlusher(sink)
+	ctx := obsCtx("tenant/v1/", nil)
+
+	per := 1000
+	for i := 0; i < capObjects/per; i++ {
+		f.RecordACAccess(ctx, wideClosure(fmt.Sprintf("ac-%d", i), int64(i+1), per))
+	}
+	f.flush(triggerPeriodic)
+	f.RecordACAccess(ctx, closure("ac-after", 1, "leaf-a"))
+	f.Drain()
+
+	puts := sink.recorded()
+	if len(puts) != 2 {
+		t.Fatalf("expected 2 artifacts (window + post-flush), got %d", len(puts))
+	}
+	_, closures, err := cache.ReadLRUArtifact(bytes.NewReader(puts[1].body))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(closures) != 1 || closures[0].AC.Hash != "ac-after" {
+		t.Fatalf("post-flush observation missing: %+v", closures)
+	}
+}
+
+// TestFailedUploadDropsWindow pins retry-free failure semantics: a failed PUT
+// drops that window's observations and the flusher moves on.
+func TestFailedUploadDropsWindow(t *testing.T) {
+	sink := &fakeSink{failures: 1}
+	f := newTestFlusher(sink)
+	ctx := obsCtx("tenant/v1/", nil)
+
+	f.RecordACAccess(ctx, closure("ac-a", 1, "leaf-a"))
+	f.flush(triggerPeriodic)
+	if got := len(sink.recorded()); got != 0 {
+		t.Fatalf("failed upload produced %d artifacts, want 0", got)
+	}
+	// The next window is unaffected.
+	f.RecordACAccess(ctx, closure("ac-b", 2, "leaf-b"))
+	f.Drain()
+	puts := sink.recorded()
+	if len(puts) != 1 {
+		t.Fatalf("expected 1 artifact after recovery, got %d", len(puts))
+	}
+	_, closures, err := cache.ReadLRUArtifact(bytes.NewReader(puts[0].body))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(closures) != 1 || closures[0].AC.Hash != "ac-b" {
+		t.Fatalf("recovered window content wrong: %+v", closures)
+	}
+}
+
+// TestContentionDropsInsteadOfBlocking pins the nonblocking-admission
+// contract: RecordACAccess runs synchronously on the cache hit path, so when
+// the aggregation lock is held (a concurrent wide merge), the observation is
+// dropped immediately rather than queueing the cache request behind advisory
+// bookkeeping.
+func TestContentionDropsInsteadOfBlocking(t *testing.T) {
+	sink := &fakeSink{}
+	f := newTestFlusher(sink)
+	ctx := obsCtx("tenant/v1/", nil)
+
+	f.mu.Lock()
+	done := make(chan struct{})
+	go func() {
+		f.RecordACAccess(ctx, closure("ac-a", 1, "leaf-a"))
+		close(done)
+	}()
+	select {
+	case <-done:
+	case <-time.After(5 * time.Second):
+		f.mu.Unlock()
+		t.Fatal("RecordACAccess blocked on a held aggregation lock")
+	}
+	f.mu.Unlock()
+	f.Drain()
+	if got := len(sink.recorded()); got != 0 {
+		t.Fatalf("contended observation was buffered anyway: %d artifacts", got)
+	}
+}
+
+// ctxSink parks every PutArtifact until its context expires, simulating a
+// backend that never answers within any deadline.
+type ctxSink struct {
+	fakeSink
+}
+
+func (s *ctxSink) PutArtifact(ctx context.Context, key string, body []byte) error {
+	s.mu.Lock()
+	s.attempts++
+	s.mu.Unlock()
+	<-ctx.Done()
+	return ctx.Err()
+}
+
+// TestPassDeadlineAbandonsRemainingPrefixes pins the whole-pass wall-clock
+// bound: a stalled backend costs one put deadline, not N of them — remaining
+// prefixes are abandoned and their closures counted as lost.
+func TestPassDeadlineAbandonsRemainingPrefixes(t *testing.T) {
+	sink := &ctxSink{}
+	f := New(sink, WithFlushInterval(time.Hour), WithPassTimeout(50*time.Millisecond))
+	f.Start()
+
+	// Distinct selections per prefix so backend suppression cannot mask the
+	// deadline path.
+	for i := 0; i < 4; i++ {
+		sel := cache.S3BackendSelection{Endpoint: fmt.Sprintf("http://minio-%d:9000", i), Bucket: "b"}
+		f.RecordACAccess(obsCtx(fmt.Sprintf("tenant-%d/v1/", i), &sel), closure(fmt.Sprintf("ac-%d", i), int64(i+1), "leaf"))
+	}
+	start := time.Now()
+	f.flush(triggerPeriodic)
+	elapsed := time.Since(start)
+	if elapsed > 5*time.Second {
+		t.Fatalf("flush pass ran %v, want ~the 50ms pass deadline", elapsed)
+	}
+	if got := sink.attemptCount(); got != 1 {
+		t.Fatalf("stalled backend received %d put attempts in one pass, want 1", got)
+	}
+	f.Drain()
+}
+
+// TestBackendSuppressionSkipsSameBackendPrefixes pins per-backend failure
+// suppression: after one failed PUT to an (endpoint, bucket), remaining
+// prefixes routed to it are skipped this pass, while other backends still
+// get their artifacts.
+func TestBackendSuppressionSkipsSameBackendPrefixes(t *testing.T) {
+	sink := &fakeSink{failures: 1 << 30}
+	f := newTestFlusher(sink)
+	selSick := cache.S3BackendSelection{Endpoint: "http://minio-sick:9000", Bucket: "b"}
+	for i := 0; i < 3; i++ {
+		f.RecordACAccess(obsCtx(fmt.Sprintf("sick-%d/v1/", i), &selSick), closure(fmt.Sprintf("ac-%d", i), int64(i+1), "leaf"))
+	}
+	selOther := cache.S3BackendSelection{Endpoint: "http://minio-ok:9000", Bucket: "b"}
+	f.RecordACAccess(obsCtx("other/v1/", &selOther), closure("ac-other", 9, "leaf"))
+
+	f.flush(triggerPeriodic)
+	// One attempt for the sick backend (then suppressed), one for the other.
+	if got := sink.attemptCount(); got != 2 {
+		t.Fatalf("put attempts = %d, want 2 (1 sick + 1 other)", got)
+	}
+	f.Drain()
+}
+
+// blockingSink parks every PutArtifact until gate closes, simulating a
+// wedged MinIO.
+type blockingSink struct {
+	fakeSink
+	gate    chan struct{}
+	entered chan struct{}
+}
+
+func (s *blockingSink) PutArtifact(ctx context.Context, key string, body []byte) error {
+	s.entered <- struct{}{}
+	<-s.gate
+	return s.fakeSink.PutArtifact(ctx, key, body)
+}
+
+// TestBlockedSinkBoundsMemoryAndNeverBlocksObservations pins the review's
+// memory-bound concern in the serial design: with the flush goroutine wedged
+// on a blocked backend, observations keep being accepted up to capObjects,
+// are dropped past it, and RecordACAccess never blocks on the sink.
+func TestBlockedSinkBoundsMemoryAndNeverBlocksObservations(t *testing.T) {
+	sink := &blockingSink{
+		gate:    make(chan struct{}),
+		entered: make(chan struct{}, 64),
+	}
+	f := New(sink, WithFlushInterval(5*time.Millisecond))
+	f.Start()
+	ctx := obsCtx("tenant/v1/", nil)
+
+	f.RecordACAccess(ctx, closure("ac-first", 1, "leaf-a"))
+	// Wait for the flush loop to wedge inside the sink.
+	select {
+	case <-sink.entered:
+	case <-time.After(5 * time.Second):
+		t.Fatal("flush loop never reached the sink")
+	}
+
+	// Pour in far more than the cap while the flush goroutine is stuck.
+	per := 1000
+	for i := 0; i < capObjects/per+10; i++ {
+		f.RecordACAccess(ctx, wideClosure(fmt.Sprintf("ac-%d", i), int64(i+2), per))
+	}
+	f.mu.Lock()
+	buffered := f.objects
+	f.mu.Unlock()
+	if buffered > capObjects {
+		t.Fatalf("buffered objects %d exceed capObjects %d with a blocked sink", buffered, capObjects)
+	}
+
+	close(sink.gate)
+	f.Drain()
+}
+
+func TestPrefixWithoutTrailingSlashNormalizes(t *testing.T) {
+	sink := &fakeSink{}
+	f := newTestFlusher(sink)
+	f.RecordACAccess(obsCtx("tenant/v1", nil), closure("ac-a", 1, "leaf-a"))
+	f.Drain()
+	puts := sink.recorded()
+	if len(puts) != 1 {
+		t.Fatalf("expected 1 artifact, got %d", len(puts))
+	}
+	if !strings.HasPrefix(puts[0].key, "tenant/v1/lru/") {
+		t.Fatalf("key %q not normalized to <prefix>/lru/", puts[0].key)
+	}
+}
+
+func TestPeriodicTickerFlushes(t *testing.T) {
+	sink := &fakeSink{}
+	f := New(sink, WithFlushInterval(10*time.Millisecond))
+	f.Start()
+	f.RecordACAccess(obsCtx("tenant/v1/", nil), closure("ac-a", 1, "leaf-a"))
+
+	deadline := time.After(5 * time.Second)
+	for len(sink.recorded()) == 0 {
+		select {
+		case <-deadline:
+			t.Fatal("periodic flush never fired")
+		case <-time.After(5 * time.Millisecond):
+		}
+	}
+	f.Drain()
+	if got := len(sink.recorded()); got != 1 {
+		t.Fatalf("expected exactly 1 artifact (drain had nothing left), got %d", got)
+	}
+}
+
+// TestConcurrentObservationsAreSafe proves race-freedom under concurrent
+// observers and conservation under the nonblocking-admission policy: every
+// observation is either buffered (and lands in an artifact) or counted as a
+// contention drop — none silently vanish.
+func TestConcurrentObservationsAreSafe(t *testing.T) {
+	sink := &fakeSink{}
+	f := newTestFlusher(sink)
+	droppedBefore := testutil.ToFloat64(droppedTotal.WithLabelValues(dropReasonContention))
+	var wg sync.WaitGroup
+	for g := 0; g < 8; g++ {
+		wg.Add(1)
+		go func(g int) {
+			defer wg.Done()
+			ctx := obsCtx(fmt.Sprintf("tenant-%d/v1/", g%2), nil)
+			for i := 0; i < 500; i++ {
+				f.RecordACAccess(ctx, closure(fmt.Sprintf("ac-%d-%d", g, i), int64(i+1), "leaf-a", "leaf-b"))
+			}
+		}(g)
+	}
+	wg.Wait()
+	f.Drain()
+
+	flushed := 0
+	for _, p := range sink.recorded() {
+		_, closures, err := cache.ReadLRUArtifact(bytes.NewReader(p.body))
+		if err != nil {
+			t.Fatal(err)
+		}
+		flushed += len(closures)
+	}
+	dropped := int(testutil.ToFloat64(droppedTotal.WithLabelValues(dropReasonContention)) - droppedBefore)
+	if flushed+dropped != 8*500 {
+		t.Fatalf("flushed %d + contention-dropped %d != %d observations", flushed, dropped, 8*500)
+	}
+	if flushed == 0 {
+		t.Fatal("every observation was dropped; admission is broken, not contended")
+	}
+}
+
+func TestArtifactKeysNeverCollide(t *testing.T) {
+	sink := &fakeSink{}
+	f := newTestFlusher(sink)
+	ctx := obsCtx("tenant/v1/", nil)
+	// Two flushes of the same prefix in the same millisecond window must
+	// yield distinct keys via the uniq discriminator.
+	f.RecordACAccess(ctx, closure("ac-a", 1, "leaf-a"))
+	f.flush(triggerPeriodic)
+	f.RecordACAccess(ctx, closure("ac-b", 2, "leaf-b"))
+	f.Drain()
+
+	puts := sink.recorded()
+	if len(puts) != 2 {
+		t.Fatalf("expected 2 artifacts, got %d", len(puts))
+	}
+	if puts[0].key == puts[1].key {
+		t.Fatalf("artifact keys collided: %q", puts[0].key)
+	}
+}
diff --git a/cache/s3proxy/BUILD.bazel b/cache/s3proxy/BUILD.bazel
index c900111..2b148cb 100644
--- a/cache/s3proxy/BUILD.bazel
+++ b/cache/s3proxy/BUILD.bazel
@@ -3,6 +3,7 @@ load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
 go_library(
     name = "go_default_library",
     srcs = [
+        "artifacts.go",
         "auth_methods.go",
         "breaker.go",
         "multi.go",
@@ -24,6 +25,7 @@ go_library(
 go_test(
     name = "go_default_test",
     srcs = [
+        "artifacts_test.go",
         "s3proxy_breaker_test.go",
         "s3proxy_lru_test.go",
         "s3proxy_test.go",
@@ -31,6 +33,7 @@ go_test(
     embed = [":go_default_library"],
     deps = [
         "//cache:go_default_library",
+        "//cache/lruflush:go_default_library",
         "//utils/backendproxy:go_default_library",
         "@com_github_johannesboyne_gofakes3//:go_default_library",
         "@com_github_johannesboyne_gofakes3//backend/s3mem:go_default_library",
diff --git a/cache/s3proxy/artifacts.go b/cache/s3proxy/artifacts.go
new file mode 100644
index 0000000..c2ccd52
--- /dev/null
+++ b/cache/s3proxy/artifacts.go
@@ -0,0 +1,67 @@
+package s3proxy
+
+import (
+	"bytes"
+	"context"
+
+	"github.com/minio/minio-go/v7"
+)
+
+// LRU observation artifacts must land in the same (endpoint, bucket, prefix)
+// keyspace as the cache entries they describe: the web-side retention sweep
+// follows the namespace row's shard pin and reads `<storagePrefix>lru/`, so
+// an artifact written anywhere else is invisible to it. The cache.Proxy
+// interface has no vocabulary for non-cache-entry objects, hence this narrow
+// side surface implemented by both the single- and multi-backend proxies.
+// The interface it satisfies is owned by the sole consumer, lruflush.Sink:
+// routing (which backend, which bucket) comes from the request-scoped
+// selection on ctx, exactly like cache operations; the caller owns key
+// composition and retries.
+
+// artifactObjectTagKey/Value tag every artifact so a single bucket-wide MinIO
+// lifecycle (ILM) expiration rule can target LRU artifacts by tag. The `lru/`
+// segment is nested inside the per-tenant storage prefix, so a key-prefix
+// lifecycle filter cannot match it; a tag filter can. The TTL is a backstop
+// to the sweep's own artifact GC. The values are a wire contract with the
+// deployed ILM rules — do not change them.
+const (
+	artifactObjectTagKey   = "lru"
+	artifactObjectTagValue = "true"
+)
+
+// PutArtifact uploads one artifact through this backend's minio core. The
+// per-backend breaker is consulted READ-ONLY for the sick-shard fail-fast
+// (when the shard is struggling, failing the flush fast beats stacking
+// artifact PUTs on top of it), but the call runs outside the breaker and
+// never records an outcome. Advisory traffic is contractually barred from
+// influencing cache behavior in either direction: an artifact-only failure
+// (say, a bucket tagging-permission gap) must not open the breaker against
+// customer reads, and a small artifact success must not reset the failure
+// streak of a real read brownout or consume the half-open probe slot that a
+// cache read needs to close the breaker.
+func (c *s3Cache) PutArtifact(ctx context.Context, key string, body []byte) error {
+	bucket := c.bucketForContext(ctx)
+	if !c.breaker.isClosed() {
+		logResponse(c.accessLogger, "LRU_ARTIFACT", bucket, key, errBreakerOpen)
+		return errBreakerOpen
+	}
+	opts := minio.PutObjectOptions{
+		ContentType: "application/x-ndjson",
+		UserTags:    map[string]string{artifactObjectTagKey: artifactObjectTagValue},
+	}
+	_, err := c.mcore.PutObject(ctx, bucket, key,
+		bytes.NewReader(body), int64(len(body)), "", "", opts)
+	logResponse(c.accessLogger, "LRU_ARTIFACT", bucket, key, err)
+	return err
+}
+
+// PutArtifact routes to the backend selected on ctx, mirroring the cache
+// operations' dispatch: missing selector uses the default backend (metered by
+// backendFor), unknown selector refuses rather than guessing a shard.
+func (m *multiS3Cache) PutArtifact(ctx context.Context, key string, body []byte) error {
+	backend := m.backendFor(ctx, "LRU_ARTIFACT")
+	if backend == nil {
+		return errUnknownBackend
+	}
+	return backend.PutArtifact(ctx, key, body)
+}
diff --git a/cache/s3proxy/artifacts_test.go b/cache/s3proxy/artifacts_test.go
new file mode 100644
index 0000000..c001b3d
--- /dev/null
+++ b/cache/s3proxy/artifacts_test.go
@@ -0,0 +1,140 @@
+package s3proxy
+
+import (
+	"bytes"
+	"context"
+	"io"
+	"testing"
+
+	"github.com/buchgr/bazel-remote/v2/cache"
+	"github.com/buchgr/bazel-remote/v2/cache/lruflush"
+
+	"github.com/minio/minio-go/v7"
+)
+
+// Both proxies must satisfy the flusher's sink contract; a signature drift
+// should fail compilation here, not at the main.go type assertion.
+var (
+	_ lruflush.Sink = (*s3Cache)(nil)
+	_ lruflush.Sink = (*multiS3Cache)(nil)
+)
+
+func readObject(t *testing.T, c *s3Cache, bucket, key string) []byte {
+	t.Helper()
+	rc, _, _, err := c.mcore.GetObject(context.Background(), bucket, key, minio.GetObjectOptions{})
+	if err != nil {
+		t.Fatalf("GetObject(%s, %s): %v", bucket, key, err)
+	}
+	defer rc.Close()
+	body, err := io.ReadAll(rc)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return body
+}
+
+// TestPutArtifactRoutesByRequestBucket pins the artifact write path to the
+// same per-request bucket semantics as cache uploads: the selection's bucket
+// when one is attached, the backend default otherwise, key used verbatim.
+func TestPutArtifactRoutesByRequestBucket(t *testing.T) {
+	c := fakeS3Backend(t, "default-bucket", "tenant-bucket-1")
+
+	ctxTenant := cache.WithS3Backend(context.Background(),
+		cache.S3BackendSelection{Endpoint: backendKeyA, Bucket: "tenant-bucket-1"})
+	tenantBody := []byte(`{"schema_version":1}` + "\n")
+	if err := c.PutArtifact(ctxTenant, "tenant/v1/lru/00000000000000001000-p-1.jsonl", tenantBody); err != nil {
+		t.Fatalf("PutArtifact(tenant bucket): %v", err)
+	}
+	got := readObject(t, c, "tenant-bucket-1", "tenant/v1/lru/00000000000000001000-p-1.jsonl")
+	if !bytes.Equal(got, tenantBody) {
+		t.Fatalf("artifact body mismatch: %q", got)
+	}
+
+	if err := c.PutArtifact(context.Background(), "tenant/v1/lru/00000000000000002000-p-2.jsonl", tenantBody); err != nil {
+		t.Fatalf("PutArtifact(default bucket): %v", err)
+	}
+	readObject(t, c, "default-bucket", "tenant/v1/lru/00000000000000002000-p-2.jsonl")
+}
+
+// TestMultiPutArtifactDispatch pins multi-backend routing: the selection's
+// endpoint picks the backend, an unknown selector refuses rather than
+// guessing a shard, and no selector uses the default backend.
+func TestMultiPutArtifactDispatch(t *testing.T) {
+	a := fakeS3Backend(t, "default-bucket")
+	b := fakeS3Backend(t, "default-bucket")
+	m := &multiS3Cache{
+		backends: map[string]*s3Cache{"endpoint-a": a, "endpoint-b": b},
+		def:      a,
+	}
+	body := []byte("x\n")
+
+	ctxB := cache.WithS3Backend(context.Background(), cache.S3BackendSelection{Endpoint: "endpoint-b"})
+	if err := m.PutArtifact(ctxB, "t/lru/1-b.jsonl", body); err != nil {
+		t.Fatalf("PutArtifact via endpoint-b: %v", err)
+	}
+	readObject(t, b, "default-bucket", "t/lru/1-b.jsonl")
+	if _, _, _, err := a.mcore.GetObject(context.Background(), "default-bucket", "t/lru/1-b.jsonl", minio.GetObjectOptions{}); err == nil {
+		t.Fatal("artifact leaked to the non-selected backend")
+	}
+
+	ctxUnknown := cache.WithS3Backend(context.Background(), cache.S3BackendSelection{Endpoint: "endpoint-nope"})
+	if err := m.PutArtifact(ctxUnknown, "t/lru/2-x.jsonl", body); err == nil {
+		t.Fatal("PutArtifact with unknown selector must refuse")
+	}
+
+	if err := m.PutArtifact(context.Background(), "t/lru/3-d.jsonl", body); err != nil {
+		t.Fatalf("PutArtifact selector-less: %v", err)
+	}
+	readObject(t, a, "default-bucket", "t/lru/3-d.jsonl")
+}
+
+// TestPutArtifactRefusesWhenBreakerOpen pins the fail-fast contract: a sick
+// shard's artifact flushes must not stack PUTs onto it.
+func TestPutArtifactRefusesWhenBreakerOpen(t *testing.T) {
+	c := fakeS3Backend(t, "default-bucket")
+	tripBreaker(t, c.breaker)
+	if err := c.PutArtifact(context.Background(), "t/lru/1-a.jsonl", []byte("x\n")); err == nil {
+		t.Fatal("PutArtifact with open breaker must fail fast")
+	}
+}
+
+// TestArtifactFailuresCannotTripBreaker pins one direction of the
+// failure-isolation contract: artifact-only failures (here, a bucket that
+// does not exist — the shape of a tagging/policy/IAM gap) must never open
+// the data-plane breaker against customer traffic, no matter how many occur.
+func TestArtifactFailuresCannotTripBreaker(t *testing.T) {
+	c := fakeS3Backend(t, "default-bucket")
+	ctxBad := cache.WithS3Backend(context.Background(),
+		cache.S3BackendSelection{Endpoint: backendKeyA, Bucket: "no-such-bucket"})
+	for i := 0; i < 2*breakerConsecutiveFailures; i++ {
+		if err := c.PutArtifact(ctxBad, "t/lru/fail.jsonl", []byte("x\n")); err == nil {
+			t.Fatal("PutArtifact to a missing bucket must fail")
+		}
+	}
+	if got := c.breaker.State(); got != breakerClosed {
+		t.Fatalf("breaker state after %d artifact failures = %v, want closed", 2*breakerConsecutiveFailures, got)
+	}
+	// Customer traffic is unaffected.
+	if err := c.PutArtifact(context.Background(), "t/lru/ok.jsonl", []byte("x\n")); err != nil {
+		t.Fatalf("data plane degraded by artifact failures: %v", err)
+	}
+}
+
+// TestArtifactSuccessCannotHealFailureStreak pins the other direction: a
+// small artifact success must not reset a real data-plane failure streak.
+// Four data-plane failures, one successful artifact PUT, then the fifth
+// failure — the breaker must open, proving the artifact success recorded
+// nothing.
+func TestArtifactSuccessCannotHealFailureStreak(t *testing.T) {
+	c := fakeS3Backend(t, "default-bucket")
+	for i := 0; i < breakerConsecutiveFailures-1; i++ {
+		_ = c.breaker.Execute(func() breakerOutcome { return outcomeFailure })
+	}
+	if err := c.PutArtifact(context.Background(), "t/lru/mid-streak.jsonl", []byte("x\n")); err != nil {
+		t.Fatalf("artifact PUT during a not-yet-open streak should succeed: %v", err)
+	}
+	_ = c.breaker.Execute(func() breakerOutcome { return outcomeFailure })
+	if got := c.breaker.State(); got != breakerOpen {
+		t.Fatalf("breaker state after 4 failures + artifact success + 1 failure = %v, want open (artifact success must not reset the streak)", got)
+	}
+}
diff --git a/cache/s3proxy/breaker.go b/cache/s3proxy/breaker.go
index 1f5cd45..57cb5ef 100644
--- a/cache/s3proxy/breaker.go
+++ b/cache/s3proxy/breaker.go
@@ -62,9 +62,9 @@ var breakerTimeout = 15 * time.Second
 // Breaker states. The numeric values are the wire contract of the
 // bazel_remote_s3_breaker_state gauge.
 const (
-	breakerClosed int32 = iota // 0
-	breakerHalfOpen            // 1
-	breakerOpen                // 2
+	breakerClosed   int32 = iota // 0
+	breakerHalfOpen              // 1
+	breakerOpen                  // 2
 )
 
 func breakerStateName(state int32) string {
@@ -152,6 +152,18 @@ func (b *breaker) ExecuteNoProbe(call func() breakerOutcome) error {
 	return nil
 }
 
+// isClosed reports whether the breaker is currently closed, WITHOUT
+// advancing state or claiming the half-open probe. It exists for advisory
+// traffic (LRU artifact PUTs) that wants the sick-shard fail-fast but is
+// contractually forbidden from influencing the data-plane breaker in
+// either direction: it must not record outcomes, and it must not consume
+// the probe slot a real cache read needs to close the breaker.
+func (b *breaker) isClosed() bool {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	return b.state == breakerClosed
+}
+
 // allow decides whether a call may dial the backend, advancing
 // open -> half-open lazily once breakerTimeout has elapsed.
 func (b *breaker) allow() bool {
diff --git a/config/config.go b/config/config.go
index db96293..00b0c4b 100644
--- a/config/config.go
+++ b/config/config.go
@@ -97,6 +97,7 @@ type Config struct {
 	Dir                         string                    `yaml:"dir"`
 	MaxSize                     int                       `yaml:"max_size"`
 	MaxSizeHardLimit            int                       `yaml:"max_size_hard_limit"`
+	MaxEntries                  int64                     `yaml:"max_entries"`
 	StorageMode                 string                    `yaml:"storage_mode"`
 	ZstdImplementation          string                    `yaml:"zstd_implementation"`
 	HtpasswdFile                string                    `yaml:"htpasswd_file"`
@@ -207,6 +208,7 @@ func newFromArgs(dir string, maxSize int, storageMode string, zstdImplementation
 	accessLogLevel string,
 	logTimezone string,
 	maxSizeHardLimit int,
+	maxEntries int64,
 	maxBlobSize int64,
 	maxProxyBlobSize int64) (*Config, error) {
 
@@ -217,6 +219,7 @@ func newFromArgs(dir string, maxSize int, storageMode string, zstdImplementation
 		Dir:                         dir,
 		MaxSize:                     maxSize,
 		MaxSizeHardLimit:            maxSizeHardLimit,
+		MaxEntries:                  maxEntries,
 		StorageMode:                 storageMode,
 		ZstdImplementation:          zstdImplementation,
 		HtpasswdFile:                htpasswdFile,
@@ -763,6 +766,7 @@ func get(ctx *cli.Context) (*Config, error) {
 		ctx.String("access_log_level"),
 		ctx.String("log_timezone"),
 		ctx.Int("max_size_hard_limit"),
+		ctx.Int64("max_entries"),
 		ctx.Int64("max_blob_size"),
 		ctx.Int64("max_proxy_blob_size"),
 	)
diff --git a/main.go b/main.go
index e8d5166..b7f2c30 100644
--- a/main.go
+++ b/main.go
@@ -16,6 +16,7 @@ import (
 	auth "github.com/abbot/go-http-auth"
 
 	"github.com/buchgr/bazel-remote/v2/cache/disk"
+	"github.com/buchgr/bazel-remote/v2/cache/lruflush"
 
 	"github.com/buchgr/bazel-remote/v2/config"
 	"github.com/buchgr/bazel-remote/v2/ldap"
@@ -162,6 +163,7 @@ func run(ctx *cli.Context) error {
 		disk.WithMaxBlobSize(c.MaxBlobSize),
 		disk.WithProxyMaxBlobSize(c.MaxProxyBlobSize),
 		disk.WithMaxSizeHardLimit(int64(c.MaxSizeHardLimit) * 1024 * 1024 * 1024),
+		disk.WithMaxEntries(c.MaxEntries),
 		disk.WithAccessLogger(c.AccessLogger),
 	}
 	if c.ProxyBackend != nil {
@@ -171,11 +173,39 @@ func run(ctx *cli.Context) error {
 		opts = append(opts, disk.WithEndpointMetrics())
 	}
 
+	// LRU observation artifacts: on trusted (L1) nodes with an S3 proxy,
+	// buffer AC-access closures per tenant prefix and flush them as JSONL
+	// artifacts next to the objects they describe, for the web-side
+	// retention sweep. Advisory only — never affects request behavior.
+	// Requires trust mode because only trusted-mode requests carry the
+	// storage prefix an artifact is keyed under. Dark by default: the
+	// feature is explicitly opt-in (=1) so a routine binary roll can never
+	// activate it fleet-wide; rollout enables it per node and widens with
+	// evidence.
+	var lruFlusher *lruflush.Flusher
+	if os.Getenv("BAZEL_REMOTE_TRUST_STORAGE_PREFIX_HEADER") == "1" {
+		switch {
+		case os.Getenv("BAZEL_REMOTE_LRU_ARTIFACTS") != "1":
+			log.Println("LRU observation artifacts disabled (BAZEL_REMOTE_LRU_ARTIFACTS=1 enables)")
+		default:
+			if sink, ok := c.ProxyBackend.(lruflush.Sink); ok {
+				lruFlusher = lruflush.New(sink)
+				opts = append(opts, disk.WithLRUObserver(lruFlusher))
+				log.Println("LRU observation artifacts enabled: buffering AC-access closures per tenant prefix, periodic flush to the tenant's backend")
+			} else {
+				log.Println("LRU observation artifacts disabled: proxy backend cannot store artifacts (requires the S3 proxy)")
+			}
+		}
+	}
+
 	diskCache, err := disk.New(c.Dir, int64(c.MaxSize)*1024*1024*1024, opts...)
 	if err != nil {
 		log.Fatal(err)
 	}
 	diskCache.RegisterMetrics()
+	if lruFlusher != nil {
+		lruFlusher.Start()
+	}
 
 	servers := new(errgroup.Group)
 
@@ -242,7 +272,14 @@ func run(ctx *cli.Context) error {
 		idleTimer.Start()
 	}
 
-	return servers.Wait()
+	err = servers.Wait()
+	// Both servers have stopped accepting requests, so no new observations
+	// can arrive: flush what remains and wait for in-flight uploads.
+	if lruFlusher != nil {
+		log.Println("Draining buffered LRU observation artifacts")
+		lruFlusher.Drain()
+	}
+	return err
 }
 
 // validateMultiBackendTrust refuses to start a multi-backend (s3_proxy
diff --git a/utils/flags/flags.go b/utils/flags/flags.go
index 2ed0771..f6ae866 100644
--- a/utils/flags/flags.go
+++ b/utils/flags/flags.go
@@ -59,6 +59,17 @@ func GetCliFlags() []cli.Flag {
 				"value might be 5% larger than --max_size.",
 			EnvVars: []string{"BAZEL_REMOTE_MAX_SIZE_HARD_LIMIT"},
 		},
+		&cli.Int64Flag{
+			Name:  "max_entries",
+			Value: 0,
+			Usage: "If positive, the maximum number of entries resident in the disk cache's " +
+				"in-memory LRU index; least-recently-used entries are evicted past the cap, " +
+				"exactly like the byte budget but counting entries. Each resident entry costs " +
+				"~270 bytes of process memory regardless of blob size, so this bounds index " +
+				"memory on small-blob-heavy workloads that the byte budget alone does not. " +
+				"Zero or negative means no entry-count bound.",
+			EnvVars: []string{"BAZEL_REMOTE_MAX_ENTRIES"},
+		},
 		&cli.StringFlag{
 			Name:    "storage_mode",
 			Value:   "zstd",