fix: bound memory and per-query cost that scaled with accumulated state

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 != "" {

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))
+		}
+	}
+}

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)
+	}
+}