diff --git a/.github/workflows/branch.yml b/.github/workflows/branch.yml
index 8f8368a8..398d45a6 100644
--- a/.github/workflows/branch.yml
+++ b/.github/workflows/branch.yml
@@ -175,6 +175,7 @@ jobs:
     env:
       GOLF_HUB_TEST_DB_URL: postgresql://moonbase_test:moonbase_test@postgres:5432/moonbase_test
       PG_TEST_DB_URL: postgresql://moonbase_test:moonbase_test@postgres:5432/moonbase_test
+      STATS_TEST_DB_URL: postgresql://moonbase_test:moonbase_test@postgres:5432/moonbase_test
 
     steps:
       - uses: actions/checkout@v7
diff --git a/domains/platform/apis/stats/BUILD.bazel b/domains/platform/apis/stats/BUILD.bazel
index 7cada874..f0712801 100644
--- a/domains/platform/apis/stats/BUILD.bazel
+++ b/domains/platform/apis/stats/BUILD.bazel
@@ -20,6 +20,8 @@ go_library(
 
 # store_test.go skips without STATS_TEST_DB_URL, like the repo's other
 # Postgres-gated suites — a local green run may have exercised no SQL.
+# CI supplies it from the postgres service; without env_inherit the
+# sandbox strips it and the suite skips there too.
 go_test(
     name = "stats_test",
     size = "small",
@@ -31,6 +33,11 @@ go_test(
         "store_test.go",
     ],
     embed = [":stats_lib"],
+    env_inherit = ["STATS_TEST_DB_URL"],
+    # One shared scratch database, and the version-bump test drops every
+    # aggregate table: concurrent runs would pull the tables out from under
+    # each other mid-flight.
+    tags = ["exclusive"],
 )
 
 go_binary(
diff --git a/domains/platform/apis/stats/README.md b/domains/platform/apis/stats/README.md
index 334b30e1..997f1411 100644
--- a/domains/platform/apis/stats/README.md
+++ b/domains/platform/apis/stats/README.md
@@ -16,18 +16,54 @@ conflict arm makes a duplicate application a no-op — so counts survive
 crashes without double-counting. Per-object failures are logged and
 retried next pass.
 
-Aggregates are bounded on purpose: hosts are Caddy's vhosts, methods
-collapse through the nine-verb rule the metrics rails use, user agents
-collapse to four classes (`ai_scraper`, `bot`, `browser`, `other`), and
-iili slugs are the one caller-shaped key — one path segment, max 64
-bytes, only on the redirect routes. The raw lines stay in S3, so a
-better classifier is a re-aggregation, not lost data.
+Aggregates are bounded per row, on purpose: hosts are Caddy's vhosts,
+methods collapse through the nine-verb rule the metrics rails use, and
+user agents collapse to four classes (`ai_scraper`, `bot`, `browser`,
+`other`). Each request row also carries a bounded agent name (#1458): the
+marker that classified it for AI scrapers and named bots, the UA's first
+product token (max 32 bytes) for the anonymous tail, and nothing for
+browsers, since every browser's token is `mozilla`. The token is the one
+caller-shaped key besides iili slugs, so an object may mint at most 500
+distinct ones before the rest collapse into a single `(more)` row — a
+scanner rotating its User-Agent gets one row, not one per request. The
+probe rollup counts requests whose path matched one of the scanner
+families in `classify.go` (`wordpress`, `env`, `git`, `php`, ...) and
+mints nothing for ordinary routes; there is deliberately no `admin`
+family, and backup-file shapes match only at the root, because Forgejo
+serves real archives and `.sql` files under deeper paths. iili slugs are
+one path segment, max 64 bytes, only on the two routes that reach iili.
+Row width is what's bounded; row count is what Postgres is for, which is
+the division of labor #1460 drew against the tsdb.
+
+The raw lines stay in S3, so a better classifier is a re-aggregation,
+not lost data — and re-aggregation is a mechanism, not a runbook. The
+store records a rollup version made of `RollupVersion` plus a hash of
+the schema DDL; a boot that finds a different one drops every aggregate
+table and processed marker in one transaction, recreates the tables, and
+the next pass recomputes everything from S3. Editing a table re-aggregates
+by itself; bump the constant when a classifier changes what a row means.
+The pass runs while the API serves, so for its length the counts climb
+back up from zero — minutes at this scale, and the log says when it is
+done.
+
+What stays ad hoc: IP-range clusters (a /24 key is caller-shaped and
+unbounded) and geography (nothing in the repo maps addresses to
+countries). Both are one query over the raw partitions in S3, which keep
+`request.remote_ip` and `request.client_ip` per line.
 
 ## The API
 
 - `GET /stats/v1/summary?days=7` — per day/host/agent-class request and
   error counts
 - `GET /stats/v1/iili/top?days=30&limit=20` — most-followed short links
+- `GET /stats/v1/agents?days=30&limit=500` — per day/host/class/agent
+  request and 403 counts, busiest rows first: which scrapers and bots hit
+  which host, and whether they back off after being refused
+- `GET /stats/v1/probes?days=30` — per host/scanner-family request counts
+  and how many were served (status < 400). On vhosts whose Caddy block
+  has no catch-all 404 (`api.muchq.com`, `gpt.muchq.com`), an unmatched
+  path is answered with an empty 200 and so reads as served; the 1d4 and
+  iili vhosts do have one, and there the column means what it says
 - `GET /health`
 
 Public through Caddy at `api.muchq.com/stats/v1/*`; the reasons for 500s
diff --git a/domains/platform/apis/stats/aggregate.go b/domains/platform/apis/stats/aggregate.go
index 36e555b5..39479a40 100644
--- a/domains/platform/apis/stats/aggregate.go
+++ b/domains/platform/apis/stats/aggregate.go
@@ -7,16 +7,20 @@ import (
 	"io"
 )
 
-// RequestKey is one row of the per-day request rollup: everything bounded,
-// nothing caller-controlled — host is one of Caddy's configured vhosts,
-// the method collapses through the same nine-verb rule the metrics rails
-// use, and the agent class is the four-value vocabulary in classify.go.
+// RequestKey is one row of the per-day request rollup. Host is one of
+// Caddy's configured vhosts, the method collapses through the same
+// nine-verb rule the metrics rails use, the agent class is the four-value
+// vocabulary in classify.go, and the agent is the bounded name AgentOf
+// pairs with it — so a host's traffic can be opened up into which
+// scrapers, which bots, and what the unclassified tail actually sends,
+// from the same rows the class totals come from.
 type RequestKey struct {
 	Date       string
 	Host       string
 	Status     int
 	Method     string
 	AgentClass string
+	Agent      string
 }
 
 // SlugKey is one row of the iili redirect rollup. The slug is
@@ -28,6 +32,28 @@ type SlugKey struct {
 	Status int
 }
 
+// ProbeKey is one row of the scanner rollup. Rows exist only for requests
+// whose path matched a family in ProbeOf, so the key is bounded by that
+// vocabulary and ordinary traffic mints nothing here.
+type ProbeKey struct {
+	Date   string
+	Host   string
+	Probe  string
+	Status int
+}
+
+// The most distinct product-token agent names one object may mint. Marker
+// names are bounded by their lists; tokens are whatever the anonymous tail
+// sends, and a scanner rotating its User-Agent per request would otherwise
+// turn one log object into one row per request — held in memory here and
+// then written one statement at a time. Past the cap the tail collapses
+// into one row named overflowAgent, so the counts stay right and the
+// object stays applicable.
+const (
+	maxTailAgentsPerObject = 500
+	overflowAgent          = "(more)"
+)
+
 // Rollup is one processed object's aggregates, accumulated in memory and
 // applied to the store in a single transaction with the processed marker —
 // so a crash between the two reprocesses the object rather than losing or
@@ -35,10 +61,34 @@ type SlugKey struct {
 type Rollup struct {
 	Requests map[RequestKey]int64
 	Slugs    map[SlugKey]int64
+	Probes   map[ProbeKey]int64
+
+	tailAgents map[string]bool
 }
 
 func NewRollup() *Rollup {
-	return &Rollup{Requests: map[RequestKey]int64{}, Slugs: map[SlugKey]int64{}}
+	return &Rollup{
+		Requests:   map[RequestKey]int64{},
+		Slugs:      map[SlugKey]int64{},
+		Probes:     map[ProbeKey]int64{},
+		tailAgents: map[string]bool{},
+	}
+}
+
+// boundedAgent applies the per-object cap to names that came from a
+// product token rather than a marker list. Browsers have no name to cap.
+func (r *Rollup) boundedAgent(class, agent string) string {
+	if class == AgentBrowser || markerNames[agent] {
+		return agent
+	}
+	if r.tailAgents[agent] {
+		return agent
+	}
+	if len(r.tailAgents) >= maxTailAgentsPerObject {
+		return overflowAgent
+	}
+	r.tailAgents[agent] = true
+	return agent
 }
 
 // caddyLine is the slice of Caddy's JSON access log this pipeline reads.
@@ -95,13 +145,18 @@ func (r *Rollup) Consume(reader io.Reader, date string) (skipped int, err error)
 			continue
 		}
 		method := boundedMethod(parsed.Request.Method)
+		agentClass, agent := AgentOf(parsed.userAgent())
 		r.Requests[RequestKey{
 			Date:       date,
 			Host:       parsed.Request.Host,
 			Status:     parsed.Status,
 			Method:     method,
-			AgentClass: AgentClassOf(parsed.userAgent()),
+			AgentClass: agentClass,
+			Agent:      r.boundedAgent(agentClass, agent),
 		}]++
+		if probe := ProbeOf(parsed.Request.URI); probe != "" {
+			r.Probes[ProbeKey{Date: date, Host: parsed.Request.Host, Probe: probe, Status: parsed.Status}]++
+		}
 		if slug := SlugOf(parsed.Request.Host, parsed.Request.Method, parsed.Request.URI); slug != "" {
 			r.Slugs[SlugKey{Date: date, Slug: slug, Status: parsed.Status}]++
 		}
diff --git a/domains/platform/apis/stats/aggregate_test.go b/domains/platform/apis/stats/aggregate_test.go
index 9e489c7d..18b5562f 100644
--- a/domains/platform/apis/stats/aggregate_test.go
+++ b/domains/platform/apis/stats/aggregate_test.go
@@ -1,6 +1,7 @@
 package stats
 
 import (
+	"fmt"
 	"strings"
 	"testing"
 )
@@ -14,6 +15,8 @@ const sampleLines = `{"status":200,"request":{"host":"api.1d4.net","method":"POS
 not json at all
 {"status":418,"request":{"method":"GET","uri":"/hostless","headers":{}}}
 {"status":200,"request":{"host":"api.muchq.com","method":"WEIRD","uri":"/x","headers":{}}}
+{"status":404,"request":{"host":"api.muchq.com","method":"GET","uri":"/wp-login.php","headers":{"User-Agent":["python-requests/2.32.0"]}}}
+{"status":404,"request":{"host":"api.muchq.com","method":"GET","uri":"/.env","headers":{"User-Agent":["Mozilla/5.0 (compatible; GPTBot/1.2)"]}}}
 `
 
 func TestConsumeAggregatesRequestsSlugsAndSkipsCorruptLines(t *testing.T) {
@@ -28,14 +31,23 @@ func TestConsumeAggregatesRequestsSlugsAndSkipsCorruptLines(t *testing.T) {
 	if skipped != 2 {
 		t.Errorf("skipped = %d, want 2", skipped)
 	}
-	if got := rollup.Requests[RequestKey{"2026-08-30", "api.1d4.net", 200, "POST", AgentBrowser}]; got != 2 {
+	// Browsers are one unnamed bucket; every other row carries its agent's
+	// bounded name, so "did meta back off after the 403" is a query over the
+	// same rows the class totals come from rather than a re-aggregation.
+	if got := rollup.Requests[RequestKey{"2026-08-30", "api.1d4.net", 200, "POST", AgentBrowser, ""}]; got != 2 {
 		t.Errorf("mcp browser POSTs = %d, want 2", got)
 	}
-	if got := rollup.Requests[RequestKey{"2026-08-30", "git.muchq.com", 403, "GET", AgentAIScraper}]; got != 1 {
+	if got := rollup.Requests[RequestKey{"2026-08-30", "git.muchq.com", 403, "GET", AgentAIScraper, "meta-externalagent"}]; got != 1 {
 		t.Errorf("blocked ai scraper = %d, want 1", got)
 	}
+	if got := rollup.Requests[RequestKey{"2026-08-30", "i.iili.uk", 302, "GET", AgentBot, "curl"}]; got != 1 {
+		t.Errorf("curl redirects = %d, want 1", got)
+	}
+	if got := rollup.Requests[RequestKey{"2026-08-30", "i.iili.uk", 404, "GET", AgentOther, "(empty)"}]; got != 1 {
+		t.Errorf("empty-UA rows = %d, want 1", got)
+	}
 	// An invented verb collapses like every metrics rail's method label.
-	if got := rollup.Requests[RequestKey{"2026-08-30", "api.muchq.com", 200, "CUSTOM", AgentOther}]; got != 1 {
+	if got := rollup.Requests[RequestKey{"2026-08-30", "api.muchq.com", 200, "CUSTOM", AgentOther, "(empty)"}]; got != 1 {
 		t.Errorf("CUSTOM-method row = %d, want 1", got)
 	}
 	// The redirect rollup counts per slug and status, across agent classes.
@@ -45,6 +57,59 @@ func TestConsumeAggregatesRequestsSlugsAndSkipsCorruptLines(t *testing.T) {
 	if got := rollup.Slugs[SlugKey{"2026-08-30", "gone", 404}]; got != 1 {
 		t.Errorf("gone-slug 404s = %d, want 1", got)
 	}
+	// Probe rows exist only for paths that match a scanner family.
+	if got := rollup.Probes[ProbeKey{"2026-08-30", "api.muchq.com", ProbeWordpress, 404}]; got != 1 {
+		t.Errorf("wordpress probes = %d, want 1", got)
+	}
+	if got := rollup.Probes[ProbeKey{"2026-08-30", "api.muchq.com", ProbeEnv, 404}]; got != 1 {
+		t.Errorf("env probes = %d, want 1", got)
+	}
+	if len(rollup.Probes) != 2 {
+		t.Errorf("probe rows = %v; ordinary routes must not mint any", rollup.Probes)
+	}
+}
+
+func TestConsumeCapsTheAnonymousAgentTailPerObject(t *testing.T) {
+	var lines strings.Builder
+	for i := 0; i < maxTailAgentsPerObject+100; i++ {
+		fmt.Fprintf(&lines, `{"status":200,"request":{"host":"h","method":"GET","uri":"/","headers":{"User-Agent":["junk-%d/1.0"]}}}`+"\n", i)
+	}
+	// Marker-named agents arriving after the cap keep their names; only
+	// product tokens are capped, and browsers were never named.
+	lines.WriteString(`{"status":200,"request":{"host":"h","method":"GET","uri":"/","headers":{"User-Agent":["Mozilla/5.0 (compatible; GPTBot/1.2)"]}}}` + "\n")
+	lines.WriteString(`{"status":200,"request":{"host":"h","method":"GET","uri":"/","headers":{"User-Agent":["curl/8.6.0"]}}}` + "\n")
+	lines.WriteString(`{"status":200,"request":{"host":"h","method":"GET","uri":"/","headers":{"User-Agent":["Mozilla/5.0 (Macintosh) Chrome/126.0"]}}}` + "\n")
+	rollup := NewRollup()
+
+	if _, err := rollup.Consume(strings.NewReader(lines.String()), "2026-08-30"); err != nil {
+		t.Fatal(err)
+	}
+
+	var total, overflow int64
+	names := map[string]bool{}
+	for key, count := range rollup.Requests {
+		total += count
+		if key.Agent == overflowAgent {
+			overflow += count
+		}
+		if key.AgentClass == AgentOther {
+			names[key.Agent] = true
+		}
+	}
+	if total != int64(maxTailAgentsPerObject+103) {
+		t.Errorf("total requests = %d; the cap must not lose a count", total)
+	}
+	if overflow != 100 {
+		t.Errorf("overflow row = %d, want the 100 past the cap", overflow)
+	}
+	if len(names) != maxTailAgentsPerObject+1 {
+		t.Errorf("distinct other-class names = %d, want the cap plus the overflow row", len(names))
+	}
+	if rollup.Requests[RequestKey{"2026-08-30", "h", 200, "GET", AgentAIScraper, "gptbot"}] != 1 ||
+		rollup.Requests[RequestKey{"2026-08-30", "h", 200, "GET", AgentBot, "curl"}] != 1 ||
+		rollup.Requests[RequestKey{"2026-08-30", "h", 200, "GET", AgentBrowser, ""}] != 1 {
+		t.Errorf("marker-named and browser rows were caught by the cap: %v", rollup.Requests)
+	}
 }
 
 func TestConsumeSurvivesOversizedLines(t *testing.T) {
diff --git a/domains/platform/apis/stats/api.go b/domains/platform/apis/stats/api.go
index 43c8cb33..c13c8185 100644
--- a/domains/platform/apis/stats/api.go
+++ b/domains/platform/apis/stats/api.go
@@ -13,6 +13,8 @@ import (
 type Reader interface {
 	Summary(ctx context.Context, days int) ([]SummaryRow, error)
 	TopSlugs(ctx context.Context, days, limit int) ([]SlugRow, error)
+	Agents(ctx context.Context, days, limit int) ([]AgentRow, error)
+	Probes(ctx context.Context, days int) ([]ProbeRow, error)
 }
 
 type Handlers struct {
@@ -67,6 +69,27 @@ func (h *Handlers) GetTopSlugs(w http.ResponseWriter, r *http.Request) {
 	writeJSON(w, map[string]any{"days": days, "rows": emptyIfNil(rows)})
 }
 
+func (h *Handlers) GetAgents(w http.ResponseWriter, r *http.Request) {
+	days := queryInt(r, "days", 30, 365)
+	limit := queryInt(r, "limit", 500, 2000)
+	rows, err := h.reader.Agents(r.Context(), days, limit)
+	if err != nil {
+		h.serverError(w, "agents", err)
+		return
+	}
+	writeJSON(w, map[string]any{"days": days, "rows": emptyIfNil(rows)})
+}
+
+func (h *Handlers) GetProbes(w http.ResponseWriter, r *http.Request) {
+	days := queryInt(r, "days", 30, 365)
+	rows, err := h.reader.Probes(r.Context(), days)
+	if err != nil {
+		h.serverError(w, "probes", err)
+		return
+	}
+	writeJSON(w, map[string]any{"days": days, "rows": emptyIfNil(rows)})
+}
+
 func (h *Handlers) serverError(w http.ResponseWriter, what string, err error) {
 	h.logger.Error("stats query failed", "query", what, "error", err)
 	// The reason goes to the log, not the wire: these endpoints are public.
diff --git a/domains/platform/apis/stats/api_test.go b/domains/platform/apis/stats/api_test.go
index 3a588f0e..228ad54b 100644
--- a/domains/platform/apis/stats/api_test.go
+++ b/domains/platform/apis/stats/api_test.go
@@ -15,6 +15,8 @@ import (
 type fakeReader struct {
 	summary   []SummaryRow
 	slugs     []SlugRow
+	agents    []AgentRow
+	probes    []ProbeRow
 	lastDays  int
 	lastLimit int
 	fail      bool
@@ -36,6 +38,22 @@ func (f *fakeReader) TopSlugs(_ context.Context, days, limit int) ([]SlugRow, er
 	return f.slugs, nil
 }
 
+func (f *fakeReader) Agents(_ context.Context, days, limit int) ([]AgentRow, error) {
+	if f.fail {
+		return nil, errors.New("db is having a day")
+	}
+	f.lastDays, f.lastLimit = days, limit
+	return f.agents, nil
+}
+
+func (f *fakeReader) Probes(_ context.Context, days int) ([]ProbeRow, error) {
+	if f.fail {
+		return nil, errors.New("db is having a day")
+	}
+	f.lastDays = days
+	return f.probes, nil
+}
+
 func handlersWith(reader *fakeReader) *Handlers {
 	return NewHandlers(reader, slog.New(slog.NewTextHandler(io.Discard, nil)))
 }
@@ -91,6 +109,24 @@ func TestTopSlugsPassesWindowAndLimit(t *testing.T) {
 	}
 }
 
+func TestNonPositiveAndOversizedParametersClampRatherThan400(t *testing.T) {
+	reader := &fakeReader{}
+	handlers := handlersWith(reader)
+
+	for _, raw := range []string{"0", "-5"} {
+		if recorder, _ := get(t, handlers.GetSummary, "/stats/v1/summary?days="+raw); recorder.Code != http.StatusOK {
+			t.Errorf("days=%s answered %d, want 200", raw, recorder.Code)
+		}
+		if reader.lastDays != 1 {
+			t.Errorf("days=%s clamped to %d, want 1", raw, reader.lastDays)
+		}
+	}
+	get(t, handlers.GetTopSlugs, "/stats/v1/iili/top?limit=99999")
+	if reader.lastLimit != 200 {
+		t.Errorf("slug limit clamped to %d, want 200", reader.lastLimit)
+	}
+}
+
 func TestAStoreFailureIs500WithoutTheReasonOnTheWire(t *testing.T) {
 	recorder, _ := get(t, handlersWith(&fakeReader{fail: true}).GetSummary, "/stats/v1/summary")
 	if recorder.Code != http.StatusInternalServerError {
@@ -100,3 +136,54 @@ func TestAStoreFailureIs500WithoutTheReasonOnTheWire(t *testing.T) {
 		t.Error("the failure reason leaked to a public endpoint")
 	}
 }
+
+func TestAgentsAndProbesShareTheWindowRules(t *testing.T) {
+	reader := &fakeReader{
+		agents: []AgentRow{{Date: "2026-08-30", Host: "git.muchq.com", AgentClass: AgentAIScraper, Agent: "meta-externalagent", Requests: 9, Blocked: 9}},
+		probes: []ProbeRow{{Host: "api.muchq.com", Probe: ProbeWordpress, Requests: 4, Served: 0}},
+	}
+	handlers := handlersWith(reader)
+
+	_, body := get(t, handlers.GetAgents, "/stats/v1/agents")
+	if reader.lastDays != 30 || reader.lastLimit != 500 {
+		t.Errorf("default agents (days, limit) = (%d, %d), want (30, 500)", reader.lastDays, reader.lastLimit)
+	}
+	row := body["rows"].([]any)[0].(map[string]any)
+	if row["agent"] != "meta-externalagent" || row["agent_class"] != AgentAIScraper || row["blocked"] != float64(9) {
+		t.Errorf("agent row = %v", row)
+	}
+
+	get(t, handlers.GetAgents, "/stats/v1/agents?days=99999&limit=99999")
+	if reader.lastDays != 365 || reader.lastLimit != 2000 {
+		t.Errorf("clamped agents (days, limit) = (%d, %d), want (365, 2000)", reader.lastDays, reader.lastLimit)
+	}
+
+	get(t, handlers.GetProbes, "/stats/v1/probes")
+	if reader.lastDays != 30 {
+		t.Errorf("default probes window = %d, want 30", reader.lastDays)
+	}
+	_, body = get(t, handlers.GetProbes, "/stats/v1/probes?days=99999")
+	if reader.lastDays != 365 {
+		t.Errorf("probe window clamped to %d, want 365", reader.lastDays)
+	}
+	row = body["rows"].([]any)[0].(map[string]any)
+	if row["probe"] != ProbeWordpress || row["served"] != float64(0) {
+		t.Errorf("probe row = %v", row)
+	}
+
+	// Empty is [], not null, on both — the dashboard maps them.
+	_, body = get(t, handlersWith(&fakeReader{}).GetAgents, "/stats/v1/agents")
+	if body["rows"] == nil {
+		t.Error("agents rows serialized as null; want []")
+	}
+	_, body = get(t, handlersWith(&fakeReader{}).GetProbes, "/stats/v1/probes")
+	if body["rows"] == nil {
+		t.Error("probes rows serialized as null; want []")
+	}
+	failing := handlersWith(&fakeReader{fail: true})
+	for name, handler := range map[string]http.HandlerFunc{"agents": failing.GetAgents, "probes": failing.GetProbes} {
+		if recorder, _ := get(t, handler, "/x"); recorder.Code != http.StatusInternalServerError {
+			t.Errorf("%s on a failing store = %d, want 500", name, recorder.Code)
+		}
+	}
+}
diff --git a/domains/platform/apis/stats/classify.go b/domains/platform/apis/stats/classify.go
index 44eea63f..182160ea 100644
--- a/domains/platform/apis/stats/classify.go
+++ b/domains/platform/apis/stats/classify.go
@@ -1,6 +1,9 @@
 package stats
 
-import "strings"
+import (
+	"regexp"
+	"strings"
+)
 
 // The bounded user-agent vocabulary the stats tables key on. Four values,
 // not a UA string per row: the point of classification is that "how much of
@@ -42,52 +45,191 @@ var aiScraperMarkers = []string{
 	"youbot",
 }
 
+// Bots worth a row of their own: search engines, SEO crawlers, link
+// unfurlers, internet scanners, and the HTTP libraries scanners drive. The
+// marker doubles as the agent name, so the vocabulary is this list. First
+// match wins, so a marker another agent's UA quotes must come after it —
+// Telegram's UA reads "TelegramBot (like TwitterBot)".
+var namedBotMarkers = []string{
+	"googlebot", "bingbot", "yandexbot", "duckduckbot", "baiduspider", "applebot",
+	"ahrefsbot", "semrushbot", "mj12bot", "dotbot", "petalbot", "dataforseobot",
+	"facebookexternalhit", "telegrambot", "twitterbot", "linkedinbot", "slackbot",
+	"discordbot", "whatsapp", "uptimerobot",
+	"censysinspect", "zgrab", "nuclei", "masscan",
+	"python-requests", "go-http-client", "okhttp", "curl", "wget", "scrapy",
+}
+
+// Generic shapes that mark a bot without naming one. Anything also in
+// namedBotMarkers is matched there first and does not belong here.
 var botMarkers = []string{
-	"bot", "spider", "crawl", "curl", "wget", "python-requests", "python/",
-	"go-http-client", "libwww", "httpclient", "okhttp", "scrapy", "java/",
-	"apache-httpclient", "phantom", "headless", "scanner", "nmap", "zgrab",
-	"masscan", "nuclei", "censys",
+	"bot", "spider", "crawl", "python/", "libwww", "httpclient", "java/",
+	"apache-httpclient", "phantom", "headless", "scanner", "nmap", "censys",
 }
 
-// AgentClassOf buckets a User-Agent header. Order matters: AI scrapers
-// self-identify with names that also match the generic bot markers.
-func AgentClassOf(userAgent string) string {
-	ua := strings.ToLower(userAgent)
-	if ua == "" {
-		return AgentOther
+// markerNames is every name AgentOf can return from a marker list, so the
+// rollup can tell a bounded name from a caller-shaped product token.
+var markerNames = func() map[string]bool {
+	names := map[string]bool{}
+	for _, marker := range aiScraperMarkers {
+		names[marker] = true
 	}
+	for _, marker := range namedBotMarkers {
+		names[marker] = true
+	}
+	return names
+}()
+
+// AgentOf buckets a User-Agent header and names the agent within the
+// bucket, bounded per class: AI scrapers and named bots name themselves by
+// the marker that matched, so those columns' vocabularies are exactly the
+// lists above; anonymous bots and the unclassified tail keep their product
+// token (one run of [a-z0-9._-], max 32 bytes) so a new crawler is readable
+// before it has a marker — except that a browser-shaped token is "mozilla"
+// for every one of them, so a generic bot with that token is named by the
+// marker it tripped ("headless", "bot") instead. Browsers are one unnamed
+// bucket. Order matters: AI scrapers self-identify with names that also
+// match the generic bot markers.
+func AgentOf(userAgent string) (class, name string) {
+	ua := strings.ToLower(userAgent)
 	for _, marker := range aiScraperMarkers {
 		if strings.Contains(ua, marker) {
-			return AgentAIScraper
+			return AgentAIScraper, marker
+		}
+	}
+	for _, marker := range namedBotMarkers {
+		if strings.Contains(ua, marker) {
+			return AgentBot, marker
 		}
 	}
 	for _, marker := range botMarkers {
 		if strings.Contains(ua, marker) {
-			return AgentBot
+			if token := productToken(ua); token != "mozilla" {
+				return AgentBot, token
+			}
+			return AgentBot, marker
 		}
 	}
 	if strings.HasPrefix(ua, "mozilla/") {
-		return AgentBrowser
+		return AgentBrowser, ""
+	}
+	return AgentOther, productToken(ua)
+}
+
+const (
+	emptyToken     = "(empty)"
+	maxTokenLength = 32
+)
+
+// productToken is the first run of [a-z0-9._-] in a lowercased UA — the
+// product name of "product/version (comment)" — truncated to a bound so a
+// scanner spraying UAs cannot mint wide rows, only many (and Rollup caps
+// how many).
+func productToken(lowerUA string) string {
+	start := -1
+	for i := 0; i < len(lowerUA); i++ {
+		if isTokenByte(lowerUA[i]) {
+			if start < 0 {
+				start = i
+			}
+			continue
+		}
+		if start >= 0 {
+			return clampToken(lowerUA[start:i])
+		}
+	}
+	if start < 0 {
+		return emptyToken
 	}
-	return AgentOther
+	return clampToken(lowerUA[start:])
+}
+
+func isTokenByte(b byte) bool {
+	return b >= 'a' && b <= 'z' || b >= '0' && b <= '9' || b == '.' || b == '_' || b == '-'
+}
+
+func clampToken(token string) string {
+	if len(token) > maxTokenLength {
+		return token[:maxTokenLength]
+	}
+	return token
+}
+
+// The bounded scanner-path vocabulary. A family is a shape scanners probe
+// for on any host — WordPress logins, dotfiles, PHP endpoints on hosts that
+// serve no PHP — and a request either matches one family or is not a
+// probe. A family that also matches real traffic is worse than none, which
+// is why there is no "admin" family (/admin/ is a one_d4 route) and why
+// backup files match only at the root: Forgejo serves repository archives
+// and raw files with the same extensions, several segments deep.
+const (
+	ProbeTraversal  = "traversal"
+	ProbeWordpress  = "wordpress"
+	ProbeEnv        = "env"
+	ProbeGit        = "git"
+	ProbeSecrets    = "secrets"
+	ProbePhpmyadmin = "phpmyadmin"
+	ProbePhp        = "php"
+	ProbeBackup     = "backup"
+	ProbeCgi        = "cgi"
+	ProbeJava       = "java"
+	ProbeRouter     = "router"
+)
+
+// Ordered: the first family to match wins, so the specific ones (a
+// traversal through cgi-bin, phpMyAdmin's index.php) sit above the shapes
+// they also match.
+var probeFamilies = []struct {
+	name  string
+	match *regexp.Regexp
+}{
+	{ProbeTraversal, regexp.MustCompile(`\.\./|\.\.\\|%2e%2e|/etc/passwd`)},
+	{ProbeWordpress, regexp.MustCompile(`wp-login|wp-admin|wp-content|wp-includes|wp-json|wp-config|xmlrpc\.php|wlwmanifest`)},
+	{ProbeEnv, regexp.MustCompile(`/\.env`)},
+	{ProbeGit, regexp.MustCompile(`/\.git(/|$)`)},
+	{ProbeSecrets, regexp.MustCompile(`/\.aws/|/\.ssh/|id_rsa|\.htpasswd|\.htaccess|\.bash_history|/\.docker/`)},
+	{ProbePhpmyadmin, regexp.MustCompile(`phpmyadmin|myadmin|/pma/|adminer`)},
+	{ProbePhp, regexp.MustCompile(`\.php($|/)`)},
+	{ProbeBackup, regexp.MustCompile(`^/[^/]+\.(sql|bak|zip|tar|tar\.gz|tgz|rar|7z|old|orig|swp)$`)},
+	{ProbeCgi, regexp.MustCompile(`/cgi-bin/`)},
+	{ProbeJava, regexp.MustCompile(`/actuator|/solr/|/jenkins|/manager/html|jmx-console`)},
+	{ProbeRouter, regexp.MustCompile(`/boaform/|/hnap1|/gponform/|/goform/|/tmui/`)},
+}
+
+// ProbeOf names the scanner family a request path belongs to, or "" when
+// the path is not a known probe shape. Matching is on the lowercased path
+// with the query string removed.
+func ProbeOf(uri string) string {
+	path := uri
+	if q := strings.IndexByte(path, '?'); q >= 0 {
+		path = path[:q]
+	}
+	path = strings.ToLower(path)
+	for _, family := range probeFamilies {
+		if family.match.MatchString(path) {
+			return family.name
+		}
+	}
+	return ""
 }
 
 // SlugOf extracts the iili short-link slug from a request, or "" when the
-// request is not a redirect lookup. Two shapes reach iili: the public
-// i.iili.uk/r/{slug} host and the api.muchq.com/iili/v1/r/{slug} route.
+// request is not a redirect lookup. Two shapes reach iili, and only two:
+// the public i.iili.uk/r/{slug} host (GET and HEAD) and the GET-only
+// api.muchq.com/iili/v1/r/{slug} route. The same path on another vhost, or
+// a HEAD on the api one, never reaches iili — Caddy answers it itself — so
+// it is not a follow.
 func SlugOf(host, method, uri string) string {
-	if method != "GET" && method != "HEAD" {
-		return ""
-	}
 	path := uri
 	if q := strings.IndexByte(path, '?'); q >= 0 {
 		path = path[:q]
 	}
 	var rest string
 	switch {
-	case strings.HasPrefix(host, "i.iili.uk") && strings.HasPrefix(path, "/r/"):
+	case strings.HasPrefix(host, "i.iili.uk") && (method == "GET" || method == "HEAD") &&
+		strings.HasPrefix(path, "/r/"):
 		rest = path[len("/r/"):]
-	case strings.HasPrefix(path, "/iili/v1/r/"):
+	case strings.HasPrefix(host, "api.muchq.com") && method == "GET" &&
+		strings.HasPrefix(path, "/iili/v1/r/"):
 		rest = path[len("/iili/v1/r/"):]
 	default:
 		return ""
diff --git a/domains/platform/apis/stats/classify_test.go b/domains/platform/apis/stats/classify_test.go
index 6f23ef45..0fa3dca7 100644
--- a/domains/platform/apis/stats/classify_test.go
+++ b/domains/platform/apis/stats/classify_test.go
@@ -28,8 +28,147 @@ func TestAgentClassificationCoversTheVocabulary(t *testing.T) {
 		{"definitely-not-a-browser", AgentOther},
 	}
 	for _, c := range cases {
-		if got := AgentClassOf(c.ua); got != c.want {
-			t.Errorf("AgentClassOf(%q) = %s, want %s", c.ua, got, c.want)
+		if got, _ := AgentOf(c.ua); got != c.want {
+			t.Errorf("AgentOf(%q) = %s, want %s", c.ua, got, c.want)
+		}
+	}
+}
+
+func TestAgentNamesAreBoundedPerClass(t *testing.T) {
+	cases := []struct {
+		ua        string
+		wantClass string
+		wantName  string
+	}{
+		// AI scrapers name themselves by marker, whatever else the UA says.
+		{"Mozilla/5.0 AppleWebKit/537.36; compatible; GPTBot/1.2; +https://openai.com/gptbot", AgentAIScraper, "gptbot"},
+		{"meta-externalagent/1.1 (+https://developers.facebook.com/docs/sharing/webmasters/crawler)", AgentAIScraper, "meta-externalagent"},
+		// Named bots by marker; anonymous tooling by its product token.
+		{"Mozilla/5.0 (compatible; Googlebot/2.1; +http://www.google.com/bot.html)", AgentBot, "googlebot"},
+		{"Mozilla/5.0 (compatible; AhrefsBot/7.0; +http://ahrefs.com/robot/)", AgentBot, "ahrefsbot"},
+		{"curl/8.6.0", AgentBot, "curl"},
+		{"python-requests/2.32.0", AgentBot, "python-requests"},
+		{"Go-http-client/2.0", AgentBot, "go-http-client"},
+		{"my-crawler/0.1 (+https://example.com)", AgentBot, "my-crawler"},
+		// Telegram quotes Twitter's marker in its own UA; the real one wins.
+		{"TelegramBot (like TwitterBot)", AgentBot, "telegrambot"},
+		{"Twitterbot/1.0", AgentBot, "twitterbot"},
+		// A browser-shaped generic bot would be "mozilla" like every browser,
+		// so the marker it tripped names it instead.
+		{"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) HeadlessChrome/120.0.0.0 Safari/537.36", AgentBot, "headless"},
+		{"Mozilla/5.0 (compatible; SomeNewBot/1.0)", AgentBot, "bot"},
+		// Browsers are one bucket: the token would be "mozilla" for all of them.
+		{"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0 Safari/537.36", AgentBrowser, ""},
+		// "other" keeps its product token so the unclassified tail is readable.
+		{"", AgentOther, "(empty)"},
+		{"definitely-not-a-browser", AgentOther, "definitely-not-a-browser"},
+		{"Weird Client 3.0", AgentOther, "weird"},
+		{"<script>alert(1)</script>", AgentOther, "script"},
+		{strings.Repeat("a", 200) + "/1.0", AgentOther, strings.Repeat("a", 32)},
+		{"/////", AgentOther, "(empty)"},
+	}
+	for _, c := range cases {
+		class, name := AgentOf(c.ua)
+		if class != c.wantClass || name != c.wantName {
+			t.Errorf("AgentOf(%q) = (%s, %q), want (%s, %q)", c.ua, class, name, c.wantClass, c.wantName)
+		}
+	}
+	// Every marker names itself, so the agent column's vocabulary for the
+	// two marker classes is exactly the lists and cannot drift from them —
+	// and markerNames knows every one of them.
+	for _, marker := range aiScraperMarkers {
+		ua := "Mozilla/5.0 (compatible; " + strings.ToUpper(marker) + "/1.0)"
+		if class, name := AgentOf(ua); class != AgentAIScraper || name != marker {
+			t.Errorf("AgentOf(%q) = (%s, %q), want (%s, %q)", ua, class, name, AgentAIScraper, marker)
+		}
+		if !markerNames[marker] {
+			t.Errorf("markerNames lacks %q", marker)
+		}
+	}
+	for _, marker := range namedBotMarkers {
+		ua := "Mozilla/5.0 (compatible; " + strings.ToUpper(marker) + "/1.0)"
+		if class, name := AgentOf(ua); class != AgentBot || name != marker {
+			t.Errorf("AgentOf(%q) = (%s, %q), want (%s, %q)", ua, class, name, AgentBot, marker)
+		}
+		if !markerNames[marker] {
+			t.Errorf("markerNames lacks %q", marker)
+		}
+	}
+	// A generic marker that a named one already covers is unreachable;
+	// keeping the lists disjoint is what makes the named list the vocabulary.
+	for _, marker := range botMarkers {
+		if markerNames[marker] {
+			t.Errorf("botMarkers repeats %q, which namedBotMarkers matches first", marker)
+		}
+	}
+}
+
+// The bare "bot" marker has no word boundary, so a phone brand ending in
+// it reads as a bot. Known and kept: a boundary rule would also lose
+// "Googlebot"-shaped names, and the AI list is consulted first regardless.
+func TestBotSubstringHasNoWordBoundaryOnPurpose(t *testing.T) {
+	if class, name := AgentOf("Mozilla/5.0 (Linux; Android 10; CUBOT X30) AppleWebKit/537.36 Chrome/120 Mobile Safari/537.36"); class != AgentBot || name != "bot" {
+		t.Errorf("CUBOT = (%s, %q); if this changed on purpose, update the comment above", class, name)
+	}
+}
+
+func TestProbeFamiliesAreBoundedAndRouteScoped(t *testing.T) {
+	cases := []struct {
+		uri  string
+		want string
+	}{
+		{"/wp-login.php", ProbeWordpress},
+		{"/wp-admin/", ProbeWordpress},
+		{"/xmlrpc.php", ProbeWordpress},
+		{"/blog/wp-includes/wlwmanifest.xml", ProbeWordpress},
+		{"/.env", ProbeEnv},
+		{"/.env.production?x=1", ProbeEnv},
+		{"/api/.env.bak", ProbeEnv},
+		{"/.envrc", ProbeEnv},
+		{"/.git/config", ProbeGit},
+		{"/.git/HEAD", ProbeGit},
+		{"/phpmyadmin/index.php", ProbePhpmyadmin},
+		{"/PMA/", ProbePhpmyadmin},
+		{"/adminer.php", ProbePhpmyadmin},
+		{"/vendor/phpunit/phpunit/src/Util/PHP/eval-stdin.php", ProbePhp},
+		{"/index.php?s=/Index/think/app/invokefunction", ProbePhp},
+		{"/.aws/credentials", ProbeSecrets},
+		{"/.ssh/id_rsa", ProbeSecrets},
+		{"/.htpasswd", ProbeSecrets},
+		{"/backup.sql", ProbeBackup},
+		{"/site.tar.gz", ProbeBackup},
+		{"/db.zip", ProbeBackup},
+		{"/../../etc/passwd", ProbeTraversal},
+		{"/cgi-bin/%2e%2e/%2e%2e/bin/sh", ProbeTraversal},
+		{"/cgi-bin/luci", ProbeCgi},
+		{"/manager/html", ProbeJava},
+		{"/actuator/health", ProbeJava},
+		{"/solr/admin/info/system", ProbeJava},
+		{"/boaform/admin/formLogin", ProbeRouter},
+		{"/HNAP1/", ProbeRouter},
+		{"/GponForm/diag_Form", ProbeRouter},
+		{"/WP-LOGIN.PHP", ProbeWordpress}, // case-insensitive
+
+		// Real routes on these hosts are not probes, however they are spelled.
+		{"/", ""},
+		{"/mcp", ""},
+		{"/iili/v1/r/abc", ""},
+		{"/stats/v1/summary?days=7", ""},
+		{"/.well-known/acme-challenge/token", ""},
+		{"/muchq/moonbase/src/branch/main/README.md", ""},
+		{"/index.html", ""},
+		{"/admin/reanalyze", ""}, // one_d4's real admin route; "admin" is not a family
+		{"/environment", ""},
+		{"/gitignore", ""},
+		{"/muchq/MoonBase.git/info/refs", ""}, // an HTTP clone, not a dotdir probe
+		// Forgejo serves archives and raw files with backup-looking
+		// extensions, always several segments deep; backups probe the root.
+		{"/muchq/MoonBase/archive/main.tar.gz", ""},
+		{"/muchq/MoonBase/raw/branch/main/migrations/V004__x.sql", ""},
+	}
+	for _, c := range cases {
+		if got := ProbeOf(c.uri); got != c.want {
+			t.Errorf("ProbeOf(%q) = %q, want %q", c.uri, got, c.want)
 		}
 	}
 }
@@ -50,6 +189,10 @@ func TestSlugExtractionIsBoundedAndRouteScoped(t *testing.T) {
 		{"i.iili.uk", "GET", "/r/" + strings.Repeat("a", 100), ""},
 		{"api.muchq.com", "GET", "/portrait/v1/trace", ""},
 		{"git.muchq.com", "GET", "/r/abc", ""},
+		// Only api.muchq.com routes /iili/v1/r/ to iili, and only for GET;
+		// anywhere else Caddy answers the path itself, so nothing was followed.
+		{"gpt.muchq.com", "GET", "/iili/v1/r/anything", ""},
+		{"api.muchq.com", "HEAD", "/iili/v1/r/xyz", ""},
 	}
 	for _, c := range cases {
 		if got := SlugOf(c.host, c.method, c.uri); got != c.want {
diff --git a/domains/platform/apis/stats/loop_test.go b/domains/platform/apis/stats/loop_test.go
index dbe00d2e..7b3a8bc4 100644
--- a/domains/platform/apis/stats/loop_test.go
+++ b/domains/platform/apis/stats/loop_test.go
@@ -98,7 +98,7 @@ func TestRunOnceAggregatesNewObjectsAndSkipsProcessedAndForeignKeys(t *testing.T
 		t.Fatal("the new object was not applied")
 	}
 	// The partition date keys the rollup — the object's own dt=, not today.
-	if got := rollup.Requests[RequestKey{"2026-08-31", "api.1d4.net", 200, "GET", AgentOther}]; got != 1 {
+	if got := rollup.Requests[RequestKey{"2026-08-31", "api.1d4.net", 200, "GET", AgentOther, "(empty)"}]; got != 1 {
 		t.Errorf("rollup rows = %v", rollup.Requests)
 	}
 }
diff --git a/domains/platform/apis/stats/main/main.go b/domains/platform/apis/stats/main/main.go
index d7692c86..57522db6 100644
--- a/domains/platform/apis/stats/main/main.go
+++ b/domains/platform/apis/stats/main/main.go
@@ -82,6 +82,8 @@ func main() {
 	router.HandleFunc("GET /health", handlers.Health)
 	router.HandleFunc("GET /stats/v1/summary", handlers.GetSummary)
 	router.HandleFunc("GET /stats/v1/iili/top", handlers.GetTopSlugs)
+	router.HandleFunc("GET /stats/v1/agents", handlers.GetAgents)
+	router.HandleFunc("GET /stats/v1/probes", handlers.GetProbes)
 
 	logger.Info("stats started", "port", port, "interval", interval.String())
 	if err := http.ListenAndServe(":"+port, router); err != nil {
diff --git a/domains/platform/apis/stats/store.go b/domains/platform/apis/stats/store.go
index b2e76ec0..5020405c 100644
--- a/domains/platform/apis/stats/store.go
+++ b/domains/platform/apis/stats/store.go
@@ -2,7 +2,9 @@ package stats
 
 import (
 	"context"
+	"crypto/sha256"
 	"fmt"
+	"strings"
 
 	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/pgxpool"
@@ -22,8 +24,9 @@ var schema = []string{
 		status int NOT NULL,
 		http_method text NOT NULL,
 		agent_class text NOT NULL,
+		agent text NOT NULL,
 		requests bigint NOT NULL,
-		PRIMARY KEY (dt, host, status, http_method, agent_class)
+		PRIMARY KEY (dt, host, status, http_method, agent_class, agent)
 	)`,
 	`CREATE TABLE IF NOT EXISTS iili_slug_stats (
 		dt date NOT NULL,
@@ -32,8 +35,46 @@ var schema = []string{
 		requests bigint NOT NULL,
 		PRIMARY KEY (dt, slug, status)
 	)`,
+	`CREATE TABLE IF NOT EXISTS probe_stats (
+		dt date NOT NULL,
+		host text NOT NULL,
+		probe text NOT NULL,
+		status int NOT NULL,
+		requests bigint NOT NULL,
+		PRIMARY KEY (dt, host, probe, status)
+	)`,
 }
 
+// RollupVersion is the meaning of what Consume computes. Bump it when a
+// classifier changes what a row means; a change to the tables themselves
+// needs no bump, because the recorded version also carries a hash of the
+// schema DDL. Either way, at boot a store whose recorded version differs
+// drops every aggregate table and every processed marker in one
+// transaction, recreates them from the schema above, and the next pass
+// recomputes all of it from the raw lines in S3. That is what "a schema
+// change is a re-aggregation" costs — one full pass, during which the
+// served counts climb back up from zero — and what makes it never data
+// loss. Version 2 named agents and added probe_stats.
+const RollupVersion = "2"
+
+func rollupVersionFor(meaning string, ddl []string) string {
+	sum := sha256.Sum256([]byte(strings.Join(ddl, "\n")))
+	return fmt.Sprintf("%s-%x", meaning, sum[:8])
+}
+
+func currentRollupVersion() string { return rollupVersionFor(RollupVersion, schema) }
+
+// The tables a re-aggregation rebuilds; every aggregate table in schema
+// belongs here, or a version bump leaves it double-counted.
+var rollupTables = []string{
+	"processed_log_objects", "request_stats", "iili_slug_stats", "probe_stats",
+}
+
+const metaSchema = `CREATE TABLE IF NOT EXISTS stats_meta (
+	key text PRIMARY KEY,
+	value text NOT NULL
+)`
+
 type Store struct {
 	pool *pgxpool.Pool
 }
@@ -43,13 +84,57 @@ func NewStore(ctx context.Context, databaseURL string) (*Store, error) {
 	if err != nil {
 		return nil, err
 	}
+	store := &Store{pool: pool}
+	if _, err := pool.Exec(ctx, metaSchema); err != nil {
+		pool.Close()
+		return nil, fmt.Errorf("applying schema: %w", err)
+	}
+	if err := store.dropAggregatesOnVersionChange(ctx); err != nil {
+		pool.Close()
+		return nil, err
+	}
 	for _, ddl := range schema {
 		if _, err := pool.Exec(ctx, ddl); err != nil {
 			pool.Close()
 			return nil, fmt.Errorf("applying schema: %w", err)
 		}
 	}
-	return &Store{pool: pool}, nil
+	return store, nil
+}
+
+// dropAggregatesOnVersionChange runs before the schema so that a version
+// bump which changed a table's columns recreates it: DROP rather than
+// TRUNCATE is what makes a column change and a new table the same
+// operation. The version row lands in the same transaction, so a crash
+// in between leaves either the old tables at the old version or no
+// tables at the new one — never new-version tables holding old rows.
+func (s *Store) dropAggregatesOnVersionChange(ctx context.Context) error {
+	tx, err := s.pool.Begin(ctx)
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback(ctx)
+
+	var recorded string
+	err = tx.QueryRow(ctx,
+		`SELECT value FROM stats_meta WHERE key = 'rollup_version' FOR UPDATE`).Scan(&recorded)
+	if err != nil && err != pgx.ErrNoRows {
+		return fmt.Errorf("reading rollup version: %w", err)
+	}
+	if recorded == currentRollupVersion() {
+		return nil
+	}
+	for _, table := range rollupTables {
+		if _, err := tx.Exec(ctx, "DROP TABLE IF EXISTS "+table); err != nil {
+			return fmt.Errorf("dropping %s for re-aggregation: %w", table, err)
+		}
+	}
+	if _, err := tx.Exec(ctx,
+		`INSERT INTO stats_meta (key, value) VALUES ('rollup_version', $1)
+		 ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value`, currentRollupVersion()); err != nil {
+		return err
+	}
+	return tx.Commit(ctx)
 }
 
 func (s *Store) Close() { s.pool.Close() }
@@ -100,11 +185,11 @@ func (s *Store) ApplyRollup(ctx context.Context, key string, rollup *Rollup) err
 	}
 	for k, count := range rollup.Requests {
 		if _, err := tx.Exec(ctx,
-			`INSERT INTO request_stats (dt, host, status, http_method, agent_class, requests)
-			 VALUES ($1, $2, $3, $4, $5, $6)
-			 ON CONFLICT (dt, host, status, http_method, agent_class)
+			`INSERT INTO request_stats (dt, host, status, http_method, agent_class, agent, requests)
+			 VALUES ($1, $2, $3, $4, $5, $6, $7)
+			 ON CONFLICT (dt, host, status, http_method, agent_class, agent)
 			 DO UPDATE SET requests = request_stats.requests + EXCLUDED.requests`,
-			k.Date, k.Host, k.Status, k.Method, k.AgentClass, count); err != nil {
+			k.Date, k.Host, k.Status, k.Method, k.AgentClass, k.Agent, count); err != nil {
 			return err
 		}
 	}
@@ -118,6 +203,16 @@ func (s *Store) ApplyRollup(ctx context.Context, key string, rollup *Rollup) err
 			return err
 		}
 	}
+	for k, count := range rollup.Probes {
+		if _, err := tx.Exec(ctx,
+			`INSERT INTO probe_stats (dt, host, probe, status, requests)
+			 VALUES ($1, $2, $3, $4, $5)
+			 ON CONFLICT (dt, host, probe, status)
+			 DO UPDATE SET requests = probe_stats.requests + EXCLUDED.requests`,
+			k.Date, k.Host, k.Probe, k.Status, count); err != nil {
+			return err
+		}
+	}
 	return tx.Commit(ctx)
 }
 
@@ -134,6 +229,29 @@ type SlugRow struct {
 	Requests int64  `json:"requests"`
 }
 
+// AgentRow is one day of one named agent on one host. Blocked is the 403
+// count — the signal for whether a scraper backs off after being refused.
+// The agent column is the one caller-shaped key in request_stats, so the
+// query takes a limit and hands back the busiest rows first: the tail a
+// UA-rotating scanner leaves is exactly the part that never makes the cut.
+type AgentRow struct {
+	Date       string `json:"date"`
+	Host       string `json:"host"`
+	AgentClass string `json:"agent_class"`
+	Agent      string `json:"agent"`
+	Requests   int64  `json:"requests"`
+	Blocked    int64  `json:"blocked"`
+}
+
+// ProbeRow is one scanner family on one host over the window. Served is
+// the sub-400 count: a probe that got an answer is the row to look at.
+type ProbeRow struct {
+	Host     string `json:"host"`
+	Probe    string `json:"probe"`
+	Requests int64  `json:"requests"`
+	Served   int64  `json:"served"`
+}
+
 func (s *Store) Summary(ctx context.Context, days int) ([]SummaryRow, error) {
 	rows, err := s.pool.Query(ctx,
 		`SELECT dt::text, host, agent_class,
@@ -176,3 +294,49 @@ func (s *Store) TopSlugs(ctx context.Context, days, limit int) ([]SlugRow, error
 	})
 	return out, err
 }
+
+func (s *Store) Agents(ctx context.Context, days, limit int) ([]AgentRow, error) {
+	rows, err := s.pool.Query(ctx,
+		`SELECT dt::text, host, agent_class, agent,
+		        SUM(requests) AS requests,
+		        COALESCE(SUM(requests) FILTER (WHERE status = 403), 0) AS blocked
+		 FROM request_stats
+		 WHERE dt >= current_date - $1::int
+		 GROUP BY dt, host, agent_class, agent
+		 ORDER BY requests DESC, dt DESC, host, agent_class, agent
+		 LIMIT $2`, days, limit)
+	if err != nil {
+		return nil, err
+	}
+	var out []AgentRow
+	var row AgentRow
+	_, err = pgx.ForEachRow(rows,
+		[]any{&row.Date, &row.Host, &row.AgentClass, &row.Agent, &row.Requests, &row.Blocked},
+		func() error {
+			out = append(out, row)
+			return nil
+		})
+	return out, err
+}
+
+func (s *Store) Probes(ctx context.Context, days int) ([]ProbeRow, error) {
+	rows, err := s.pool.Query(ctx,
+		`SELECT host, probe,
+		        SUM(requests) AS requests,
+		        COALESCE(SUM(requests) FILTER (WHERE status < 400), 0) AS served
+		 FROM probe_stats
+		 WHERE dt >= current_date - $1::int
+		 GROUP BY host, probe
+		 ORDER BY requests DESC, host, probe`, days)
+	if err != nil {
+		return nil, err
+	}
+	var out []ProbeRow
+	var row ProbeRow
+	_, err = pgx.ForEachRow(rows, []any{&row.Host, &row.Probe, &row.Requests, &row.Served},
+		func() error {
+			out = append(out, row)
+			return nil
+		})
+	return out, err
+}
diff --git a/domains/platform/apis/stats/store_test.go b/domains/platform/apis/stats/store_test.go
index d9d239f7..11107459 100644
--- a/domains/platform/apis/stats/store_test.go
+++ b/domains/platform/apis/stats/store_test.go
@@ -4,12 +4,13 @@ import (
 	"context"
 	"fmt"
 	"os"
+	"strings"
 	"testing"
 	"time"
 )
 
 // Real-database coverage for the store: schema, the processed-marker
-// transaction, upsert accumulation, and both read queries. Gated the same
+// transaction, upsert accumulation, and every read query. Gated the same
 // way the repo's other Postgres suites are: without STATS_TEST_DB_URL this
 // skips, and CI supplies the URL from its postgres service.
 func testStore(t *testing.T) *Store {
@@ -26,18 +27,30 @@ func testStore(t *testing.T) *Store {
 	return store
 }
 
+// The database persists across runs, so every run works under its own
+// host and object key; that is what lets the assertions below be exact.
+func uniqueFixture(t *testing.T) (date, host, key string) {
+	t.Helper()
+	nanos := time.Now().UnixNano()
+	date = time.Now().UTC().Format("2006-01-02")
+	host = fmt.Sprintf("host-%d.example", nanos)
+	key = fmt.Sprintf("logs/source=caddy/dt=%s/%s-%d.log.gz", date, t.Name(), nanos)
+	return date, host, key
+}
+
 func TestApplyRollupIsTransactionalIdempotentAndReadable(t *testing.T) {
 	store := testStore(t)
 	ctx := context.Background()
-	// Unique key per run: the database persists across test runs.
-	key := fmt.Sprintf("logs/source=caddy/dt=%s/test-%d.log.gz",
-		time.Now().UTC().Format("2006-01-02"), time.Now().UnixNano())
-	date := time.Now().UTC().Format("2006-01-02")
+	date, host, key := uniqueFixture(t)
 
 	rollup := NewRollup()
-	rollup.Requests[RequestKey{date, "test-host.example", 200, "GET", AgentBrowser}] = 5
-	rollup.Requests[RequestKey{date, "test-host.example", 403, "GET", AgentAIScraper}] = 2
-	rollup.Slugs[SlugKey{date, "test-slug", 302}] = 3
+	rollup.Requests[RequestKey{date, host, 200, "GET", AgentBrowser, ""}] = 5
+	rollup.Requests[RequestKey{date, host, 403, "GET", AgentAIScraper, "gptbot"}] = 2
+	rollup.Requests[RequestKey{date, host, 200, "GET", AgentAIScraper, "gptbot"}] = 1
+	rollup.Requests[RequestKey{date, host, 404, "GET", AgentOther, "(empty)"}] = 4
+	rollup.Slugs[SlugKey{date, host + "-slug", 302}] = 3
+	rollup.Probes[ProbeKey{date, host, ProbeEnv, 404}] = 4
+	rollup.Probes[ProbeKey{date, host, ProbeEnv, 200}] = 1
 
 	if pending, err := store.Unprocessed(ctx, []string{key}); err != nil || len(pending) != 1 {
 		t.Fatalf("Unprocessed = (%v, %v), want the fresh key pending", pending, err)
@@ -46,7 +59,8 @@ func TestApplyRollupIsTransactionalIdempotentAndReadable(t *testing.T) {
 		t.Fatal(err)
 	}
 	// Applying the same object twice must not double-count: the marker's
-	// conflict arm turns the second application into a no-op.
+	// conflict arm turns the second application into a no-op. The exact
+	// counts below are what prove it.
 	if err := store.ApplyRollup(ctx, key, rollup); err != nil {
 		t.Fatal(err)
 	}
@@ -58,34 +72,227 @@ func TestApplyRollupIsTransactionalIdempotentAndReadable(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	var browser, scraper *SummaryRow
-	for i := range summary {
-		row := &summary[i]
-		if row.Host == "test-host.example" && row.AgentClass == AgentBrowser {
-			browser = row
-		}
-		if row.Host == "test-host.example" && row.AgentClass == AgentAIScraper {
-			scraper = row
+	got := map[string]SummaryRow{}
+	for _, row := range summary {
+		if row.Host == host {
+			got[row.AgentClass] = row
 		}
 	}
-	if browser == nil || browser.Requests < 5 || browser.Errors != 0 {
-		t.Errorf("browser row = %+v", browser)
+	if row := got[AgentBrowser]; row.Requests != 5 || row.Errors != 0 {
+		t.Errorf("browser row = %+v, want 5 requests and no errors", row)
 	}
-	if scraper == nil || scraper.Requests < 2 || scraper.Errors < 2 {
-		t.Errorf("scraper row = %+v; 403s must count as errors", scraper)
+	if row := got[AgentAIScraper]; row.Requests != 3 || row.Errors != 2 {
+		t.Errorf("scraper row = %+v, want 3 requests with the 403s as errors", row)
+	}
+	if row := got[AgentOther]; row.Requests != 4 || row.Errors != 4 {
+		t.Errorf("other row = %+v, want 4 requests, all errors", row)
 	}
 
-	slugs, err := store.TopSlugs(ctx, 2, 100)
+	slugs, err := store.TopSlugs(ctx, 2, 1000)
 	if err != nil {
 		t.Fatal(err)
 	}
 	found := false
 	for _, row := range slugs {
-		if row.Slug == "test-slug" && row.Requests >= 3 {
+		if row.Slug == host+"-slug" {
 			found = true
+			if row.Requests != 3 {
+				t.Errorf("slug row = %+v, want 3", row)
+			}
 		}
 	}
 	if !found {
-		t.Errorf("test-slug missing from top slugs: %v", slugs)
+		t.Errorf("%s-slug missing from top slugs: %v", host, slugs)
+	}
+
+	agents, err := store.Agents(ctx, 2, 2000)
+	if err != nil {
+		t.Fatal(err)
+	}
+	named := map[string]AgentRow{}
+	for _, row := range agents {
+		if row.Host == host {
+			named[row.AgentClass+" "+row.Agent] = row
+		}
+	}
+	if row := named[AgentAIScraper+" gptbot"]; row.Requests != 3 || row.Blocked != 2 || row.Date != date {
+		t.Errorf("gptbot row = %+v, want 3 requests across statuses, 2 blocked, on %s", row, date)
+	}
+	if row := named[AgentOther+" (empty)"]; row.Requests != 4 || row.Blocked != 0 {
+		t.Errorf("empty-UA row = %+v; a 404 is not a block", row)
+	}
+	if row := named[AgentBrowser+" "]; row.Requests != 5 {
+		t.Errorf("browser row = %+v, want the unnamed bucket carried through", row)
+	}
+
+	probes, err := store.Probes(ctx, 2)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var env *ProbeRow
+	for i := range probes {
+		if probes[i].Host == host && probes[i].Probe == ProbeEnv {
+			env = &probes[i]
+		}
+	}
+	if env == nil || env.Requests != 5 || env.Served != 1 {
+		t.Errorf("env probe row = %+v; want 5 across statuses with the one 200 served", env)
+	}
+}
+
+func TestAgentsHonoursTheLimitBusiestFirst(t *testing.T) {
+	store := testStore(t)
+	ctx := context.Background()
+	date, host, key := uniqueFixture(t)
+
+	rollup := NewRollup()
+	rollup.Requests[RequestKey{date, host, 200, "GET", AgentBot, "curl"}] = 50
+	rollup.Requests[RequestKey{date, host, 200, "GET", AgentBot, "wget"}] = 5
+	if err := store.ApplyRollup(ctx, key, rollup); err != nil {
+		t.Fatal(err)
+	}
+
+	agents, err := store.Agents(ctx, 2, 1)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(agents) != 1 {
+		t.Fatalf("Agents(limit=1) returned %d rows", len(agents))
+	}
+	// Other runs' rows share the window, so the survivor is whichever row
+	// is busiest overall; it must at least outrank this run's small one.
+	if agents[0].Requests < 50 {
+		t.Errorf("the one row kept was %+v; want the busiest, not the first", agents[0])
+	}
+}
+
+func TestAVersionBumpDropsAggregatesAndMarkersForReaggregation(t *testing.T) {
+	store := testStore(t)
+	ctx := context.Background()
+	date, host, key := uniqueFixture(t)
+	rollup := NewRollup()
+	rollup.Requests[RequestKey{date, host, 200, "GET", AgentBot, "curl"}] = 1
+	rollup.Slugs[SlugKey{date, host + "-slug", 302}] = 1
+	rollup.Probes[ProbeKey{date, host, ProbeGit, 404}] = 1
+	if err := store.ApplyRollup(ctx, key, rollup); err != nil {
+		t.Fatal(err)
+	}
+
+	// Pretend those aggregates came from an older rollup shape.
+	if _, err := store.pool.Exec(ctx,
+		`UPDATE stats_meta SET value = 'stale' WHERE key = 'rollup_version'`); err != nil {
+		t.Fatal(err)
+	}
+	reopened, err := NewStore(ctx, os.Getenv("STATS_TEST_DB_URL"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(reopened.Close)
+
+	// The marker is gone, so the next pass recomputes the object; every
+	// aggregate it produced is gone with it, so nothing is double-counted.
+	if pending, err := reopened.Unprocessed(ctx, []string{key}); err != nil || len(pending) != 1 {
+		t.Errorf("Unprocessed after a version bump = (%v, %v), want the key pending again", pending, err)
+	}
+	summary, err := reopened.Summary(ctx, 2)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, row := range summary {
+		if row.Host == host {
+			t.Errorf("request aggregates survived the version bump: %+v", row)
+		}
+	}
+	agents, err := reopened.Agents(ctx, 2, 2000)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, row := range agents {
+		if row.Host == host {
+			t.Errorf("agent aggregates survived the version bump: %+v", row)
+		}
+	}
+	slugs, err := reopened.TopSlugs(ctx, 2, 1000)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, row := range slugs {
+		if row.Slug == host+"-slug" {
+			t.Errorf("slug aggregates survived the version bump: %+v", row)
+		}
+	}
+	probes, err := reopened.Probes(ctx, 2)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, row := range probes {
+		if row.Host == host {
+			t.Errorf("probe aggregates survived the version bump: %+v", row)
+		}
+	}
+	var recorded string
+	if err := reopened.pool.QueryRow(ctx,
+		`SELECT value FROM stats_meta WHERE key = 'rollup_version'`).Scan(&recorded); err != nil || recorded != currentRollupVersion() {
+		t.Errorf("recorded version = (%q, %v), want %q", recorded, err, currentRollupVersion())
+	}
+
+	// Reopening at the same version is a no-op: the store must not wipe
+	// itself on every boot.
+	if err := reopened.ApplyRollup(ctx, key, rollup); err != nil {
+		t.Fatal(err)
+	}
+	again, err := NewStore(ctx, os.Getenv("STATS_TEST_DB_URL"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(again.Close)
+	if pending, err := again.Unprocessed(ctx, []string{key}); err != nil || len(pending) != 0 {
+		t.Errorf("Unprocessed after a same-version reopen = (%v, %v), want none", pending, err)
+	}
+}
+
+// A column added to a table without anyone remembering to bump the
+// constant must still re-aggregate: the recorded version carries the DDL.
+func TestTheRecordedVersionFollowsTheSchemaText(t *testing.T) {
+	changed := append([]string{}, schema...)
+	changed[1] = strings.Replace(changed[1], "agent text NOT NULL,", "agent text NOT NULL,\n\t\textra int,", 1)
+	if changed[1] == schema[1] {
+		t.Fatal("the fixture did not change the DDL; the test proves nothing")
+	}
+	if rollupVersionFor(RollupVersion, changed) == rollupVersionFor(RollupVersion, schema) {
+		t.Error("a DDL change left the rollup version unchanged")
+	}
+	if rollupVersionFor("3", schema) == rollupVersionFor(RollupVersion, schema) {
+		t.Error("a meaning bump left the rollup version unchanged")
+	}
+}
+
+// A version change recreates the tables rather than emptying them, so a
+// table whose columns changed shape comes back in the new shape.
+func TestAVersionChangeRecreatesTablesInTheirNewShape(t *testing.T) {
+	store := testStore(t)
+	ctx := context.Background()
+	date, host, key := uniqueFixture(t)
+
+	// An older deployment's request_stats, without the agent column.
+	for _, ddl := range []string{
+		`DROP TABLE request_stats`,
+		`CREATE TABLE request_stats (dt date NOT NULL, host text NOT NULL, requests bigint NOT NULL)`,
+		`UPDATE stats_meta SET value = 'older' WHERE key = 'rollup_version'`,
+	} {
+		if _, err := store.pool.Exec(ctx, ddl); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	reopened, err := NewStore(ctx, os.Getenv("STATS_TEST_DB_URL"))
+	if err != nil {
+		t.Fatalf("reopening over an old-shaped table: %v", err)
+	}
+	t.Cleanup(reopened.Close)
+	rollup := NewRollup()
+	rollup.Requests[RequestKey{date, host, 200, "GET", AgentBot, "curl"}] = 1
+	if err := reopened.ApplyRollup(ctx, key, rollup); err != nil {
+		t.Errorf("the new-shape insert failed after the version change: %v", err)
 	}
 }