diff --git a/internal/intelligence/detector.go b/internal/intelligence/detector.go
index a6fd4783..10de6c5a 100644
--- a/internal/intelligence/detector.go
+++ b/internal/intelligence/detector.go
@@ -1,3 +1,11 @@
+// Package intelligence derives anomalies and log patterns from telemetry.
+//
+// Every SQL string in this file goes through internal/query's validator, which
+// rejects "--" outright. A SQL comment in one of these statements therefore
+// fails the query at runtime, on a background goroutine, as a logged error
+// nobody is watching -- that is how latency detection stopped reporting for
+// half an hour. Rationale goes in Go comments; TestNoSQLCommentsInQueryStrings
+// enforces it.
 package intelligence
 
 import (
@@ -154,7 +162,7 @@ func (d *Detector) detectErrorRateAnomalies(ctx context.Context, start, end time
 				service as service_name,
 				COUNT(*) FILTER (WHERE status IN ('STATUS_CODE_ERROR', 'ERROR')) AS error_count,
 				COUNT(*) AS total_count,
-				(COUNT(*) FILTER (WHERE status IN ('STATUS_CODE_ERROR', 'ERROR'))::FLOAT / COUNT(*)::FLOAT) AS error_rate
+				(COUNT(*) FILTER (WHERE status IN ('STATUS_CODE_ERROR', 'ERROR'))::DOUBLE / COUNT(*)::DOUBLE) AS error_rate
 			FROM spans
 			WHERE start_unix_nano >= %d AND start_unix_nano < %d
 			%s
@@ -165,7 +173,7 @@ func (d *Detector) detectErrorRateAnomalies(ctx context.Context, start, end time
 				service as service_name,
 				COUNT(*) FILTER (WHERE status IN ('STATUS_CODE_ERROR', 'ERROR')) AS error_count,
 				COUNT(*) AS total_count,
-				(COUNT(*) FILTER (WHERE status IN ('STATUS_CODE_ERROR', 'ERROR'))::FLOAT / COUNT(*)::FLOAT) AS error_rate,
+				(COUNT(*) FILTER (WHERE status IN ('STATUS_CODE_ERROR', 'ERROR'))::DOUBLE / COUNT(*)::DOUBLE) AS error_rate,
 				STDDEV(CASE WHEN status IN ('STATUS_CODE_ERROR', 'ERROR') THEN 1.0 ELSE 0.0 END) AS error_stddev
 			FROM spans
 			WHERE start_unix_nano >= %d AND start_unix_nano < %d
@@ -220,6 +228,10 @@ func (d *Detector) detectErrorRateAnomalies(ctx context.Context, start, end time
 
 // detectLatencyAnomalies detects latency degradation
 func (d *Detector) detectLatencyAnomalies(ctx context.Context, start, end time.Time) []Anomaly {
+	// The percentile below is approx_quantile, not PERCENTILE_CONT: the exact
+	// form is holistic and retains every value of every group on the raw
+	// allocator, outside anything memory_limit bounds. This runs every 60s over
+	// a 15-minute window. See serviceRollupP95SQL.
 	startNano := start.UnixNano()
 	endNano := end.UnixNano()
 	namespace := d.duck.DefaultNamespace()
@@ -229,7 +241,7 @@ func (d *Detector) detectLatencyAnomalies(ctx context.Context, start, end time.T
 		WITH current_period AS (
 			SELECT
 				service as service_name,
-				PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY duration_ms) AS p95_latency
+				approx_quantile(duration_ms, 0.95) AS p95_latency
 			FROM spans
 			WHERE start_unix_nano >= %d AND start_unix_nano < %d
 			  AND kind = 'SPAN_KIND_SERVER'
@@ -240,7 +252,7 @@ func (d *Detector) detectLatencyAnomalies(ctx context.Context, start, end time.T
 			SELECT
 				service as service_name,
 				time_bucket(INTERVAL '5 minutes', start_time) AS bucket,
-				PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY duration_ms) AS p95_latency
+				approx_quantile(duration_ms, 0.95) AS p95_latency
 			FROM spans
 			WHERE start_unix_nano >= %d AND start_unix_nano < %d
 			  AND kind = 'SPAN_KIND_SERVER'
@@ -300,22 +312,32 @@ func (d *Detector) detectLatencyAnomalies(ctx context.Context, start, end time.T
 	return anomalies
 }
 
-// detectVolumeAnomalies detects unusual traffic volume changes
-func (d *Detector) detectVolumeAnomalies(ctx context.Context, start, end time.Time) []Anomaly {
-	startNano := start.UnixNano()
-	endNano := end.UnixNano()
-	namespace := d.duck.DefaultNamespace()
-	scope := detectorScopeClause(namespace)
-
-	sql := fmt.Sprintf(`
+// volumeAnomalySQL compares span volume in [startNano, endNano) against the
+// window of equal length immediately before it.
+//
+// Both sides are averaged over 5-minute buckets, and that symmetry is the whole
+// point. The current side used to be a single COUNT(*) over the entire window
+// while the baseline averaged per-bucket counts, so on a 15-minute window the
+// current figure was three times the baseline for every service no matter what
+// the traffic did. Every service was permanently a critical volume anomaly:
+// steady traffic scored z=75 in TestVolumeAnomalyComparesEqualWindows, and the
+// live demo sat at health score 0 with 17 meaningless critical insights.
+func volumeAnomalySQL(startNano, endNano int64, scope string) string {
+	return fmt.Sprintf(`
 		WITH current_period AS (
 			SELECT
-				service as service_name,
-				COUNT(*) AS span_count
-			FROM spans
-			WHERE start_unix_nano >= %d AND start_unix_nano < %d
-			%s
-			GROUP BY service
+				service_name,
+				AVG(cnt) AS span_count
+			FROM (
+				SELECT
+					service as service_name,
+					COUNT(*) AS cnt
+				FROM spans
+				WHERE start_unix_nano >= %d AND start_unix_nano < %d
+				%s
+				GROUP BY service, time_bucket(INTERVAL '5 minutes', start_time)
+			) subq
+			GROUP BY service_name
 		),
 		baseline_period AS (
 			SELECT
@@ -335,7 +357,7 @@ func (d *Detector) detectVolumeAnomalies(ctx context.Context, start, end time.Ti
 		)
 		SELECT
 			c.service_name,
-			c.span_count::FLOAT AS current_count,
+			c.span_count::DOUBLE AS current_count,
 			COALESCE(b.avg_count, 0.0) AS baseline_count,
 			CASE
 				WHEN b.count_stddev > 0 THEN (c.span_count - b.avg_count) / b.count_stddev
@@ -344,6 +366,16 @@ func (d *Detector) detectVolumeAnomalies(ctx context.Context, start, end time.Ti
 		FROM current_period c
 		LEFT JOIN baseline_period b ON c.service_name = b.service_name
 	`, startNano, endNano, scope, startNano-endNano+startNano, startNano, scope)
+}
+
+// detectVolumeAnomalies detects unusual traffic volume changes
+func (d *Detector) detectVolumeAnomalies(ctx context.Context, start, end time.Time) []Anomaly {
+	startNano := start.UnixNano()
+	endNano := end.UnixNano()
+	namespace := d.duck.DefaultNamespace()
+	scope := detectorScopeClause(namespace)
+
+	sql := volumeAnomalySQL(startNano, endNano, scope)
 
 	resp := d.duck.ExecuteSQL(ctx, query.SQLRequest{Query: sql})
 	if resp.Error != "" {
diff --git a/internal/intelligence/sql_comments_test.go b/internal/intelligence/sql_comments_test.go
new file mode 100644
index 00000000..b1f475d8
--- /dev/null
+++ b/internal/intelligence/sql_comments_test.go
@@ -0,0 +1,63 @@
+package intelligence
+
+import (
+	"os"
+	"regexp"
+	"strings"
+	"testing"
+)
+
+// Every statement this package issues goes through a validator that rejects
+// SQL comments outright (internal/query/sql.go: "SQL comments (--) are not
+// allowed"). A "--" inside a query string therefore fails at runtime, on a
+// background goroutine, as a logged error nobody is watching -- the anomaly
+// detector silently stopped reporting latency for half an hour that way.
+//
+// The compiler cannot catch it and no unit test here executes these strings,
+// so this reads the source instead. Rationale belongs in Go comments.
+func TestNoSQLCommentsInQueryStrings(t *testing.T) {
+	source, err := os.ReadFile("detector.go")
+	if err != nil {
+		t.Fatal(err)
+	}
+	// Raw-string literals are where the SQL lives.
+	literals := regexp.MustCompile("(?s)`[^`]*`").FindAllString(string(source), -1)
+	if len(literals) == 0 {
+		t.Fatal("no raw string literals found; this guard is no longer looking at the right thing")
+	}
+	found := 0
+	for _, literal := range literals {
+		if !strings.Contains(strings.ToUpper(literal), "SELECT") {
+			continue
+		}
+		found++
+		for i, line := range strings.Split(literal, "\n") {
+			if idx := strings.Index(line, "--"); idx >= 0 {
+				t.Errorf("SQL comment in a query string (line %d of a literal): %q\n"+
+					"the validator rejects these at runtime; put the explanation in a Go comment",
+					i+1, strings.TrimSpace(line))
+			}
+		}
+	}
+	if found == 0 {
+		t.Fatal("no SELECT literals found; this guard is no longer looking at the right thing")
+	}
+	t.Logf("checked %d SQL literals", found)
+}
+
+// DuckDB's FLOAT is single precision, so the driver returns float32 and the
+// `row["x"].(float64)` every caller here writes fails -- silently yielding 0.0.
+// internal/query now widens float32 on the way out, which disarms the trap, but
+// a rate or a count has no business being single precision in the first place
+// and the next reader should not have to know about the widening to trust it.
+func TestNoFloatCastsInQueryStrings(t *testing.T) {
+	source, err := os.ReadFile("detector.go")
+	if err != nil {
+		t.Fatal(err)
+	}
+	for i, line := range strings.Split(string(source), "\n") {
+		if strings.Contains(strings.ToUpper(line), "::FLOAT") {
+			t.Errorf("detector.go:%d casts to FLOAT (single precision): %q\nuse ::DOUBLE", i+1, strings.TrimSpace(line))
+		}
+	}
+}
diff --git a/internal/intelligence/volume_sql_test.go b/internal/intelligence/volume_sql_test.go
new file mode 100644
index 00000000..7c982bf4
--- /dev/null
+++ b/internal/intelligence/volume_sql_test.go
@@ -0,0 +1,150 @@
+package intelligence
+
+import (
+	"database/sql"
+	"fmt"
+	"math"
+	"testing"
+	"time"
+)
+
+// The current window and the baseline window are the same length, but they were
+// not counted the same way: the baseline averaged per-5-minute-bucket counts
+// while the current side summed the whole 15 minutes. Current was therefore
+// about three times the baseline by construction, for every service, always --
+// so every service was permanently a critical volume anomaly and the demo's
+// health score sat at 0 with 17 "critical" insights that meant nothing.
+//
+// Steady traffic must produce no anomaly. That is the property here.
+func TestVolumeAnomalyComparesEqualWindows(t *testing.T) {
+	db, err := sql.Open("duckdb", "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.Close()
+	if _, err := db.Exec(`CREATE TABLE spans (
+		service TEXT, namespace TEXT, start_time TIMESTAMP, start_unix_nano BIGINT
+	)`); err != nil {
+		t.Fatal(err)
+	}
+
+	// 30 minutes of steady traffic in six 5-minute buckets, with enough jitter
+	// that STDDEV is not zero -- a perfectly flat fixture divides by zero and
+	// the CASE returns 0.0, which would pass against the broken form too.
+	end := time.Date(2026, 9, 21, 18, 0, 0, 0, time.UTC)
+	start := end.Add(-15 * time.Minute)
+	baselineStart := start.Add(-15 * time.Minute)
+	perBucket := []int{98, 103, 99, 101, 97, 102}
+	for bucket, count := range perBucket {
+		at := baselineStart.Add(time.Duration(bucket) * 5 * time.Minute)
+		for _, service := range []string{"cart", "checkout", "frontend"} {
+			for i := range count {
+				ts := at.Add(time.Duration(i) * time.Millisecond)
+				if _, err := db.Exec(`INSERT INTO spans VALUES (?, 'default', ?, ?)`,
+					service, ts, ts.UnixNano()); err != nil {
+					t.Fatal(err)
+				}
+			}
+		}
+	}
+
+	query := volumeAnomalySQL(start.UnixNano(), end.UnixNano(), "")
+	rows, err := db.Query(query)
+	if err != nil {
+		t.Fatalf("volume query: %v", err)
+	}
+	defer rows.Close()
+
+	seen := 0
+	for rows.Next() {
+		var service string
+		var current, baseline, zScore float64
+		if err := rows.Scan(&service, &current, &baseline, &zScore); err != nil {
+			t.Fatal(err)
+		}
+		seen++
+		if current <= 0 {
+			t.Errorf("%s: current = %v, want the spans it actually served", service, current)
+		}
+		ratio := current / baseline
+		if ratio < 0.8 || ratio > 1.25 {
+			t.Errorf("%s: current %.0f against baseline %.0f (%.2fx) -- the two windows are not counted the same way",
+				service, current, baseline, ratio)
+		}
+		if math.Abs(zScore) >= 3 {
+			t.Errorf("%s: steady traffic scored z=%.2f, which reports as a critical anomaly", service, zScore)
+		}
+	}
+	if err := rows.Err(); err != nil {
+		t.Fatal(err)
+	}
+	if seen != 3 {
+		t.Fatalf("got %d services, want 3", seen)
+	}
+	fmt.Print()
+}
+
+// The other direction, and the one that matters more: a test asserting only
+// that steady traffic is quiet passes just as well against a detector that
+// never fires at all. A service whose traffic really does collapse must still
+// be reported.
+func TestVolumeAnomalyStillFiresOnARealDrop(t *testing.T) {
+	db, err := sql.Open("duckdb", "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.Close()
+	if _, err := db.Exec(`CREATE TABLE spans (
+		service TEXT, namespace TEXT, start_time TIMESTAMP, start_unix_nano BIGINT
+	)`); err != nil {
+		t.Fatal(err)
+	}
+
+	end := time.Date(2026, 9, 21, 18, 0, 0, 0, time.UTC)
+	start := end.Add(-15 * time.Minute)
+	baselineStart := start.Add(-15 * time.Minute)
+	perBucket := []int{98, 103, 99, 101, 97, 102}
+	for bucket, count := range perBucket {
+		at := baselineStart.Add(time.Duration(bucket) * 5 * time.Minute)
+		inCurrentWindow := !at.Before(start)
+		for _, service := range []string{"steady", "collapsing"} {
+			// "collapsing" serves its baseline, then drops to 5% of it.
+			if service == "collapsing" && inCurrentWindow {
+				count = count / 20
+			}
+			for i := range count {
+				ts := at.Add(time.Duration(i) * time.Millisecond)
+				if _, err := db.Exec(`INSERT INTO spans VALUES (?, 'default', ?, ?)`,
+					service, ts, ts.UnixNano()); err != nil {
+					t.Fatal(err)
+				}
+			}
+			count = perBucket[bucket]
+		}
+	}
+
+	rows, err := db.Query(volumeAnomalySQL(start.UnixNano(), end.UnixNano(), ""))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer rows.Close()
+	scores := map[string]float64{}
+	for rows.Next() {
+		var service string
+		var current, baseline, zScore float64
+		if err := rows.Scan(&service, &current, &baseline, &zScore); err != nil {
+			t.Fatal(err)
+		}
+		scores[service] = zScore
+		t.Logf("%-11s current=%.1f baseline=%.1f z=%.2f", service, current, baseline, zScore)
+	}
+	if err := rows.Err(); err != nil {
+		t.Fatal(err)
+	}
+	if z, ok := scores["collapsing"]; !ok || math.Abs(z) < 3 {
+		t.Errorf("a service that lost 95%% of its traffic scored z=%.2f (present: %v); it must be reported", z, ok)
+	}
+	if z := scores["steady"]; math.Abs(z) >= 3 {
+		t.Errorf("steady traffic scored z=%.2f alongside it", z)
+	}
+}
diff --git a/internal/metrics/duckdb_memory_test.go b/internal/metrics/duckdb_memory_test.go
new file mode 100644
index 00000000..4ac7ea5b
--- /dev/null
+++ b/internal/metrics/duckdb_memory_test.go
@@ -0,0 +1,110 @@
+package metrics
+
+import (
+	"strings"
+	"sync"
+	"testing"
+
+	"github.com/prometheus/client_golang/prometheus"
+	dto "github.com/prometheus/client_model/go"
+)
+
+// gatherDuckDBMemory returns the exported per-tag memory series, by tag.
+func gatherDuckDBMemory(t *testing.T) map[string]float64 {
+	t.Helper()
+	families, err := prometheus.DefaultGatherer.Gather()
+	if err != nil {
+		t.Fatal(err)
+	}
+	out := map[string]float64{}
+	for _, family := range families {
+		if family.GetName() != "fanout_duckdb_memory_bytes" {
+			continue
+		}
+		for _, metric := range family.GetMetric() {
+			for _, label := range metric.GetLabel() {
+				if label.GetName() == "tag" {
+					out[label.GetValue()] = metric.GetGauge().GetValue()
+				}
+			}
+		}
+	}
+	return out
+}
+
+// The whole purpose of this series is to measure what DuckDB holds right now,
+// so that the untracked gap against process RSS can be derived from it. A
+// failed read must therefore export nothing. Reporting the last successful
+// figures as if they were current is the one outcome that actively misleads:
+// the gap is computed against a stale sum, and the number that was supposed to
+// reveal an impending out-of-memory kill reads normal.
+func TestDuckDBMemoryDropsSeriesWhenTheReadFails(t *testing.T) {
+	release := SetDuckDBMemorySource(func() map[string]int64 {
+		return map[string]int64{"BASE_TABLE": 4096}
+	})
+	if got := gatherDuckDBMemory(t); got["BASE_TABLE"] != 4096 {
+		release()
+		t.Fatalf("healthy read exported %v, want BASE_TABLE=4096", got)
+	}
+	release()
+
+	// The source is installed but cannot read: the closure in internal/query
+	// returns nil for a query error, a scan error, or a partial read.
+	release = SetDuckDBMemorySource(func() map[string]int64 { return nil })
+	defer release()
+	if got := gatherDuckDBMemory(t); len(got) != 0 {
+		t.Errorf("after a failed read the gauge still exports %v; stale values are reported as current", got)
+	}
+}
+
+// Prometheus scrapes concurrently, and a GaugeVec that is Reset and refilled on
+// every scrape is shared mutable state: two scrapes interleave, and one of them
+// observes the other's half-finished refill -- missing tags, and a sum that
+// understates what DuckDB holds.
+func TestDuckDBMemoryIsSafeUnderConcurrentScrapes(t *testing.T) {
+	release := SetDuckDBMemorySource(func() map[string]int64 {
+		return map[string]int64{"BASE_TABLE": 1, "HASH_TABLE": 2, "ORDER_BY": 3}
+	})
+	defer release()
+
+	var wg sync.WaitGroup
+	for range 8 {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for range 25 {
+				families, err := prometheus.DefaultGatherer.Gather()
+				if err != nil {
+					t.Error(err)
+					return
+				}
+				for _, family := range families {
+					if family.GetName() != "fanout_duckdb_memory_bytes" {
+						continue
+					}
+					if n := len(family.GetMetric()); n != 3 {
+						t.Errorf("scrape saw %d tags, want 3: a concurrent refill was observed half-done", n)
+						return
+					}
+					_ = dto.MetricType_GAUGE
+				}
+			}
+		}()
+	}
+	wg.Wait()
+}
+
+// The gauge must not invent a series when nothing has installed a source,
+// because an empty family reads exactly like "DuckDB is holding nothing".
+func TestDuckDBMemoryExportsNothingWithoutASource(t *testing.T) {
+	if got := gatherDuckDBMemory(t); len(got) != 0 {
+		t.Errorf("with no source installed the gauge exports %v", got)
+	}
+	if families, err := prometheus.DefaultGatherer.Gather(); err == nil {
+		for _, family := range families {
+			if strings.Contains(family.GetName(), "duckdb_memory_scrape") {
+				t.Errorf("%s exists only to pump a push gauge; a collector does not need it", family.GetName())
+			}
+		}
+	}
+}
diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go
index ee75a3cc..57fbf856 100644
--- a/internal/metrics/metrics.go
+++ b/internal/metrics/metrics.go
@@ -297,10 +297,10 @@ func RecordIngest(signal string, count int) {
 	IngestTotal.WithLabelValues(signal).Add(float64(count))
 }
 
-// RecordFlush records a flush event
 // RecordIngestShed counts one refused telemetry request.
 func RecordIngestShed() { IngestShedTotal.Inc() }
 
+// RecordFlush records a flush event.
 func RecordFlush(signal string, durationSec float64) {
 	FlushTotal.WithLabelValues(signal).Inc()
 	FlushDuration.WithLabelValues(signal).Observe(durationSec)
@@ -314,6 +314,64 @@ func RecordFlush(signal string, durationSec float64) {
 // on the second one. Unset, every gauge reads zero.
 var duckDBPoolStats atomic.Pointer[func() sql.DBStats]
 
+// duckDBMemoryStats is the live source for the per-tag DuckDB memory gauge.
+//
+// memory_limit bounds DuckDB's buffer pool and nothing else, so the figure that
+// actually predicts an out-of-memory kill is the gap between what the process
+// holds and what DuckDB admits to holding. Exporting the tags makes that gap
+// directly observable: untracked = process_resident_memory_bytes
+// - sum(fanout_duckdb_memory_bytes) - go_memstats_heap_sys_bytes. Without it
+// every diagnosis of this process is an inference, which is how a whole day
+// got spent optimising a Go heap that was never the problem.
+var duckDBMemoryStats atomic.Pointer[func() map[string]int64]
+
+// SetDuckDBMemorySource installs the reader for the per-tag memory gauge.
+func SetDuckDBMemorySource(tags func() map[string]int64) (release func()) {
+	if tags == nil {
+		return func() {}
+	}
+	installed := &tags
+	duckDBMemoryStats.Store(installed)
+	return func() { duckDBMemoryStats.CompareAndSwap(installed, nil) }
+}
+
+// duckDBMemoryDesc is emitted by a Collector, not held in a GaugeVec.
+//
+// The obvious implementation -- a GaugeVec that a scrape hook Resets and
+// refills -- is shared mutable state across concurrent scrapes, and a scrape
+// that lands mid-refill exports a subset of the tags. That understates what
+// DuckDB holds and correspondingly overstates the untracked gap, which is the
+// single number this series exists to compute. It also leaves the last good
+// values in place when a read fails, reporting them as current: the reading
+// that was meant to reveal an impending out-of-memory kill then looks normal.
+//
+// A Collector has neither problem. Each scrape builds its own metrics from its
+// own read, and a read that returns nothing exports nothing -- the series goes
+// absent, which is visibly different from "DuckDB is holding nothing".
+var duckDBMemoryDesc = prometheus.NewDesc(
+	"fanout_duckdb_memory_bytes",
+	"DuckDB memory usage by tag, as DuckDB itself accounts for it",
+	[]string{"tag"}, nil,
+)
+
+type duckDBMemoryCollector struct{}
+
+func (duckDBMemoryCollector) Describe(ch chan<- *prometheus.Desc) { ch <- duckDBMemoryDesc }
+
+func (duckDBMemoryCollector) Collect(ch chan<- prometheus.Metric) {
+	fn := duckDBMemoryStats.Load()
+	if fn == nil {
+		return
+	}
+	for tag, bytes := range (*fn)() {
+		ch <- prometheus.MustNewConstMetric(duckDBMemoryDesc, prometheus.GaugeValue, float64(bytes), tag)
+	}
+}
+
+func init() {
+	prometheus.DefaultRegisterer.MustRegister(duckDBMemoryCollector{})
+}
+
 // SetDuckDBPoolSource points the connection-pool gauges at a pool and returns
 // the function that stops reading it. The last caller wins; the returned
 // release only clears the source if it is still the one it installed, so a
diff --git a/internal/observability/service.go b/internal/observability/service.go
index b6a64590..7adc7b7a 100644
--- a/internal/observability/service.go
+++ b/internal/observability/service.go
@@ -28,7 +28,7 @@ var (
 type DB = queryrows.Queryer
 
 type traceReader interface {
-	Trace(context.Context, telemetry.TraceQuery) ([]telemetry.IndexedSpan, error)
+	Trace(context.Context, telemetry.TraceQuery) ([]telemetry.IndexedSpan, telemetry.TraceTotals, error)
 }
 
 type Service struct {
diff --git a/internal/observability/service_test.go b/internal/observability/service_test.go
index 8fe59615..caeaaffa 100644
--- a/internal/observability/service_test.go
+++ b/internal/observability/service_test.go
@@ -330,10 +330,6 @@ func TestTraceSelectsRecentErrorAndCorrelatesLogs(t *testing.T) {
 	}}); err != nil {
 		t.Fatalf("commit trace fixture: %v", err)
 	}
-	mock.ExpectQuery(regexp.QuoteMeta(traceSummaryQuery)).
-		WithArgs("trace-1", start, end, "prod", "prod").
-		WillReturnRows(sqlmock.NewRows([]string{"span_count", "service_count", "duration_ms", "has_error"}).
-			AddRow(int64(2), int64(2), 200.0, true))
 	mock.ExpectQuery(regexp.QuoteMeta(traceLogsQuery)).
 		WithArgs("trace-1", start, end, "prod", "prod", 20).
 		WillReturnRows(sqlmock.NewRows([]string{"time", "severity", "service", "body", "trace_id", "span_id"}).
@@ -514,10 +510,6 @@ func TestTraceLogsUseFullScopeEventTimeAcrossBatches(t *testing.T) {
 			t.Fatal(err)
 		}
 	}
-	mock.ExpectQuery(regexp.QuoteMeta(traceSummaryQuery)).
-		WithArgs("trace-order", start, start.Add(time.Hour), "prod", "prod").
-		WillReturnRows(sqlmock.NewRows([]string{"span_count", "service_count", "duration_ms", "has_error"}).
-			AddRow(int64(1), int64(1), 1.0, false))
 	mock.ExpectQuery(regexp.QuoteMeta(traceLogsQuery)).
 		WithArgs("trace-order", start, start.Add(time.Hour), "prod", "prod", 10).
 		WillReturnRows(sqlmock.NewRows([]string{"time", "severity", "service", "body", "trace_id", "span_id"}).
@@ -544,10 +536,6 @@ func TestTraceUsesIndexedParquet(t *testing.T) {
 	}}}); err != nil {
 		t.Fatal(err)
 	}
-	mock.ExpectQuery(regexp.QuoteMeta(traceSummaryQuery)).
-		WithArgs("parquet-trace", start, end, "prod", "prod").
-		WillReturnRows(sqlmock.NewRows([]string{"span_count", "service_count", "duration_ms", "has_error"}).
-			AddRow(int64(1), int64(1), 25.0, true))
 	mock.ExpectQuery(regexp.QuoteMeta(traceLogsQuery)).
 		WithArgs("parquet-trace", start, end, "prod", "prod", 10).
 		WillReturnRows(sqlmock.NewRows([]string{"time", "severity", "service", "body", "trace_id", "span_id"}).
@@ -583,10 +571,6 @@ func TestTraceCombinesIndexedSpansAcrossBatches(t *testing.T) {
 	}}}); err != nil {
 		t.Fatal(err)
 	}
-	mock.ExpectQuery(regexp.QuoteMeta(traceSummaryQuery)).
-		WithArgs("split-trace", start, end, "prod", "prod").
-		WillReturnRows(sqlmock.NewRows([]string{"span_count", "service_count", "duration_ms", "has_error"}).
-			AddRow(int64(2), int64(2), 2400025.0, true))
 	mock.ExpectQuery(regexp.QuoteMeta(traceLogsQuery)).
 		WithArgs("split-trace", start, end, "prod", "prod", 10).
 		WillReturnRows(sqlmock.NewRows([]string{"time", "severity", "service", "body", "trace_id", "span_id"}))
@@ -613,10 +597,6 @@ func TestTraceReadsRecentRootFromParquetIndex(t *testing.T) {
 	}}}); err != nil {
 		t.Fatal(err)
 	}
-	mock.ExpectQuery(regexp.QuoteMeta(traceSummaryQuery)).
-		WithArgs("new-trace", start, end, "prod", "prod").
-		WillReturnRows(sqlmock.NewRows([]string{"span_count", "service_count", "duration_ms", "has_error"}).
-			AddRow(int64(1), int64(1), 10.0, false))
 	mock.ExpectQuery(regexp.QuoteMeta(traceLogsQuery)).
 		WithArgs("new-trace", start, end, "prod", "prod", 10).
 		WillReturnRows(sqlmock.NewRows([]string{"time", "severity", "service", "body", "trace_id", "span_id"}))
@@ -643,10 +623,6 @@ func TestTraceFiltersIndexedSpansByNamespace(t *testing.T) {
 	}}); err != nil {
 		t.Fatal(err)
 	}
-	mock.ExpectQuery(regexp.QuoteMeta(traceSummaryQuery)).
-		WithArgs("shared-trace", start, end, "prod", "prod").
-		WillReturnRows(sqlmock.NewRows([]string{"span_count", "service_count", "duration_ms", "has_error"}).
-			AddRow(int64(1), int64(1), 0.0, false))
 	mock.ExpectQuery(regexp.QuoteMeta(traceLogsQuery)).
 		WithArgs("shared-trace", start, end, "prod", "prod", 10).
 		WillReturnRows(sqlmock.NewRows([]string{"time", "severity", "service", "body", "trace_id", "span_id"}))
diff --git a/internal/observability/trace.go b/internal/observability/trace.go
index aff01314..745a1467 100644
--- a/internal/observability/trace.go
+++ b/internal/observability/trace.go
@@ -19,15 +19,6 @@ ORDER BY MAX(CASE WHEN upper(status) IN ('ERROR', 'STATUS_CODE_ERROR') THEN 1 EL
          MAX(end_time) - MIN(start_time) DESC
 LIMIT 1`
 
-const traceSummaryQuery = `
-SELECT
-  CAST(count(*) AS BIGINT),
-  CAST(count(DISTINCT service) AS BIGINT),
-  COALESCE((max(end_unix_nano) - min(start_unix_nano)) / 1000000.0, 0),
-  COALESCE(bool_or(upper(status) IN ('ERROR', 'STATUS_CODE_ERROR')), false)
-FROM spans
-WHERE trace_id = ? AND start_time >= ? AND start_time < ? AND (? = '' OR namespace = ?)`
-
 const traceLogsQuery = `
 SELECT time, severity, coalesce(service, ''), body, coalesce(trace_id, ''), coalesce(span_id, '')
 FROM logs
@@ -66,7 +57,7 @@ func (s *Service) Trace(ctx context.Context, scope Scope, traceID, service strin
 	dataSource := "parquet_index"
 	data := TraceDetail{TraceID: traceID, Services: []string{}, Spans: []TraceSpan{}, Logs: []LogEntry{}}
 	if traceID != "" {
-		storedSpans, readErr := s.repository.Trace(ctx, telemetry.TraceQuery{
+		storedSpans, totals, readErr := s.repository.Trace(ctx, telemetry.TraceQuery{
 			TraceID: traceID, Namespace: scope.Namespace,
 			StartNanos: scope.Start.UnixNano(), EndNanos: scope.End.UnixNano(), Limit: limit,
 		})
@@ -78,7 +69,10 @@ func (s *Service) Trace(ctx context.Context, scope Scope, traceID, service strin
 		}
 
 		// Only the page's own service list is derived here. Duration and
-		// has_error describe the whole trace and come from the aggregate below.
+		// has_error describe the whole trace, so they come from the totals the
+		// repository accumulated over every span it walked -- not from this
+		// page, which is why trace_detail used to report the page's span count
+		// and duration as the trace's.
 		serviceSet := make(map[string]struct{})
 		for _, span := range data.Spans {
 			if span.Service != "" {
@@ -91,13 +85,15 @@ func (s *Service) Trace(ctx context.Context, scope Scope, traceID, service strin
 		sort.Strings(data.Services)
 		// The page above is what limit admitted. Everything the caller reads as
 		// a fact about the trace — how many spans it has, how many services it
-		// crosses, how long it took, whether it failed — comes from an
-		// aggregate over the whole trace instead, so a narrow page cannot
-		// silently redefine the trace. The aggregate reads no rows into this
-		// process: it is counted in DuckDB and returns one row.
-		if err := s.traceTotals(ctx, scope, traceID, &data); err != nil {
-			return Result[TraceDetail]{}, err
-		}
+		// crosses, how long it took, whether it failed — describes the whole
+		// trace, so a narrow page cannot silently redefine it.
+		//
+		// These come from the index walk that produced the page: it already
+		// decodes every row of the trace and applies the same predicates, and
+		// only the heap is bounded by limit. An aggregate over every batch file
+		// would answer the same question and cost a second pass.
+		data.SpanCount, data.ServiceCount = totals.Spans, totals.Services
+		data.DurationMS, data.HasError = totals.DurationMS, totals.HasError
 		data.Truncated = data.SpanCount > len(data.Spans)
 		data.Logs, err = s.traceLogsFromParquet(ctx, scope, traceID, limit)
 		if err != nil {
@@ -115,28 +111,6 @@ func (s *Service) Trace(ctx context.Context, scope Scope, traceID, service strin
 	return Result[TraceDetail]{Schema: TraceSchema, Summary: summary, Data: data, Provenance: s.provenanceFor(scope, dataSource)}, nil
 }
 
-// traceTotals fills the fields that describe the trace rather than the page.
-// A trace that has aged out of the window reports zeros, which leaves the
-// summary saying the trace holds no spans — true for the window asked about.
-func (s *Service) traceTotals(ctx context.Context, scope Scope, traceID string, data *TraceDetail) error {
-	rows, err := s.db.QueryContext(ctx, traceSummaryQuery, traceID, scope.Start, scope.End, scope.Namespace, scope.Namespace)
-	if err != nil {
-		return fmt.Errorf("query trace totals: %w", err)
-	}
-	defer rows.Close()
-	if rows.Next() {
-		var spanCount, serviceCount int64
-		if err := rows.Scan(&spanCount, &serviceCount, &data.DurationMS, &data.HasError); err != nil {
-			return fmt.Errorf("scan trace totals: %w", err)
-		}
-		data.SpanCount, data.ServiceCount = int(spanCount), int(serviceCount)
-	}
-	if err := rows.Err(); err != nil {
-		return fmt.Errorf("iterate trace totals: %w", err)
-	}
-	return nil
-}
-
 func (s *Service) traceLogsFromParquet(ctx context.Context, scope Scope, traceID string, limit int) ([]LogEntry, error) {
 	rows, err := s.db.QueryContext(ctx, traceLogsQuery, traceID, scope.Start, scope.End, scope.Namespace, scope.Namespace, limit)
 	if err != nil {
diff --git a/internal/observability/trace_test.go b/internal/observability/trace_test.go
index 6821b745..9e2f8ebf 100644
--- a/internal/observability/trace_test.go
+++ b/internal/observability/trace_test.go
@@ -2,7 +2,6 @@ package observability
 
 import (
 	"context"
-	"database/sql"
 	"regexp"
 	"strings"
 	"testing"
@@ -44,10 +43,6 @@ func TestTracePagedSpansStillDescribeTheWholeTrace(t *testing.T) {
 	}}); err != nil {
 		t.Fatal(err)
 	}
-	mock.ExpectQuery(regexp.QuoteMeta(traceSummaryQuery)).
-		WithArgs("wide-trace", start, end, "prod", "prod").
-		WillReturnRows(sqlmock.NewRows([]string{"span_count", "service_count", "duration_ms", "has_error"}).
-			AddRow(int64(3), int64(2), 3.0, true))
 	mock.ExpectQuery(regexp.QuoteMeta(traceLogsQuery)).
 		WithArgs("wide-trace", start, end, "prod", "prod", 1).
 		WillReturnRows(sqlmock.NewRows([]string{"time", "severity", "service", "body", "trace_id", "span_id"}))
@@ -72,6 +67,11 @@ func TestTracePagedSpansStillDescribeTheWholeTrace(t *testing.T) {
 	if !result.Data.HasError {
 		t.Error("has_error = false, want true: the trace errors in a span the page omitted")
 	}
+	// Spans start at +0 (2ms), +1ms (1ms) and +2ms (1ms), so the trace runs 3ms
+	// end to end -- a figure the single returned span cannot produce on its own.
+	if result.Data.DurationMS != 3 {
+		t.Errorf("duration_ms = %v, want 3: the duration spans the trace, not the page", result.Data.DurationMS)
+	}
 	if strings.Contains(result.Summary, "1 spans") || strings.Contains(result.Summary, "1 services") {
 		t.Errorf("summary reports the page as the trace: %q", result.Summary)
 	}
@@ -96,10 +96,6 @@ func TestTraceWholeTraceIsNotReportedAsTruncated(t *testing.T) {
 	}}}); err != nil {
 		t.Fatal(err)
 	}
-	mock.ExpectQuery(regexp.QuoteMeta(traceSummaryQuery)).
-		WithArgs("small-trace", start, end, "prod", "prod").
-		WillReturnRows(sqlmock.NewRows([]string{"span_count", "service_count", "duration_ms", "has_error"}).
-			AddRow(int64(1), int64(1), 25.0, false))
 	mock.ExpectQuery(regexp.QuoteMeta(traceLogsQuery)).
 		WithArgs("small-trace", start, end, "prod", "prod", 10).
 		WillReturnRows(sqlmock.NewRows([]string{"time", "severity", "service", "body", "trace_id", "span_id"}))
@@ -119,63 +115,3 @@ func TestTraceWholeTraceIsNotReportedAsTruncated(t *testing.T) {
 		t.Fatal(err)
 	}
 }
-
-// The sqlmock tests above prove the plumbing but never execute the SQL, so the
-// aggregate's own arithmetic needs DuckDB to check it. start_time and end_time
-// are TIMESTAMP (internal/query/views.go:19-20); subtracting them yields an
-// INTERVAL, which is not a nanosecond count and cannot be scaled to
-// milliseconds by division. Only the BIGINT *_unix_nano columns can.
-func TestTraceSummaryQueryComputesDurationInMilliseconds(t *testing.T) {
-	db, err := sql.Open("duckdb", "")
-	if err != nil {
-		t.Fatalf("open duckdb: %v", err)
-	}
-	defer db.Close()
-	if _, err := db.Exec(`CREATE TABLE spans (
-		namespace VARCHAR, trace_id VARCHAR, service VARCHAR, status VARCHAR,
-		start_time TIMESTAMP, end_time TIMESTAMP,
-		start_unix_nano BIGINT, end_unix_nano BIGINT)`); err != nil {
-		t.Fatalf("create spans: %v", err)
-	}
-	base := time.Date(2026, 7, 20, 11, 0, 0, 0, time.UTC)
-	// The trace spans 3ms end-to-end: first span starts at base, last ends 3ms later.
-	rows := []struct {
-		service string
-		status  string
-		startMS int64
-		endMS   int64
-	}{
-		{"cart", "OK", 0, 3},
-		{"cart", "OK", 1, 2},
-		{"flagd", "ERROR", 2, 3},
-	}
-	for _, r := range rows {
-		start := base.Add(time.Duration(r.startMS) * time.Millisecond)
-		end := base.Add(time.Duration(r.endMS) * time.Millisecond)
-		if _, err := db.Exec(`INSERT INTO spans VALUES (?, ?, ?, ?, ?, ?, ?, ?)`,
-			"prod", "t1", r.service, r.status, start, end, start.UnixNano(), end.UnixNano()); err != nil {
-			t.Fatalf("insert span: %v", err)
-		}
-	}
-
-	var spanCount, serviceCount int64
-	var durationMS float64
-	var hasError bool
-	if err := db.QueryRow(traceSummaryQuery, "t1", base, base.Add(time.Hour), "prod", "prod").
-		Scan(&spanCount, &serviceCount, &durationMS, &hasError); err != nil {
-		t.Fatalf("trace summary query: %v", err)
-	}
-
-	if spanCount != 3 {
-		t.Errorf("span_count = %d, want 3", spanCount)
-	}
-	if serviceCount != 2 {
-		t.Errorf("service_count = %d, want 2", serviceCount)
-	}
-	if durationMS != 3 {
-		t.Errorf("duration_ms = %v, want 3", durationMS)
-	}
-	if !hasError {
-		t.Error("has_error = false, want true")
-	}
-}
diff --git a/internal/query/duck.go b/internal/query/duck.go
index b69e3396..aa23c8dd 100644
--- a/internal/query/duck.go
+++ b/internal/query/duck.go
@@ -35,7 +35,8 @@ type Duck struct {
 	closeErr  error
 	// releasePoolGauges stops the pool gauges reading this Duck's pool, and
 	// only if they are still reading it.
-	releasePoolGauges func()
+	releasePoolGauges   func()
+	releaseMemoryGauges func()
 	// writeDB is the single connection every rollup and maintenance write uses.
 	//
 	// Those writes are serialized by writeGate anyway, so one connection is all
@@ -264,6 +265,32 @@ func NewDuck(ctx context.Context, cfg config.Config, repository *telemetrystore.
 	d := &Duck{DB: db, writeDB: writeDB, cfg: cfg, repository: repository, rollupLagNanos: int64(rollupPublicationSafetyLag)}
 	// Reads are the pool that can starve, so it is the one worth watching.
 	d.releasePoolGauges = metrics.SetDuckDBPoolSource(db.Stats)
+	// And what DuckDB admits to holding, so the gap against process RSS is
+	// observable rather than inferred.
+	d.releaseMemoryGauges = metrics.SetDuckDBMemorySource(func() map[string]int64 {
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+		rows, err := db.QueryContext(ctx, "SELECT tag, sum(memory_usage_bytes) FROM duckdb_memory() GROUP BY tag")
+		if err != nil {
+			return nil
+		}
+		defer rows.Close()
+		tags := make(map[string]int64, 16)
+		for rows.Next() {
+			var tag string
+			var bytes int64
+			if err := rows.Scan(&tag, &bytes); err != nil {
+				return nil
+			}
+			tags[tag] = bytes
+		}
+		// A partial read is worse than no read: it publishes a low sum, and
+		// the untracked figure derived from it reads correspondingly high.
+		if err := rows.Err(); err != nil {
+			return nil
+		}
+		return tags
+	})
 	if cfg.DuckDBMemory == "" {
 		// Only when the operator hasn't pinned storage.duckdb.memory: keep DuckDB's
 		// cgroup-aware auto limit on big boxes but leave absolute RAM headroom on
@@ -430,6 +457,13 @@ func (d *Duck) Close() error {
 		if d.releasePoolGauges != nil {
 			d.releasePoolGauges()
 		}
+		// Same reason as the pool source: the closure captures this Duck's
+		// read pool, so leaving it installed past Close means the next scrape
+		// queries a closed pool and the gauges freeze at stale values that
+		// still look live.
+		if d.releaseMemoryGauges != nil {
+			d.releaseMemoryGauges()
+		}
 		var errs []error
 		if d.writeDB != nil {
 			// The write handle holds a connector that does not close the
@@ -1287,6 +1321,35 @@ SET last_ingested_unix_nano = excluded.last_ingested_unix_nano,
 	return err
 }
 
+// serviceRollupP50SQL and serviceRollupP95SQL build the latency columns of a
+// rollup pass.
+//
+// These are approximate on purpose. quantile_cont is holistic: it retains every
+// value of every group in a vector on the raw allocator, so its cost grows with
+// the rows a pass scans, and memory_limit does not bound it -- DuckDB reports
+// zero usage while the process grows by gigabytes. A pass covering an hour of
+// ingest at this product's certified rate scans tens of millions of spans, and
+// that is how the process reached 11.5 GiB anon-rss on a box configured for a
+// 4 GB limit.
+//
+// approx_quantile keeps a fixed-size t-digest per group instead. Measured on
+// 30M rows: 861 MiB of untracked growth becomes 8 MiB. The error is a fraction
+// of a percent on a latency percentile that already describes a one-minute
+// bucket, which is not a figure anyone reads to three significant digits.
+//
+// The COALESCE keeps a service that only makes outbound calls -- a load
+// generator, a cron worker -- measured on what it does have rather than
+// silently reported as having no latency at all.
+func serviceRollupP50SQL(alias string) string { return serviceRollupQuantileSQL(alias, 1, "p50_ms") }
+
+func serviceRollupP95SQL(alias string) string { return serviceRollupQuantileSQL(alias, 2, "p95_ms") }
+
+func serviceRollupQuantileSQL(alias string, index int, column string) string {
+	served := fmt.Sprintf("approx_quantile(%[1]s.duration_ms, [0.50, 0.95]) FILTER (WHERE COALESCE(%[1]s.kind, '') NOT IN ('SPAN_KIND_CLIENT', 'SPAN_KIND_PRODUCER'))", alias)
+	all := fmt.Sprintf("approx_quantile(%s.duration_ms, [0.50, 0.95])", alias)
+	return fmt.Sprintf("COALESCE(%s[%d], %s[%d]) AS %s", served, index, all, index, column)
+}
+
 // rollupSafetyLagNanos is how far behind the max ingested timestamp the rollup
 // watermark is held, covering the worst-case delay between a row being stamped at
 // ingest and committed to Parquet (the bounded commit retry window plus queueing).
@@ -1351,7 +1414,7 @@ WHERE EXISTS (
     AND affected.service = service_rollup.service
 );`
 
-const serviceRollupInsertSQL = `
+var serviceRollupInsertSQL = `
 WITH affected AS (
   SELECT DISTINCT namespace, date_trunc('minute', start_time) AS bucket, service
   FROM spans
@@ -1395,14 +1458,8 @@ span_agg AS (
     -- COALESCE keeps a service that only makes outbound calls — a load
     -- generator, a cron worker — measured on what it does have rather than
     -- silently reported as having no latency at all.
-    COALESCE(
-      quantile_cont(s.duration_ms, 0.50) FILTER (WHERE COALESCE(s.kind, '') NOT IN ('SPAN_KIND_CLIENT', 'SPAN_KIND_PRODUCER')),
-      quantile_cont(s.duration_ms, 0.50)
-    ) AS p50_ms,
-    COALESCE(
-      quantile_cont(s.duration_ms, 0.95) FILTER (WHERE COALESCE(s.kind, '') NOT IN ('SPAN_KIND_CLIENT', 'SPAN_KIND_PRODUCER')),
-      quantile_cont(s.duration_ms, 0.95)
-    ) AS p95_ms,
+    ` + serviceRollupP50SQL("s") + `,
+    ` + serviceRollupP95SQL("s") + `,
     avg(CASE WHEN s.status IN ('STATUS_CODE_ERROR', 'ERROR') THEN 1.0 ELSE 0.0 END) AS error_rate
   FROM spans s
   JOIN affected a
@@ -1594,6 +1651,36 @@ WITH affected AS (
     AND start_time >= ?
     AND start_time < ?
 ),
+-- The affected bucket range, computed once. As four scalar subqueries
+-- repeated across the predicates below, these read as correlated to the
+-- planner and cost a re-derivation each.
+bounds AS (
+  SELECT MIN(bucket) AS lo, MAX(bucket) AS hi FROM affected
+),
+-- The parent side is range-filtered HERE rather than in the join's ON
+-- clause, which keeps the hash build off every span ever ingested. The
+-- affected window is one minute of newly ingested spans; the table behind
+-- this view is the whole retention period, so the two differ by orders of
+-- magnitude and the ON-clause form pays for the difference on every pass.
+--
+-- Measured (TestEdgeRollupPlanUsesHashJoins, spans as a Parquet-backed view,
+-- 600k spans over 25 days, one affected minute): 14ms here against 33ms with
+-- the range predicate in the ON clause. Both plan a HASH_JOIN -- the cost is
+-- the size of the build side, not the join algorithm.
+--
+-- Parents more than 1h outside the affected bucket range are dropped, which
+-- is acceptable for minute-bucket dependency edges. Caveat retained from the
+-- original: buckets come from span start_time while the window bounds
+-- ingested_unix_nano, so one late-arriving span with an old start_time
+-- widens this range for the pass that ingests it.
+parent_scope AS (
+  SELECT parent.namespace, parent.span_id, parent.trace_id, parent.service
+  FROM spans parent, bounds
+  WHERE parent.start_time >= bounds.lo - INTERVAL 1 HOUR
+    AND parent.start_time <= bounds.hi + INTERVAL 1 HOUR
+    AND parent.service IS NOT NULL
+    AND parent.service != ''
+),
 call_edges AS (
   SELECT
     child.namespace,
@@ -1605,29 +1692,27 @@ call_edges AS (
     AVG(CASE WHEN child.status IN ('STATUS_CODE_ERROR', 'ERROR') THEN 1.0 ELSE 0.0 END) AS error_rate,
     'call' AS edge_type
   FROM spans child
-  JOIN spans parent
+  JOIN parent_scope parent
     ON child.parent_span_id = parent.span_id
    AND child.trace_id = parent.trace_id
    AND child.namespace = parent.namespace
-   -- Bound the parent side to the affected BUCKET RANGE ±1h: without this
-   -- the hash build covers every span ever ingested. Parents more than 1h
-   -- outside [MIN(affected.bucket), MAX(affected.bucket)] are dropped —
-   -- acceptable for minute-bucket dependency edges. Caveat: buckets come
-   -- from span start_time while the window bounds ingested_unix_nano, so
-   -- one late-arriving span with an old start_time widens the range (and
-   -- this scan) back to that bucket for the pass that ingests it.
-   AND parent.start_time >= (SELECT MIN(bucket) FROM affected) - INTERVAL 1 HOUR
-   AND parent.start_time <= (SELECT MAX(bucket) FROM affected) + INTERVAL 1 HOUR
   JOIN affected a
     ON a.namespace = child.namespace
    AND a.bucket = date_trunc('minute', child.start_time)
-  WHERE parent.service IS NOT NULL
-    AND parent.service != ''
-    AND child.service IS NOT NULL
+  -- bounds joins last, and as an explicit CROSS JOIN. Written as
+  -- "FROM spans child, bounds JOIN parent_scope ON child...", the comma binds
+  -- looser than JOIN: that parses as "spans child" comma "(bounds JOIN
+  -- parent_scope ON ...)", so DuckDB materialises a CROSS_PRODUCT of the child
+  -- scan with bounds instead of folding the single bounds row into the filters.
+  -- Same rows either way, so nothing fails and no row assertion notices; on the
+  -- fixture above it is 21ms against 14ms. TestEdgeRollupPlanUsesHashJoins
+  -- asserts the CROSS_PRODUCT is absent here and present in the comma form.
+  CROSS JOIN bounds
+  WHERE child.service IS NOT NULL
     AND child.service != ''
     AND parent.service != child.service
-    AND child.start_time >= (SELECT MIN(bucket) FROM affected)
-    AND child.start_time < (SELECT MAX(bucket) FROM affected) + INTERVAL 1 MINUTE
+    AND child.start_time >= bounds.lo
+    AND child.start_time < bounds.hi + INTERVAL 1 MINUTE
   GROUP BY child.namespace, date_trunc('minute', child.start_time), parent.service, child.service
 ),
 -- Producers and consumers are aggregated per (namespace, bucket, service,
@@ -1648,8 +1733,8 @@ producers AS (
     ON a.namespace = s.namespace
    AND a.bucket = date_trunc('minute', s.start_time)
   WHERE s.kind = 'SPAN_KIND_PRODUCER'
-    AND s.start_time >= (SELECT MIN(bucket) FROM affected)
-    AND s.start_time < (SELECT MAX(bucket) FROM affected) + INTERVAL 1 MINUTE
+    AND s.start_time >= (SELECT lo FROM bounds)
+    AND s.start_time < (SELECT hi FROM bounds) + INTERVAL 1 MINUTE
     AND s.service IS NOT NULL
     AND s.service != ''
     AND json_extract_string(s.attributes_json, '$."messaging.destination.name"') IS NOT NULL
@@ -1667,8 +1752,8 @@ consumers AS (
     ON a.namespace = s.namespace
    AND a.bucket = date_trunc('minute', s.start_time)
   WHERE s.kind = 'SPAN_KIND_CONSUMER'
-    AND s.start_time >= (SELECT MIN(bucket) FROM affected)
-    AND s.start_time < (SELECT MAX(bucket) FROM affected) + INTERVAL 1 MINUTE
+    AND s.start_time >= (SELECT lo FROM bounds)
+    AND s.start_time < (SELECT hi FROM bounds) + INTERVAL 1 MINUTE
     AND s.service IS NOT NULL
     AND s.service != ''
     AND json_extract_string(s.attributes_json, '$."messaging.destination.name"') IS NOT NULL
@@ -1800,9 +1885,9 @@ func (d *Duck) QueryRowScan(ctx context.Context, dest []any, query string, args
 }
 
 // Trace pins the immutable Parquet snapshot for the full indexed-file read.
-func (d *Duck) Trace(ctx context.Context, query telemetry.TraceQuery) ([]telemetry.IndexedSpan, error) {
+func (d *Duck) Trace(ctx context.Context, query telemetry.TraceQuery) ([]telemetry.IndexedSpan, telemetry.TraceTotals, error) {
 	if err := d.lockParquetRead(ctx, readerQuery); err != nil {
-		return nil, err
+		return nil, telemetry.TraceTotals{}, err
 	}
 	defer d.parquetMu.RUnlock()
 	return d.repository.Parquet.Trace(ctx, query)
diff --git a/internal/query/duck_test.go b/internal/query/duck_test.go
index 67a1521b..c4905776 100644
--- a/internal/query/duck_test.go
+++ b/internal/query/duck_test.go
@@ -428,7 +428,7 @@ func TestIndexedTraceReadHonorsParquetGateContext(t *testing.T) {
 	mustLock(&d.parquetMu)
 	ctx, cancel := context.WithTimeout(context.Background(), 25*time.Millisecond)
 	defer cancel()
-	_, err = d.Trace(ctx, telemetry.TraceQuery{TraceID: "trace", StartNanos: 1, EndNanos: 2, Limit: 1})
+	_, _, err = d.Trace(ctx, telemetry.TraceQuery{TraceID: "trace", StartNanos: 1, EndNanos: 2, Limit: 1})
 	d.parquetMu.Unlock()
 	if !errors.Is(err, context.DeadlineExceeded) {
 		t.Fatalf("Trace error = %v, want deadline exceeded", err)
diff --git a/internal/query/edge_rollup_plan_test.go b/internal/query/edge_rollup_plan_test.go
new file mode 100644
index 00000000..9db7f691
--- /dev/null
+++ b/internal/query/edge_rollup_plan_test.go
@@ -0,0 +1,224 @@
+package query
+
+import (
+	"database/sql"
+	"fmt"
+	"os"
+	"strings"
+	"testing"
+	"time"
+)
+
+// The edge rollup is the most expensive statement this process issues, and
+// until this test nothing executed it. Both of its shape defects were therefore
+// invisible: every form returns identical rows, so a row assertion passes on
+// all of them, and the only symptom was a pass that did not finish inside its
+// budget.
+//
+// The plan is the behaviour under test. This asserts it in both directions --
+// a guard that passes on the defect it was written for is worse than none --
+// and checks that the forms agree on rows, which is what makes the rewrite a
+// performance change rather than a behaviour one.
+func TestEdgeRollupPlanUsesHashJoins(t *testing.T) {
+	db := openTestDuck(t)
+	defer db.Close()
+	seedEdgeRollupFixture(t, db)
+
+	preFix, err := os.ReadFile("testdata/edge_rollup_pre_fix.sql")
+	if err != nil {
+		t.Fatal(err)
+	}
+	commaForm := edgeRollupCommaForm(t)
+
+	// Params: the ingested window is the fresh minute only; the start_time
+	// bounds are deliberately wide, as they are in production.
+	const windowLo, windowHi = 50, 200
+	spanLo := time.Date(2026, 8, 1, 0, 0, 0, 0, time.UTC)
+	spanHi := time.Date(2026, 10, 1, 0, 0, 0, 0, time.UTC)
+
+	shippedPlan := explain(t, db, edgeRollupInsertSQL, windowLo, windowHi, spanLo, spanHi)
+	commaPlan := explain(t, db, commaForm, windowLo, windowHi, spanLo, spanHi)
+
+	// The defect stated as an assertion. If DuckDB ever folds this away, the
+	// guard below stops proving anything and this says so rather than passing.
+	if !strings.Contains(commaPlan, "CROSS_PRODUCT") {
+		t.Fatalf("the comma form no longer plans a CROSS_PRODUCT, so this guard proves nothing now:\n%s", commaPlan)
+	}
+	for _, bad := range []string{"CROSS_PRODUCT", "DELIM_JOIN", "DELIM_GET", "NESTED_LOOP_JOIN"} {
+		if strings.Contains(shippedPlan, bad) {
+			t.Errorf("edge rollup plans a %s; the FROM clause has regressed to the comma form:\n%s", bad, shippedPlan)
+		}
+	}
+	if !strings.Contains(shippedPlan, "HASH_JOIN") {
+		t.Errorf("edge rollup plans no HASH_JOIN at all:\n%s", shippedPlan)
+	}
+
+	// All three forms must agree on rows. Wall-clock is logged rather than
+	// asserted: the ratios are consistent but the absolute numbers are not
+	// stable enough across machines to fail a build on.
+	var want []string
+	for _, form := range []struct{ name, sql string }{
+		{"shipped", edgeRollupInsertSQL},
+		{"comma", commaForm},
+		{"pre-fix (range predicate in the ON clause)", string(preFix)},
+	} {
+		start := time.Now()
+		edges := runEdgeRollup(t, db, form.sql, windowLo, windowHi, spanLo, spanHi)
+		t.Logf("%-44s %6.0fms  %d edges", form.name, float64(time.Since(start).Microseconds())/1000, len(edges))
+		if want == nil {
+			if len(edges) == 0 {
+				t.Fatal("the fixture produced no edges, so no plan was exercised")
+			}
+			want = edges
+			continue
+		}
+		if strings.Join(edges, "|") != strings.Join(want, "|") {
+			t.Errorf("%s disagrees on rows:\n got:  %v\n want: %v", form.name, edges, want)
+		}
+	}
+	// frontend -> api -> db from the hand-built trace, svc-4 -> svc-5 from the
+	// bulk rows, and frontend -> worker across the queue.
+	const expected = "api->db call 1|frontend->api call 1|frontend->worker messaging 1|svc-4->svc-5 call 1000"
+	if strings.Join(want, "|") != expected {
+		t.Errorf("edges = %v,\n want %s", want, expected)
+	}
+}
+
+// edgeRollupCommaForm rebuilds the pre-fix FROM clause from the shipped
+// statement, so the two differ in exactly the thing under test and the guard
+// cannot rot into comparing two stale copies.
+func edgeRollupCommaForm(t *testing.T) string {
+	t.Helper()
+	const joined = "  FROM spans child\n  JOIN parent_scope parent"
+	if !strings.Contains(edgeRollupInsertSQL, joined) {
+		t.Fatal("the edge rollup FROM clause no longer matches what this test rewrites")
+	}
+	form := strings.Replace(edgeRollupInsertSQL, joined, "  FROM spans child, bounds\n  JOIN parent_scope parent", 1)
+	if !strings.Contains(form, "\n  CROSS JOIN bounds\n") {
+		t.Fatal("no CROSS JOIN bounds to remove")
+	}
+	return strings.Replace(form, "\n  CROSS JOIN bounds\n", "\n", 1)
+}
+
+func explain(t *testing.T, db *sql.DB, query string, args ...any) string {
+	t.Helper()
+	rows, err := db.Query("EXPLAIN "+query, args...)
+	if err != nil {
+		t.Fatalf("EXPLAIN: %v", err)
+	}
+	defer rows.Close()
+	var plan strings.Builder
+	for rows.Next() {
+		var key, value string
+		if err := rows.Scan(&key, &value); err != nil {
+			t.Fatal(err)
+		}
+		plan.WriteString(value)
+	}
+	if err := rows.Err(); err != nil {
+		t.Fatal(err)
+	}
+	return plan.String()
+}
+
+// runEdgeRollup executes the statement into an empty edge_rollup and returns
+// the edges it wrote, so forms can be compared on rows as well as on plan.
+func runEdgeRollup(t *testing.T, db *sql.DB, query string, args ...any) []string {
+	t.Helper()
+	if _, err := db.Exec("DELETE FROM edge_rollup"); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := db.Exec(query, args...); err != nil {
+		t.Fatalf("edge rollup: %v", err)
+	}
+	rows, err := db.Query(`SELECT caller, callee, edge_type, calls FROM edge_rollup ORDER BY caller, callee, edge_type`)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer rows.Close()
+	var edges []string
+	for rows.Next() {
+		var caller, callee, edgeType string
+		var calls int64
+		if err := rows.Scan(&caller, &callee, &edgeType, &calls); err != nil {
+			t.Fatal(err)
+		}
+		edges = append(edges, fmt.Sprintf("%s->%s %s %d", caller, callee, edgeType, calls))
+	}
+	if err := rows.Err(); err != nil {
+		t.Fatal(err)
+	}
+	return edges
+}
+
+// seedEdgeRollupFixture builds the production shape: a long history of spans
+// that were rolled up long ago, and one freshly ingested minute. The two differ
+// by orders of magnitude, which is what makes where the parent side is filtered
+// matter. spans is a view over Parquet, as it is in production -- against a
+// DuckDB table the planner sees different statistics and the forms converge.
+func seedEdgeRollupFixture(t *testing.T, db *sql.DB) {
+	t.Helper()
+	if _, err := db.Exec(createEdgeRollupTable); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := db.Exec(`CREATE TABLE spans (
+		namespace TEXT, trace_id TEXT, span_id TEXT, parent_span_id TEXT,
+		service TEXT, kind TEXT, status TEXT, duration_ms DOUBLE,
+		start_time TIMESTAMP, ingested_unix_nano BIGINT, attributes_json TEXT
+	)`); err != nil {
+		t.Fatal(err)
+	}
+
+	// The smallest trace that exercises both halves: a two-hop call chain and a
+	// producer/consumer pair, all inside the freshly ingested minute.
+	fresh := time.Date(2026, 9, 20, 12, 0, 0, 0, time.UTC)
+	for _, s := range []struct{ id, parent, service, kind, attributes string }{
+		{id: "root", service: "frontend", kind: "SPAN_KIND_SERVER"},
+		{id: "a", parent: "root", service: "api", kind: "SPAN_KIND_SERVER"},
+		{id: "b", parent: "a", service: "db", kind: "SPAN_KIND_CLIENT"},
+		{id: "p", parent: "root", service: "frontend", kind: "SPAN_KIND_PRODUCER",
+			attributes: `{"messaging.destination.name":"jobs","messaging.system":"kafka"}`},
+		{id: "c", service: "worker", kind: "SPAN_KIND_CONSUMER",
+			attributes: `{"messaging.destination.name":"jobs","messaging.system":"kafka"}`},
+	} {
+		attributes := s.attributes
+		if attributes == "" {
+			attributes = "{}"
+		}
+		if _, err := db.Exec(`INSERT INTO spans VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
+			"default", "trace-1", s.id, s.parent, s.service, s.kind,
+			"STATUS_CODE_OK", 5.0, fresh, int64(100), attributes); err != nil {
+			t.Fatal(err)
+		}
+	}
+	// 2k more spans in that same minute, so the fresh window is not trivially
+	// small relative to the joins.
+	if _, err := db.Exec(`INSERT INTO spans
+		SELECT 'default', 'new-' || (i // 2)::VARCHAR, 'new-' || i::VARCHAR,
+		       CASE WHEN i % 2 = 1 THEN 'new-' || (i - 1)::VARCHAR ELSE '' END,
+		       'svc-' || (i % 2 + 4)::VARCHAR, 'SPAN_KIND_SERVER', 'STATUS_CODE_OK', 5.0,
+		       TIMESTAMP '2026-09-20 12:00:00', 100, '{}'
+		FROM range(2000) t(i)`); err != nil {
+		t.Fatal(err)
+	}
+	// 600k spans across 25 days, ingested long before the window under test.
+	if _, err := db.Exec(`INSERT INTO spans
+		SELECT 'default', 'old-' || (i // 2)::VARCHAR, 'old-' || i::VARCHAR,
+		       CASE WHEN i % 2 = 1 THEN 'old-' || (i - 1)::VARCHAR ELSE '' END,
+		       'svc-' || (i % 8)::VARCHAR, 'SPAN_KIND_SERVER', 'STATUS_CODE_OK', 5.0,
+		       TIMESTAMP '2026-08-26 00:00:00' + INTERVAL (i % 36000) MINUTE, 1, '{}'
+		FROM range(600000) t(i)`); err != nil {
+		t.Fatal(err)
+	}
+
+	dir := t.TempDir()
+	if _, err := db.Exec(`COPY (SELECT * FROM spans) TO '` + dir + `/spans.parquet' (FORMAT PARQUET)`); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := db.Exec(`DROP TABLE spans`); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := db.Exec(`CREATE VIEW spans AS SELECT * FROM read_parquet('` + dir + `/spans.parquet')`); err != nil {
+		t.Fatal(err)
+	}
+}
diff --git a/internal/query/rollup_memory_test.go b/internal/query/rollup_memory_test.go
new file mode 100644
index 00000000..6cc91474
--- /dev/null
+++ b/internal/query/rollup_memory_test.go
@@ -0,0 +1,110 @@
+//go:build !race
+
+package query
+
+import (
+	"database/sql"
+	"fmt"
+	"os"
+	"os/exec"
+	"strconv"
+	"strings"
+	"testing"
+)
+
+// rollupUntrackedLimitMiB is how far past DuckDB's configured memory_limit one
+// rollup pass may allocate.
+//
+// memory_limit bounds the buffer pool and nothing else. Holistic aggregates --
+// quantile_cont, PERCENTILE_CONT -- keep every value of every group in a vector
+// on the raw allocator, so they allocate proportionally to rows scanned with no
+// ceiling and no pressure signal: DuckDB reports zero usage while the kernel
+// sees gigabytes. That is what OOM-killed this process, and the configured
+// limit could never have prevented it.
+//
+// Measured on 30M rows at memory_limit=100MB: the production expression shape
+// grew RSS by 861 MiB with duckdb_memory() reporting 0.0 MiB, while the
+// approximate form grew it by 8 MiB. The limit below sits far below the former
+// and far above the latter.
+const rollupUntrackedLimitMiB = 200
+
+// processRSSMiB reads this process's resident size, or skips the test where it
+// cannot: a busybox ps without -o/-p would otherwise fail the whole suite for a
+// reason unrelated to the regression this guards.
+func processRSSMiB(t *testing.T) float64 {
+	t.Helper()
+	if raw, err := os.ReadFile("/proc/self/statm"); err == nil {
+		fields := strings.Fields(string(raw))
+		if len(fields) > 1 {
+			if pages, err := strconv.ParseFloat(fields[1], 64); err == nil {
+				return pages * float64(os.Getpagesize()) / (1 << 20)
+			}
+		}
+	}
+	out, err := exec.Command("ps", "-o", "rss=", "-p", strconv.Itoa(os.Getpid())).Output()
+	if err != nil {
+		t.Skipf("no way to read RSS on this platform: %v", err)
+	}
+	kb, err := strconv.ParseFloat(strings.TrimSpace(string(out)), 64)
+	if err != nil {
+		t.Skipf("unreadable ps output %q: %v", out, err)
+	}
+	return kb / 1024
+}
+
+// A rollup pass scans an hour of ingest with no row bound, so its latency
+// aggregation must not allocate in proportion to the rows it reads.
+func TestServiceRollupLatencyDoesNotAllocatePerRow(t *testing.T) {
+	db, err := sql.Open("duckdb", "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.Close()
+	for _, stmt := range []string{"SET memory_limit='100MB'", "SET threads=4"} {
+		if _, err := db.Exec(stmt); err != nil {
+			t.Fatalf("%s: %v", stmt, err)
+		}
+	}
+
+	if testing.Short() {
+		t.Skip("scans 30M rows")
+	}
+	baseline := processRSSMiB(t)
+	query := fmt.Sprintf(`SELECT service, %s, %s
+		FROM (
+			SELECT (i %% 20)::VARCHAR AS service,
+			       (i %% 1000)::DOUBLE AS duration_ms,
+			       CASE WHEN i %% 7 = 0 THEN 'SPAN_KIND_CLIENT' ELSE 'SPAN_KIND_SERVER' END AS kind
+			FROM range(30000000) t(i)
+		) s
+		GROUP BY service`, serviceRollupP50SQL("s"), serviceRollupP95SQL("s"))
+
+	rows, err := db.Query(query)
+	if err != nil {
+		t.Fatalf("rollup latency query: %v", err)
+	}
+	groups := 0
+	for rows.Next() {
+		groups++
+	}
+	if err := rows.Err(); err != nil {
+		rows.Close()
+		t.Fatal(err)
+	}
+	rows.Close()
+
+	grew := processRSSMiB(t) - baseline
+	var tracked float64
+	if err := db.QueryRow("SELECT COALESCE(sum(memory_usage_bytes), 0) / 1048576.0 FROM duckdb_memory()").Scan(&tracked); err != nil {
+		t.Fatalf("duckdb_memory: %v", err)
+	}
+	t.Logf("groups=%d  RSS +%.0f MiB over a 100 MB memory_limit  duckdb_memory reports %.1f MiB", groups, grew, tracked)
+
+	if groups != 20 {
+		t.Fatalf("groups = %d, want 20: the fixture is wrong", groups)
+	}
+	if grew > rollupUntrackedLimitMiB {
+		t.Errorf("one rollup pass grew RSS by %.0f MiB against a 100 MB limit (DuckDB reported %.1f MiB): the latency aggregation allocates per row and memory_limit cannot see it",
+			grew, tracked)
+	}
+}
diff --git a/internal/query/rowmap_float_test.go b/internal/query/rowmap_float_test.go
new file mode 100644
index 00000000..5111f0ca
--- /dev/null
+++ b/internal/query/rowmap_float_test.go
@@ -0,0 +1,47 @@
+package query
+
+import (
+	"context"
+	"testing"
+)
+
+// DuckDB's FLOAT is single precision, so the driver hands back a float32 and
+// every `row["x"].(float64)` on it fails -- yielding 0.0 from a value that was
+// never zero. Callers write that assertion because every other numeric column
+// they touch (DOUBLE, AVG, SUM) really is float64, so the one FLOAT column in a
+// query fails silently while the rest of the row is fine.
+//
+// That is not hypothetical: internal/intelligence cast three columns to FLOAT
+// and reported 0 for all of them for as long as the code existed -- volume
+// anomalies showed "0 spans" for services that were serving thousands, and the
+// error-rate detector's current and baseline rates were always 0.
+//
+// Fixing the casts alone fixes today's queries and leaves the trap armed for
+// the next one, so the conversion is what changes here.
+func TestRowMapWidensFloat32(t *testing.T) {
+	db := openTestDuck(t)
+	defer db.Close()
+	d := &Duck{DB: db}
+
+	resp := d.ExecuteSQL(context.Background(), SQLRequest{
+		Query: "SELECT 327::FLOAT AS as_float, 0.25::FLOAT AS rate, 327::DOUBLE AS as_double",
+	})
+	if resp.Error != "" {
+		t.Fatalf("ExecuteSQL: %s", resp.Error)
+	}
+	if len(resp.Results) != 1 {
+		t.Fatalf("got %d rows, want 1", len(resp.Results))
+	}
+	row := resp.Results[0]
+
+	for name, want := range map[string]float64{"as_float": 327, "rate": 0.25, "as_double": 327} {
+		got, ok := row[name].(float64)
+		if !ok {
+			t.Errorf("row[%q] is %T, not float64: every caller asserting float64 reads 0 from it", name, row[name])
+			continue
+		}
+		if got != want {
+			t.Errorf("row[%q] = %v, want %v", name, got, want)
+		}
+	}
+}
diff --git a/internal/query/sql.go b/internal/query/sql.go
index 6776ed32..e2504129 100644
--- a/internal/query/sql.go
+++ b/internal/query/sql.go
@@ -148,10 +148,18 @@ func (d *Duck) ExecuteSQL(ctx context.Context, req SQLRequest) (resp SQLResponse
 			row := make(RowMap)
 			for i, col := range columns {
 				val := values[i]
-				// Convert []uint8 to string for better JSON representation
-				if b, ok := val.([]byte); ok {
-					row[col] = string(b)
-				} else {
+				switch v := val.(type) {
+				case []byte:
+					// Convert []uint8 to string for better JSON representation
+					row[col] = string(v)
+				case float32:
+					// DuckDB's FLOAT is single precision, so the driver returns
+					// float32. Callers assert float64 -- every other numeric
+					// column they meet is one -- and a failed assertion yields
+					// 0.0 from a value that was never zero, silently. Widening
+					// here costs nothing and disarms the whole class.
+					row[col] = float64(v)
+				default:
 					row[col] = val
 				}
 			}
diff --git a/internal/query/testdata/edge_rollup_pre_fix.sql b/internal/query/testdata/edge_rollup_pre_fix.sql
new file mode 100644
index 00000000..60497482
--- /dev/null
+++ b/internal/query/testdata/edge_rollup_pre_fix.sql
@@ -0,0 +1,126 @@
+
+WITH affected AS (
+  SELECT DISTINCT namespace, date_trunc('minute', start_time) AS bucket
+  FROM spans
+  WHERE ingested_unix_nano > ?
+    AND ingested_unix_nano <= ?
+    AND start_time IS NOT NULL
+    AND start_time >= ?
+    AND start_time < ?
+),
+call_edges AS (
+  SELECT
+    child.namespace,
+    date_trunc('minute', child.start_time) AS bucket,
+    parent.service AS caller,
+    child.service AS callee,
+    COUNT(*) AS calls,
+    AVG(child.duration_ms) AS avg_ms,
+    AVG(CASE WHEN child.status IN ('STATUS_CODE_ERROR', 'ERROR') THEN 1.0 ELSE 0.0 END) AS error_rate,
+    'call' AS edge_type
+  FROM spans child
+  JOIN spans parent
+    ON child.parent_span_id = parent.span_id
+   AND child.trace_id = parent.trace_id
+   AND child.namespace = parent.namespace
+   -- Bound the parent side to the affected BUCKET RANGE ±1h: without this
+   -- the hash build covers every span ever ingested. Parents more than 1h
+   -- outside [MIN(affected.bucket), MAX(affected.bucket)] are dropped —
+   -- acceptable for minute-bucket dependency edges. Caveat: buckets come
+   -- from span start_time while the window bounds ingested_unix_nano, so
+   -- one late-arriving span with an old start_time widens the range (and
+   -- this scan) back to that bucket for the pass that ingests it.
+   AND parent.start_time >= (SELECT MIN(bucket) FROM affected) - INTERVAL 1 HOUR
+   AND parent.start_time <= (SELECT MAX(bucket) FROM affected) + INTERVAL 1 HOUR
+  JOIN affected a
+    ON a.namespace = child.namespace
+   AND a.bucket = date_trunc('minute', child.start_time)
+  WHERE parent.service IS NOT NULL
+    AND parent.service != ''
+    AND child.service IS NOT NULL
+    AND child.service != ''
+    AND parent.service != child.service
+    AND child.start_time >= (SELECT MIN(bucket) FROM affected)
+    AND child.start_time < (SELECT MAX(bucket) FROM affected) + INTERVAL 1 MINUTE
+  GROUP BY child.namespace, date_trunc('minute', child.start_time), parent.service, child.service
+),
+-- Producers and consumers are aggregated per (namespace, bucket, service,
+-- destination, msg_system) BEFORE the join — joining raw span rows multiplies
+-- producer spans by consumer spans per destination, quadratic in per-bucket
+-- span count. calls = consumed messages on the destination, attributed to
+-- each producer service publishing to it (a message consumed once appears on
+-- every producer's edge).
+producers AS (
+  SELECT DISTINCT
+    s.namespace,
+    date_trunc('minute', s.start_time) AS bucket,
+    s.service,
+    json_extract_string(s.attributes_json, '$."messaging.destination.name"') AS destination,
+    json_extract_string(s.attributes_json, '$."messaging.system"') AS msg_system
+  FROM spans s
+  JOIN affected a
+    ON a.namespace = s.namespace
+   AND a.bucket = date_trunc('minute', s.start_time)
+  WHERE s.kind = 'SPAN_KIND_PRODUCER'
+    AND s.start_time >= (SELECT MIN(bucket) FROM affected)
+    AND s.start_time < (SELECT MAX(bucket) FROM affected) + INTERVAL 1 MINUTE
+    AND s.service IS NOT NULL
+    AND s.service != ''
+    AND json_extract_string(s.attributes_json, '$."messaging.destination.name"') IS NOT NULL
+),
+consumers AS (
+  SELECT
+    s.namespace,
+    date_trunc('minute', s.start_time) AS bucket,
+    s.service,
+    json_extract_string(s.attributes_json, '$."messaging.destination.name"') AS destination,
+    json_extract_string(s.attributes_json, '$."messaging.system"') AS msg_system,
+    COUNT(*) AS calls
+  FROM spans s
+  JOIN affected a
+    ON a.namespace = s.namespace
+   AND a.bucket = date_trunc('minute', s.start_time)
+  WHERE s.kind = 'SPAN_KIND_CONSUMER'
+    AND s.start_time >= (SELECT MIN(bucket) FROM affected)
+    AND s.start_time < (SELECT MAX(bucket) FROM affected) + INTERVAL 1 MINUTE
+    AND s.service IS NOT NULL
+    AND s.service != ''
+    AND json_extract_string(s.attributes_json, '$."messaging.destination.name"') IS NOT NULL
+  GROUP BY s.namespace, date_trunc('minute', s.start_time), s.service,
+    json_extract_string(s.attributes_json, '$."messaging.destination.name"'),
+    json_extract_string(s.attributes_json, '$."messaging.system"')
+),
+messaging_edges AS (
+  SELECT
+    p.namespace,
+    p.bucket,
+    p.service AS caller,
+    c.service AS callee,
+    SUM(c.calls) AS calls,
+    0.0 AS avg_ms,
+    0.0 AS error_rate,
+    'messaging' AS edge_type
+  FROM producers p
+  JOIN consumers c
+    ON c.namespace = p.namespace
+   AND c.bucket = p.bucket
+   AND c.destination = p.destination
+   AND c.msg_system = p.msg_system
+  WHERE p.service != c.service
+  GROUP BY p.namespace, p.bucket, p.service, c.service
+)
+INSERT INTO edge_rollup (
+  namespace,
+  bucket,
+  caller,
+  callee,
+  calls,
+  avg_ms,
+  error_rate,
+  edge_type
+)
+SELECT namespace, bucket, caller, callee, calls, avg_ms, error_rate, edge_type
+FROM call_edges
+UNION ALL
+SELECT namespace, bucket, caller, callee, calls, avg_ms, error_rate, edge_type
+FROM messaging_edges;
\ No newline at end of file
diff --git a/internal/query/views.go b/internal/query/views.go
index 5583f68d..c79ea40b 100644
--- a/internal/query/views.go
+++ b/internal/query/views.go
@@ -295,17 +295,39 @@ func CreateCacheTables(db *sql.DB) error {
 
 // CreateParquetViews exposes the repository's open Parquet files under the
 // canonical telemetry schema used by Fanout's SQL kernel.
+//
+// The schema is pinned explicitly rather than unioned from the files. Both
+// approaches tolerate an older batch that predates an added column -- it reads
+// as NULL either way, which is the property the views cannot do without, since
+// CreateViews names every column and binds eagerly, so a glob that rejects one
+// old file stops the process from starting.
+//
+// They differ entirely in cost. union_by_name derives the column set by opening
+// every file in the glob at bind and holding a reader and footer per file for
+// the statement's life, on the raw allocator where memory_limit cannot see it:
+// measured at 510 MiB per query against 3,281 live batches and 857 MiB against
+// 5,709, scaling at roughly 150 KB per file and paid again by every concurrent
+// binder. Pinning the schema opens nothing at bind, so the same query costs
+// tens of MiB and stops tracking how many batches happen to exist.
 func CreateParquetViews(db *sql.DB, parquetDir string) error {
 	if _, err := db.Exec(`CREATE SCHEMA IF NOT EXISTS telemetry`); err != nil {
 		return err
 	}
 	for _, signal := range []string{"spans", "logs", "metrics"} {
 		pattern := filepath.ToSlash(filepath.Join(parquetDir, "batches", "*.batch", signal+".parquet"))
-		projection := "*"
-		if signal == "spans" {
-			projection = "* EXCLUDE (_trace_hash)"
+		// _schema.batch is rewritten from the current binary's row structs on
+		// every open (ParquetStore.ensureSchemaBatch), so it is the definition
+		// of "every column this build knows about". Reading its schema keeps
+		// the view in lockstep with the writer instead of duplicating the
+		// column list here for someone to forget. That rewrite is load-bearing:
+		// while the file was written once and kept, these views were pinned to
+		// whichever build created the data directory.
+		schemaFile := filepath.ToSlash(filepath.Join(parquetDir, "batches", "_schema.batch", signal+".parquet"))
+		columns, err := parquetSchemaMap(db, schemaFile, signal == "spans")
+		if err != nil {
+			return err
 		}
-		stmt := fmt.Sprintf(`CREATE OR REPLACE VIEW telemetry.%s AS SELECT %s FROM read_parquet(%s, union_by_name=true)`, signal, projection, sqlLiteral(pattern))
+		stmt := fmt.Sprintf(`CREATE OR REPLACE VIEW telemetry.%s AS SELECT * FROM read_parquet(%s, schema=%s)`, signal, sqlLiteral(pattern), columns)
 		if _, err := db.Exec(stmt); err != nil {
 			return fmt.Errorf("create parquet view telemetry.%s: %w", signal, err)
 		}
@@ -313,6 +335,37 @@ func CreateParquetViews(db *sql.DB, parquetDir string) error {
 	return nil
 }
 
+// parquetSchemaMap renders the schema argument for read_parquet from one file's
+// own schema. A column absent from a given file takes its default, which is how
+// an older batch survives a build that added a column.
+func parquetSchemaMap(db *sql.DB, file string, dropTraceHash bool) (string, error) {
+	rows, err := db.Query(fmt.Sprintf(`SELECT column_name, column_type FROM (DESCRIBE SELECT * FROM read_parquet(%s))`, sqlLiteral(file)))
+	if err != nil {
+		return "", fmt.Errorf("describe %s: %w", file, err)
+	}
+	defer rows.Close()
+	var entries []string
+	for rows.Next() {
+		var name, columnType string
+		if err := rows.Scan(&name, &columnType); err != nil {
+			return "", err
+		}
+		// _trace_hash is a physical sort key, not part of the telemetry schema.
+		if dropTraceHash && name == "_trace_hash" {
+			continue
+		}
+		entries = append(entries, fmt.Sprintf("%s: {'name': %s, 'type': %s, 'default_value': NULL}",
+			sqlLiteral(name), sqlLiteral(name), sqlLiteral(columnType)))
+	}
+	if err := rows.Err(); err != nil {
+		return "", err
+	}
+	if len(entries) == 0 {
+		return "", fmt.Errorf("no columns described for %s", file)
+	}
+	return "MAP{" + strings.Join(entries, ", ") + "}", nil
+}
+
 // CreateViews creates stable clean-name views plus the attr() macro.
 func CreateViews(db *sql.DB) error {
 	for _, stmt := range []string{macroAttr, viewSpans, viewLogs, viewMetrics} {
diff --git a/internal/query/views_schema_test.go b/internal/query/views_schema_test.go
new file mode 100644
index 00000000..e1dc874e
--- /dev/null
+++ b/internal/query/views_schema_test.go
@@ -0,0 +1,78 @@
+package query
+
+import (
+	"database/sql"
+	"fmt"
+	"os"
+	"path/filepath"
+	"testing"
+)
+
+// The telemetry views must survive a build that adds a column: older batches
+// on disk lack it and must read as NULL rather than failing the glob.
+//
+// This is not a theoretical concern. CreateViews names every column explicitly
+// and DuckDB binds a view eagerly, so a glob that rejects one old file fails at
+// NewDuck and the process does not start at all -- it stays down until every
+// pre-deploy batch has aged out. An earlier attempt to cut the views' bind cost
+// by dropping union_by_name did exactly that, which is why this test exists
+// rather than a comment saying to be careful.
+func TestParquetViewsToleratePreDeploySchemas(t *testing.T) {
+	dir := t.TempDir()
+	batches := filepath.Join(dir, "batches")
+	for _, name := range []string{"_schema.batch", "old.batch", "new.batch"} {
+		if err := os.MkdirAll(filepath.Join(batches, name), 0o755); err != nil {
+			t.Fatal(err)
+		}
+	}
+	db, err := sql.Open("duckdb", "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.Close()
+
+	write := func(batch, signal, sel string) {
+		t.Helper()
+		path := filepath.ToSlash(filepath.Join(batches, batch, signal+".parquet"))
+		if _, err := db.Exec(fmt.Sprintf(`COPY (%s) TO '%s' (FORMAT PARQUET)`, sel, path)); err != nil {
+			t.Fatalf("write %s/%s: %v", batch, signal, err)
+		}
+	}
+	// The current build knows about added_later; the old batch predates it.
+	for _, signal := range []string{"spans", "logs", "metrics"} {
+		write("_schema.batch", signal, "SELECT 1 AS id, 'x' AS added_later WHERE false")
+		write("old.batch", signal, "SELECT 1 AS id")
+		write("new.batch", signal, "SELECT 2 AS id, 'present' AS added_later")
+	}
+
+	if err := CreateParquetViews(db, dir); err != nil {
+		t.Fatalf("CreateParquetViews must not fail on a pre-deploy batch: %v", err)
+	}
+
+	for _, signal := range []string{"spans", "logs", "metrics"} {
+		rows, err := db.Query(fmt.Sprintf(`SELECT id, added_later FROM telemetry.%s ORDER BY id`, signal))
+		if err != nil {
+			t.Fatalf("query telemetry.%s: %v", signal, err)
+		}
+		seen := map[int]bool{}
+		for rows.Next() {
+			var id int
+			var added sql.NullString
+			if err := rows.Scan(&id, &added); err != nil {
+				rows.Close()
+				t.Fatal(err)
+			}
+			seen[id] = true
+			if id == 1 && added.Valid {
+				t.Errorf("%s: the pre-deploy batch reported %q for a column it does not have", signal, added.String)
+			}
+			if id == 2 && added.String != "present" {
+				t.Errorf("%s: current batch read added_later=%q, want \"present\"", signal, added.String)
+			}
+		}
+		rows.Close()
+		if !seen[1] || !seen[2] {
+			t.Errorf("%s: read %v, want rows from both the old and the new batch", signal, seen)
+		}
+	}
+}
diff --git a/internal/telemetry/parquet.go b/internal/telemetry/parquet.go
index 365e8c3c..03970629 100644
--- a/internal/telemetry/parquet.go
+++ b/internal/telemetry/parquet.go
@@ -439,18 +439,35 @@ func (p *ParquetStore) RestoreRetiredInputs(inputs []string, replacementID strin
 // Trace reads only ranges selected by the persistent hash index. Scope filters
 // and the limit are applied while decoding so a pathological trace cannot grow
 // request memory without bound.
-func (p *ParquetStore) Trace(ctx context.Context, query TraceQuery) ([]IndexedSpan, error) {
+// TraceTotals describes the whole trace, not the page a limit admitted.
+//
+// These come free: readIndexedTrace already decodes every row of the trace and
+// applies the same predicates a separate aggregate would, and only the heap is
+// bounded by Limit. Deriving them here removes a second pass over every batch
+// file -- the thing that made trace_detail pay for its own correctness.
+type TraceTotals struct {
+	Spans        int
+	Services     int
+	DurationMS   float64
+	HasError     bool
+	MinStartNano int64
+	MaxEndNano   int64
+}
+
+func (p *ParquetStore) Trace(ctx context.Context, query TraceQuery) ([]IndexedSpan, TraceTotals, error) {
+	var totals TraceTotals
+	services := make(map[string]struct{}, 8)
 	if query.TraceID == "" {
-		return nil, nil
+		return nil, totals, nil
 	}
 	if query.Limit <= 0 {
-		return nil, errors.New("trace query limit must be positive")
+		return nil, totals, errors.New("trace query limit must be positive")
 	}
 	if query.Limit > maxTraceQueryResults {
-		return nil, fmt.Errorf("trace query limit exceeds %d", maxTraceQueryResults)
+		return nil, totals, fmt.Errorf("trace query limit exceeds %d", maxTraceQueryResults)
 	}
 	if query.StartNanos >= query.EndNanos {
-		return nil, errors.New("trace query time range must be positive")
+		return nil, totals, errors.New("trace query time range must be positive")
 	}
 	hash := xxh3.HashString(query.TraceID)
 	p.mu.RLock()
@@ -462,7 +479,7 @@ func (p *ParquetStore) Trace(ctx context.Context, query TraceQuery) ([]IndexedSp
 	selected := make(indexedSpanHeap, 0, query.Limit)
 	for _, batch := range batches {
 		if err := ctx.Err(); err != nil {
-			return nil, err
+			return nil, totals, err
 		}
 		if batch.metadata.MaxSpanStartNanos > 0 &&
 			(batch.metadata.MaxSpanStartNanos < query.StartNanos || batch.metadata.MinSpanStartNanos >= query.EndNanos) {
@@ -470,23 +487,27 @@ func (p *ParquetStore) Trace(ctx context.Context, query TraceQuery) ([]IndexedSp
 		}
 		match, found, err := batch.traces.Lookup(hash)
 		if err != nil {
-			return nil, err
+			return nil, totals, err
 		}
 		if !found {
 			continue
 		}
-		if err := readIndexedTrace(ctx, batch, match, query, &selected); err != nil {
-			return nil, err
+		if err := readIndexedTrace(ctx, batch, match, query, &selected, &totals, services); err != nil {
+			return nil, totals, err
 		}
 	}
+	totals.Services = len(services)
+	if totals.Spans > 0 && totals.MaxEndNano > totals.MinStartNano {
+		totals.DurationMS = float64(totals.MaxEndNano-totals.MinStartNano) / float64(time.Millisecond)
+	}
 	out := []IndexedSpan(selected)
 	sort.Slice(out, func(i, j int) bool {
 		return indexedSpanEarlier(out[i], out[j])
 	})
-	return out, nil
+	return out, totals, nil
 }
 
-func readIndexedTrace(ctx context.Context, batch *storedBatch, match traceRange, query TraceQuery, selected *indexedSpanHeap) (err error) {
+func readIndexedTrace(ctx context.Context, batch *storedBatch, match traceRange, query TraceQuery, selected *indexedSpanHeap, totals *TraceTotals, services map[string]struct{}) (err error) {
 	if match.row > uint64(math.MaxInt64) {
 		return errors.New("trace index row exceeds Parquet reader limit")
 	}
@@ -522,6 +543,21 @@ func readIndexedTrace(ctx context.Context, batch *storedBatch, match traceRange,
 				(query.Namespace != "" && row.Namespace != query.Namespace) {
 				continue
 			}
+			// Counted before the heap, so the totals describe the trace while
+			// the heap keeps describing the page.
+			totals.Spans++
+			if row.Service != "" {
+				services[row.Service] = struct{}{}
+			}
+			if strings.Contains(strings.ToUpper(row.Status), "ERROR") {
+				totals.HasError = true
+			}
+			if totals.MinStartNano == 0 || row.StartUnixNano < totals.MinStartNano {
+				totals.MinStartNano = row.StartUnixNano
+			}
+			if end := row.StartUnixNano + int64(row.DurationMS*float64(time.Millisecond)); end > totals.MaxEndNano {
+				totals.MaxEndNano = end
+			}
 			selected.Add(row.span(), query.Limit)
 		}
 		remaining -= uint64(n)
@@ -902,14 +938,24 @@ func (p *ParquetStore) PublishReplacement(stage string, metadata BatchMetadata,
 	return errors.Join(removeErr, syncDirectory(p.batchesDir))
 }
 
+// ensureSchemaBatch writes _schema.batch: an empty Parquet batch whose only
+// purpose is to carry this binary's column list. internal/query derives every
+// view's explicit read_parquet(schema=MAP{...}) from it, so a column absent
+// here is a column absent from the view.
+//
+// It is rewritten on every open rather than created once. Written once, it
+// pinned the views to whichever build first created the data directory, for
+// the life of that deployment -- and a later build that added a column to
+// spanParquetRow could not start at all: the view would not expose it,
+// CreateViews would fail to bind it, NewDuck would return an error, and
+// restarting would not help because the stale file was still there. The files
+// are empty, so this is three small writes per boot.
 func (p *ParquetStore) ensureSchemaBatch() error {
 	final := filepath.Join(p.batchesDir, SchemaBatch)
-	if info, err := os.Stat(final); err == nil && info.IsDir() {
-		return nil
-	} else if err != nil && !errors.Is(err, os.ErrNotExist) {
+	stage := filepath.Join(p.stagingDir, "_schema")
+	if err := os.RemoveAll(stage); err != nil {
 		return err
 	}
-	stage := filepath.Join(p.stagingDir, "_schema")
 	if err := os.Mkdir(stage, 0o755); err != nil {
 		return err
 	}
@@ -925,6 +971,12 @@ func (p *ParquetStore) ensureSchemaBatch() error {
 	if err := syncDirectory(stage); err != nil {
 		return err
 	}
+	// Rename cannot replace a non-empty directory, so the old one is removed
+	// first. A crash between the two leaves no _schema.batch at all, which the
+	// next open recreates: the batch holds no rows, only the column list.
+	if err := os.RemoveAll(final); err != nil {
+		return err
+	}
 	if err := os.Rename(stage, final); err != nil {
 		return err
 	}
diff --git a/internal/telemetry/parquet_test.go b/internal/telemetry/parquet_test.go
index 17e4d80c..bdd07fc2 100644
--- a/internal/telemetry/parquet_test.go
+++ b/internal/telemetry/parquet_test.go
@@ -163,7 +163,7 @@ func TestParquetStoreTraceFiltersAndBoundsResultsDuringRead(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	got, err := store.Trace(context.Background(), TraceQuery{
+	got, _, err := store.Trace(context.Background(), TraceQuery{
 		TraceID: "large-trace", Namespace: "prod", StartNanos: 10, EndNanos: 90, Limit: 3,
 	})
 	if err != nil {
@@ -226,7 +226,7 @@ func TestParquetStoreSkipsTraceIndexesOutsideTimeWindow(t *testing.T) {
 	if err := os.Remove(filepath.Join(store.BatchPath("old-traces"), "trace.fidx")); err != nil {
 		t.Fatal(err)
 	}
-	got, err := store.Trace(context.Background(), TraceQuery{
+	got, _, err := store.Trace(context.Background(), TraceQuery{
 		TraceID: "wanted", StartNanos: 1_000, EndNanos: 2_000, Limit: 10,
 	})
 	if err != nil {
@@ -502,9 +502,10 @@ func completeTestSpan() Span {
 }
 
 func traceAll(store *ParquetStore, traceID string) ([]IndexedSpan, error) {
-	return store.Trace(context.Background(), TraceQuery{
+	spans, _, err := store.Trace(context.Background(), TraceQuery{
 		TraceID: traceID, StartNanos: math.MinInt64, EndNanos: math.MaxInt64, Limit: maxTraceQueryResults,
 	})
+	return spans, err
 }
 
 func readOneParquetRow[T any](t *testing.T, path string) T {
diff --git a/internal/telemetry/schema_batch_test.go b/internal/telemetry/schema_batch_test.go
new file mode 100644
index 00000000..5d54baf7
--- /dev/null
+++ b/internal/telemetry/schema_batch_test.go
@@ -0,0 +1,89 @@
+package telemetry
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/parquet-go/parquet-go"
+)
+
+// _schema.batch is an empty Parquet batch whose only job is to carry this
+// binary's column list: internal/query derives every view's explicit
+// read_parquet(schema=MAP{...}) from it, so a column missing there is a column
+// missing from the view.
+//
+// It was written once, on the first boot into an empty data directory, and
+// never revisited. Every deployment since has been reading the column list of
+// whichever build happened to create the directory. Adding a column to
+// spanParquetRow would have shipped a binary whose views cannot bind it --
+// CreateViews fails, NewDuck returns an error, the process does not start, and
+// restarting does not help because the stale file is still there.
+//
+// The direction that matters is an OLD file and a NEW binary, so that is what
+// this sets up. A test that writes a fresh _schema.batch and reads it back
+// passes against the broken code.
+func TestSchemaBatchTracksTheRunningBinary(t *testing.T) {
+	dir := t.TempDir()
+	store, err := OpenParquetStore(dir)
+	if err != nil {
+		t.Fatal(err)
+	}
+	store.Close()
+
+	spansFile := filepath.Join(dir, "batches", SchemaBatch, "spans.parquet")
+	current := parquetColumns(t, spansFile)
+	if len(current) < 10 {
+		t.Fatalf("only %d columns in a freshly written schema batch: %v", len(current), current)
+	}
+
+	// An older build's _schema.batch: same file, fewer columns.
+	type oldSpanRow struct {
+		Namespace string `parquet:"namespace"`
+		TraceID   string `parquet:"trace_id"`
+		SpanID    string `parquet:"span_id"`
+	}
+	if err := os.Remove(spansFile); err != nil {
+		t.Fatal(err)
+	}
+	if err := writeTypedParquet(spansFile, []oldSpanRow{}, parquetPageSize); err != nil {
+		t.Fatal(err)
+	}
+	if got := parquetColumns(t, spansFile); len(got) != 3 {
+		t.Fatalf("fixture is wrong: wrote %d columns, want 3", len(got))
+	}
+
+	reopened, err := OpenParquetStore(dir)
+	if err != nil {
+		t.Fatalf("reopen against an older schema batch: %v", err)
+	}
+	defer reopened.Close()
+
+	got := parquetColumns(t, spansFile)
+	if len(got) != len(current) {
+		t.Errorf("after reopening, the schema batch has %d columns, want this binary's %d\n got:  %v\n want: %v",
+			len(got), len(current), got, current)
+	}
+}
+
+func parquetColumns(t *testing.T, path string) []string {
+	t.Helper()
+	file, err := os.Open(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer file.Close()
+	info, err := file.Stat()
+	if err != nil {
+		t.Fatal(err)
+	}
+	parquetFile, err := parquet.OpenFile(file, info.Size())
+	if err != nil {
+		t.Fatal(err)
+	}
+	var columns []string
+	for _, field := range parquetFile.Schema().Fields() {
+		columns = append(columns, field.Name())
+	}
+	return columns
+}
diff --git a/internal/telemetry/store/compaction.go b/internal/telemetry/store/compaction.go
index c10f5062..a2d0153b 100644
--- a/internal/telemetry/store/compaction.go
+++ b/internal/telemetry/store/compaction.go
@@ -16,8 +16,26 @@ import (
 )
 
 const (
-	maxCompactionRows  = 25_000_000
-	maxCompactionBytes = 256 << 20
+	maxCompactionRows = 25_000_000
+
+	// maxCompactionBytes bounds a pass by the bytes it admits, because that is
+	// what a merge costs: parquet-go opens a cursor per row group across every
+	// input and holds each one's dictionaries at once. Measured on real batches,
+	// peak RSS tracks the on-disk input bytes almost exactly -- 4 inputs
+	// totalling 442 MiB on disk merged at +413 MiB RSS.
+	//
+	// Row counts cannot stand in for it: wide spans carrying large attribute
+	// payloads and bare log lines differ by an order of magnitude at equal row
+	// counts.
+	//
+	// The figure is a memory budget, so it also decides how fast compaction can
+	// consolidate. Set too low it throttles the cascade: at 256 MiB a pass
+	// admitted only ~16 of the 16 MB generation-2 batches instead of the 128 the
+	// count ceiling allows, and 290 of them piled up unpromoted. 768 MiB keeps
+	// merge peak well inside the headroom the process now has -- it sits at
+	// 1.5 GiB against a 4 GB DuckDB budget -- while admitting enough inputs per
+	// pass for the higher generations to drain.
+	maxCompactionBytes = 768 << 20
 
 	// assumedCompactionBytesPerRow prices a batch whose size was never
 	// measured. Measured batches run well under this -- span rows with JSON
@@ -249,15 +267,46 @@ func selectCompactionBatches(batches []telemetry.BatchMetadata, maxBatches int)
 		key := compactionKey{day: batch.MaxIngestedNanos / int64(24*time.Hour), generation: batch.Generation}
 		groups[key] = append(groups[key], batch)
 	}
+	// Within a day, take the HIGHEST generation that has a compactable group,
+	// not the lowest.
+	//
+	// Lowest-first starves everything above it. Ingest refills generation 0
+	// continuously, so if it always wins no later generation is ever selected,
+	// while each pass it does run adds one more batch to the generation above.
+	// Observed on the live box as 508 / 371 / 290 batches at generations
+	// 0 / 1 / 2, generation 2 untouched for hours, and the live file count
+	// settling at an equilibrium instead of falling. That count is what every
+	// query pays per-file overhead on, so an equilibrium there is a permanent
+	// tax rather than a backlog that clears.
+	//
+	// Highest-first cannot starve anything. A generation above 0 only grows by
+	// one batch per pass of the generation below it, so its group fills slowly
+	// and empties in a few passes; once it no longer has a full group it stops
+	// being a candidate and generation 0 -- which is almost always ready --
+	// takes every remaining pass. Reducing the count by a group is worth the
+	// same wherever it happens, because the cost being reduced is per file.
+	//
+	// The older day still wins outright. Reclaiming space retention is about to
+	// drop matters more than consolidating today, and a day no longer being
+	// written finishes and stays finished.
+	newestDay := int64(math.MinInt64)
+	for key := range groups {
+		if key.day > newestDay {
+			newestDay = key.day
+		}
+	}
 	var chosen compactionKey
 	var selected []telemetry.BatchMetadata
 	found := false
 	for key, group := range groups {
-		candidate := selectBoundedCompactionGroup(group, maxBatches)
+		candidate := selectBoundedCompactionGroup(group, maxBatches, key.day < newestDay)
 		if len(candidate) < 2 {
 			continue
 		}
-		if !found || key.day < chosen.day || key.day == chosen.day && key.generation < chosen.generation {
+		better := !found ||
+			key.day < chosen.day ||
+			key.day == chosen.day && key.generation > chosen.generation
+		if better {
 			chosen, selected, found = key, candidate, true
 		}
 	}
@@ -271,7 +320,7 @@ func selectCompactionBatches(batches []telemetry.BatchMetadata, maxBatches int)
 // small files, but admits a smaller group when the row ceiling fills first.
 // Without the latter, one successful generation can make every later group too
 // large for the ceiling and permanently strand those files.
-func selectBoundedCompactionGroup(group []telemetry.BatchMetadata, maxBatches int) []telemetry.BatchMetadata {
+func selectBoundedCompactionGroup(group []telemetry.BatchMetadata, maxBatches int, finished bool) []telemetry.BatchMetadata {
 	ordered := append([]telemetry.BatchMetadata(nil), group...)
 	sort.Slice(ordered, func(i, j int) bool {
 		left, right := compactionBatchRows(ordered[i]), compactionBatchRows(ordered[j])
@@ -326,9 +375,25 @@ func selectBoundedCompactionGroup(group []telemetry.BatchMetadata, maxBatches in
 	if bytes == maxCompactionBytes {
 		saturated = true
 	}
+	// A group is worth merging when it is full, when a ceiling stopped it, or
+	// when it belongs to a day that has finished and already holds enough files
+	// to be worth the merge.
+	//
+	// Holding out for a full group is right while a day is still being written:
+	// more batches are coming, and merging early wastes the work. It is wrong
+	// once the day is over, because the group will never grow again. On the
+	// live box that stranded almost everything -- 1,243 batches in 69
+	// (day, generation) groups across 25 days, not one of them a candidate,
+	// because only the current day ever reached 128 files. Every compaction
+	// output was generation 1 and every older day was untouchable, which is why
+	// the live file count held at an equilibrium that neither reordering nor a
+	// larger byte budget could move.
 	if len(candidate) == maxBatches || saturated && len(candidate) >= 2 {
 		return candidate
 	}
+	if finished && len(candidate) >= minCompactionInputs {
+		return candidate
+	}
 	return nil
 }
 
diff --git a/internal/telemetry/store/compaction_bytes_test.go b/internal/telemetry/store/compaction_bytes_test.go
index 868b4cfb..294bfa1c 100644
--- a/internal/telemetry/store/compaction_bytes_test.go
+++ b/internal/telemetry/store/compaction_bytes_test.go
@@ -24,7 +24,7 @@ func TestSelectBoundedCompactionGroupStopsAtTheByteBudget(t *testing.T) {
 		})
 	}
 
-	candidate := selectBoundedCompactionGroup(group, 128)
+	candidate := selectBoundedCompactionGroup(group, 128, false)
 
 	if len(candidate) == 0 {
 		t.Fatal("no group selected; the byte budget must still admit a mergeable group")
@@ -58,7 +58,7 @@ func TestSelectBoundedCompactionGroupChargesUnsizedBatchesAnEstimate(t *testing.
 		})
 	}
 
-	candidate := selectBoundedCompactionGroup(group, 128)
+	candidate := selectBoundedCompactionGroup(group, 128, false)
 
 	if len(candidate) == 0 {
 		t.Fatal("no group selected; unsized batches must still be compactable")
diff --git a/internal/telemetry/store/compaction_starvation_test.go b/internal/telemetry/store/compaction_starvation_test.go
new file mode 100644
index 00000000..28f70953
--- /dev/null
+++ b/internal/telemetry/store/compaction_starvation_test.go
@@ -0,0 +1,107 @@
+package store
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/labstack/fanout/internal/telemetry"
+)
+
+func genBatches(day int64, generation uint32, n int, rows int) []telemetry.BatchMetadata {
+	out := make([]telemetry.BatchMetadata, 0, n)
+	for i := range n {
+		out = append(out, telemetry.BatchMetadata{
+			ID:               fmt.Sprintf("g%d-%04d", generation, i),
+			Generation:       generation,
+			Spans:            rows,
+			Bytes:            int64(rows) * 100,
+			MaxIngestedNanos: day*int64(24*time.Hour) + int64(i),
+			MinIngestedNanos: day*int64(24*time.Hour) + int64(i),
+		})
+	}
+	return out
+}
+
+// Ingest replenishes generation 0 continuously, so a rule that always prefers
+// the lowest generation never reaches the higher ones. Observed in production:
+// generation 0 and 1 were compacting while 290 generation-2 batches sat
+// untouched for hours, and the live file count -- which is what every query
+// pays per-file overhead on -- settled at an equilibrium instead of falling.
+func TestSelectCompactionBatchesDoesNotStarveHigherGenerations(t *testing.T) {
+	// The shape of the stall, taken from the live box: generation 0 replenished
+	// by ingest, generation 2 piled up above it. Both are past the count ceiling,
+	// so both are valid candidates and the tie-break decides.
+	batches := append(genBatches(1, 0, 508, 1_000), genBatches(1, 2, 290, 200_000)...)
+
+	selected := selectCompactionBatches(batches, 128)
+	if len(selected) == 0 {
+		t.Fatal("selected nothing with two compactable groups available")
+	}
+	if selected[0].Generation != 2 {
+		t.Errorf("selected generation %d when generation 2 has 290 batches waiting: "+
+			"a rule that always takes the lowest generation lets ingest starve every generation above it",
+			selected[0].Generation)
+	}
+}
+
+// The older day still wins: reclaiming disk that retention will drop next
+// matters more than consolidating today's files.
+func TestSelectCompactionBatchesStillPrefersTheOlderDay(t *testing.T) {
+	batches := append(genBatches(1, 0, 200, 1_000), genBatches(5, 2, 290, 200_000)...)
+
+	selected := selectCompactionBatches(batches, 128)
+	if len(selected) == 0 {
+		t.Fatal("selected nothing")
+	}
+	if selected[0].MaxIngestedNanos/int64(24*time.Hour) != 1 {
+		t.Errorf("selected day %d, want the older day 1 even though day 5 has more files",
+			selected[0].MaxIngestedNanos/int64(24*time.Hour))
+	}
+}
+
+// Generation 0 must still be served: it is refilled constantly, so once the
+// higher generations no longer have a full group it should take every pass.
+func TestSelectCompactionBatchesServesGenerationZeroWhenNothingIsPiledAbove(t *testing.T) {
+	// Only generation 0 has enough to form a group.
+	batches := append(genBatches(1, 0, 200, 1_000), genBatches(1, 2, 3, 1_000)...)
+
+	selected := selectCompactionBatches(batches, 128)
+	if len(selected) == 0 {
+		t.Fatal("selected nothing with a full generation-0 group available")
+	}
+	if selected[0].Generation != 0 {
+		t.Errorf("selected generation %d, want 0: with nothing compactable above it, the newest generation takes the pass",
+			selected[0].Generation)
+	}
+}
+
+// A day that has stopped being written still has to consolidate. Its group
+// never reaches the count ceiling and never saturates a row or byte ceiling, so
+// a rule that admits only full or saturated groups strands it forever.
+//
+// Measured on the live box: 1,243 batches spread over 69 (day, generation)
+// groups covering 25 days, every one of them returning no candidate. Only the
+// current day ever reached 128 files, so every compaction output was
+// generation 1 and everything older sat untouched -- which is why the live file
+// count held at an equilibrium no amount of reordering or budget-raising moved.
+func TestSelectBoundedCompactionGroupCompactsAnUnderfilledDay(t *testing.T) {
+	// A finished day: well under the count ceiling, nowhere near saturating.
+	group := genBatches(3, 1, 56, 5_000)
+
+	candidate := selectBoundedCompactionGroup(group, 128, true)
+
+	if len(candidate) == 0 {
+		t.Fatal("no candidate for a 56-batch day: a day that will never grow to 128 files can never consolidate")
+	}
+	if len(candidate) != 56 {
+		t.Errorf("candidate = %d, want all 56: nothing here forces a smaller group", len(candidate))
+	}
+}
+
+// Still not worth waking the merge for a couple of files.
+func TestSelectBoundedCompactionGroupIgnoresATrivialDay(t *testing.T) {
+	if got := selectBoundedCompactionGroup(genBatches(3, 1, 3, 5_000), 128, true); len(got) != 0 {
+		t.Errorf("candidate = %d for a 3-batch day, want none: below the effort floor", len(got))
+	}
+}
diff --git a/internal/telemetry/store/repository_test.go b/internal/telemetry/store/repository_test.go
index 43b2a1b1..e63dae70 100644
--- a/internal/telemetry/store/repository_test.go
+++ b/internal/telemetry/store/repository_test.go
@@ -689,9 +689,10 @@ func openTestDuckDB(t *testing.T) *sql.DB {
 }
 
 func traceAll(repository *Repository, traceID string) ([]telemetry.IndexedSpan, error) {
-	return repository.Parquet.Trace(context.Background(), telemetry.TraceQuery{
+	spans, _, err := repository.Parquet.Trace(context.Background(), telemetry.TraceQuery{
 		TraceID: traceID, StartNanos: -1 << 63, EndNanos: 1<<63 - 1, Limit: 500,
 	})
+	return spans, err
 }
 
 // seedFailedCompaction leaves a live COMPACTION.json whose recovery always