feat(usage): derive compute-usage rollups and prune the raw compute log (RIG-2872)

docs/concepts/tokens-and-billing.md

@@ -60,7 +60,10 @@ health/quota signal.
   an estimated start. The sweep's ends and backfilled starts carry the
   `estimated` flag. Other events are stamped when the server records the
   binding change; a Runner-reconnect reap records the reap time. The log is
-  auditable and reconstructable.
+  auditable and reconstructable. Closed intervals feed derived hourly and daily
+  duration rollups, while open intervals are excluded. The raw compute log
+  follows the configured event retention window; rollups remain available after
+  pruning.
 - **LLM token usage and spend — recorded for display, not billed day-1.** Every
   model call's tokens-in/out and cost, captured at the gateway. It powers the
   in-product usage/spend charts the user sees, and it is *recorded* even though

go/cmd/compass-server/main.go

@@ -368,7 +368,7 @@ func registerServeFlags(fs *flag.FlagSet) serveFlags {
 		transcriptSafetyValveCapBytes: fs.Int("transcript-safety-valve-cap-bytes", 0,
 			"Hot-tail safety-valve cap in bytes. Defaults to $COMPASS_TRANSCRIPT_SAFETY_VALVE_CAP_BYTES."),
 		usageRetention: fs.String("usage-event-retention", "",
-			"How long raw token-usage events are kept before the daily prune "+
+			"How long raw token- and compute-usage events are kept before the daily prune "+
 				"deletes them, as a Go duration (e.g. 720h). The hourly and daily "+
 				"usage totals are kept. Falls back to $COMPASS_USAGE_EVENT_RETENTION, "+
 				"then 2160h (90 days). 0 disables the prune."),

go/internal/store/compute_usage_pgtest_test.go

@@ -444,3 +444,118 @@ func execAsSystem(t *testing.T, s *Store, sql string, args ...any) {
 		t.Fatalf("commit: %v", err)
 	}
 }
+
+// Two system sweeps insert ends for the same tenants in opposite natural scan
+// orders. The held first-tenant lock makes the inversion deterministic.
+func TestComputeUsageCrossTenantSweepsLockTenantsInOrder(t *testing.T) {
+	ctx := t.Context()
+	s := newTestStore(t)
+	tenantA := seedTenant(t, s, "compute-lock-a")
+	tenantB := seedTenant(t, s, "compute-lock-b")
+	first, second := tenantA, tenantB
+	if string(first) > string(second) {
+		first, second = second, first
+	}
+
+	// The runner index retains heap insertion order for its duplicate key, so
+	// seed bindings in reverse tenant order; orphan starts follow tenant order.
+	for i, tenant := range []TenantID{second, first} {
+		suffix := []string{"b", "a"}[i]
+		tenantCtx := WithTenant(ctx, tenant)
+		owner, err := s.CreateUser(tenantCtx, NewUser{
+			Handle: "compute-lock-owner-" + suffix, DisplayName: "compute-lock-owner-" + suffix,
+		})
+		if err != nil {
+			t.Fatalf("CreateUser(%s): %v", tenant, err)
+		}
+		agent, err := s.CreateAgent(tenantCtx, owner.ID, NewAgent{
+			Handle: "compute-lock-agent-" + suffix, DisplayName: "compute-lock-agent-" + suffix,
+		})
+		if err != nil {
+			t.Fatalf("CreateAgent(%s): %v", tenant, err)
+		}
+		if _, err := s.RecordSessionBinding(tenantCtx, "compute-lock-session-"+suffix, agent.ID, "compute-lock-runner"); err != nil {
+			t.Fatalf("RecordSessionBinding(%s): %v", tenant, err)
+		}
+	}
+	for i, tenant := range []TenantID{first, second} {
+		suffix := []string{"a", "b"}[i]
+		if _, err := s.pool.Exec(ctx, `
+			INSERT INTO compute_usage_events (
+			    tenant_id, id, interval_id, kind, occurred_at, agent_account_id,
+			    owner_user_id, session_id, runner_id, estimated
+			) VALUES ($1, $2, $3, 'start', clock_timestamp(), $4, $5, $6, $7, TRUE)`,
+			string(tenant), "compute-lock-orphan-start-"+suffix, "compute-lock-orphan-"+suffix,
+			"compute-lock-orphan-agent-"+suffix, "compute-lock-orphan-owner-"+suffix,
+			"compute-lock-orphan-session-"+suffix, "compute-lock-orphan-runner"); err != nil {
+			t.Fatalf("seed orphan start for %s: %v", tenant, err)
+		}
+	}
+
+	gate, err := s.pool.Begin(ctx)
+	if err != nil {
+		t.Fatalf("begin advisory-lock gate: %v", err)
+	}
+	defer func() {
+		if err := gate.Rollback(ctx); err != nil && !errors.Is(err, pgx.ErrTxClosed) {
+			t.Errorf("rollback advisory-lock gate: %v", err)
+		}
+	}()
+	if _, err := gate.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtext('compute_usage:' || $1))`, string(first)); err != nil {
+		t.Fatalf("hold first-tenant compute lock: %v", err)
+	}
+
+	type result struct {
+		name string
+		err  error
+	}
+	done := make(chan result, 2)
+	go func() {
+		_, err := s.CloseOrphanedComputeIntervals(WithSystemRole(ctx))
+		done <- result{name: "orphan sweep", err: err}
+	}()
+	waitForWaiters := func(wantWaiters int) {
+		t.Helper()
+		deadline := time.After(5 * time.Second)
+		tick := time.NewTicker(10 * time.Millisecond)
+		defer tick.Stop()
+		for {
+			var waiters int
+			if err := gate.QueryRow(ctx, `
+				WITH key AS (SELECT hashtext('compute_usage:' || $1)::bigint AS value)
+				SELECT count(*)
+				  FROM pg_locks, key
+				 WHERE locktype = 'advisory' AND NOT granted
+				   AND classid = ((key.value >> 32) & 4294967295)::oid
+				   AND objid = (key.value & 4294967295)::oid`, string(first)).Scan(&waiters); err != nil {
+				t.Fatalf("read compute advisory lock waiters: %v", err)
+			}
+			if waiters >= wantWaiters {
+				return
+			}
+			select {
+			case result := <-done:
+				t.Fatalf("%s completed before lock gate release: %v", result.name, result.err)
+			case <-deadline:
+				t.Fatalf("%d sweeps wait on first-tenant lock; want at least %d", waiters, wantWaiters)
+			case <-tick.C:
+			}
+		}
+	}
+	waitForWaiters(1)
+	go func() {
+		_, err := s.DeleteSessionBindingsForRunner(WithSystemRole(ctx), "compute-lock-runner")
+		done <- result{name: "runner sweep", err: err}
+	}()
+	waitForWaiters(2)
+
+	if err := gate.Rollback(ctx); err != nil {
+		t.Fatalf("release first-tenant compute lock: %v", err)
+	}
+	for range 2 {
+		result := <-done
+		if result.err != nil {
+			t.Errorf("%s: %v", result.name, result.err)
+		}
+	}
+}

go/internal/store/compute_usage_rollup_migration_pgtest_test.go

@@ -0,0 +1,161 @@
+//go:build pgtest
+
+package store
+
+import (
+	"context"
+	"slices"
+	"testing"
+	"time"
+
+	"github.com/jackc/pgx/v5/pgtype"
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	"github.com/RigelBuild/compass/go/internal/pgtest"
+)
+
+func TestComputeUsageRollupMigrationBackfillsClosedIntervals(t *testing.T) {
+	ctx := t.Context()
+	dsn := pgtest.RequireDSN(t)
+	// Look the migration up by file name, so its version is never hard-coded.
+	migs, err := loadMigrations()
+	if err != nil {
+		t.Fatalf("load migrations: %v", err)
+	}
+	i := slices.IndexFunc(migs, func(m migration) bool { return m.name == "0003_compute_usage_rollups.sql" })
+	if i < 0 {
+		t.Fatal("migration 0003_compute_usage_rollups.sql is not embedded")
+	}
+	version := migs[i].version
+	applyMigrationsThrough(t, dsn, version-1)
+
+	seed, err := pgxpool.New(ctx, dsn)
+	if err != nil {
+		t.Fatalf("connect to seed pre-rollup data: %v", err)
+	}
+	start := time.Date(2026, time.March, 10, 23, 30, 0, 0, time.UTC)
+	end := start.Add(time.Hour)
+	_, err = seed.Exec(ctx,
+		"INSERT INTO tenants (id, slug, display_name, created_at_unix_ms) VALUES ('rollup-tenant-a', 'rollup-tenant-a', 'A', 0), ('rollup-tenant-b', 'rollup-tenant-b', 'B', 0)")
+	if err == nil {
+		_, err = seed.Exec(ctx, `
+			INSERT INTO compute_usage_events (
+			    tenant_id, id, interval_id, kind, occurred_at, agent_account_id,
+			    owner_user_id, session_id, runner_id
+			) VALUES
+			    ('rollup-tenant-a', 'start', 'interval', 'start', $1, 'agent', 'owner', 'session', 'runner'),
+			    ('rollup-tenant-a', 'end', 'interval', 'end', $2, 'agent', 'owner', 'session', 'runner'),
+			    ('rollup-tenant-b', 'start', 'interval', 'start', $1, 'agent', 'owner', 'session', 'runner'),
+			    ('rollup-tenant-b', 'end', 'interval', 'end', $2, 'agent', 'owner', 'session', 'runner')`, start, end)
+	}
+	seed.Close()
+	if err != nil {
+		t.Fatalf("seed pre-rollup closed intervals: %v", err)
+	}
+
+	applyMigrationsThrough(t, dsn, version)
+	s := reopenStore(t, dsn)
+	var horizon pgtype.Timestamptz
+	if err := s.pool.QueryRow(ctx, "SELECT horizon FROM compute_usage_prune_horizon").Scan(&horizon); err != nil {
+		t.Fatalf("read compute prune horizon: %v", err)
+	}
+	if horizon.InfinityModifier != pgtype.NegativeInfinity {
+		t.Fatalf("compute prune horizon = %+v, want -infinity", horizon)
+	}
+	for _, tc := range []struct {
+		name      string
+		table     string
+		buckets   []time.Time
+		activeMS  []int64
+		intervals []int64
+	}{
+		{
+			name:      "hourly",
+			table:     "compute_usage_rollups_hourly",
+			buckets:   []time.Time{time.Date(2026, time.March, 10, 23, 0, 0, 0, time.UTC), time.Date(2026, time.March, 11, 0, 0, 0, 0, time.UTC)},
+			activeMS:  []int64{30 * 60 * 1000, 30 * 60 * 1000},
+			intervals: []int64{1, 0},
+		},
+		{
+			name:      "daily",
+			table:     "compute_usage_rollups_daily",
+			buckets:   []time.Time{time.Date(2026, time.March, 10, 0, 0, 0, 0, time.UTC), time.Date(2026, time.March, 11, 0, 0, 0, 0, time.UTC)},
+			activeMS:  []int64{30 * 60 * 1000, 30 * 60 * 1000},
+			intervals: []int64{1, 0},
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			rows, err := s.pool.Query(ctx, "SELECT tenant_id, bucket_start, active_ms, intervals FROM "+tc.table+" ORDER BY tenant_id, bucket_start")
+			if err != nil {
+				t.Fatalf("read %s backfill: %v", tc.name, err)
+			}
+			defer rows.Close()
+			for _, tenant := range []string{"rollup-tenant-a", "rollup-tenant-b"} {
+				for bucketIndex, bucket := range tc.buckets {
+					if !rows.Next() {
+						t.Fatalf("%s backfill has no row for %s at %s", tc.name, tenant, bucket)
+					}
+					var gotTenant string
+					var gotBucket time.Time
+					var activeMS, intervals int64
+					if err := rows.Scan(&gotTenant, &gotBucket, &activeMS, &intervals); err != nil {
+						t.Fatalf("scan %s backfill: %v", tc.name, err)
+					}
+					if gotTenant != tenant || !gotBucket.Equal(bucket) || activeMS != tc.activeMS[bucketIndex] || intervals != tc.intervals[bucketIndex] {
+						t.Errorf("%s backfill row = %s, %s, %d ms, %d intervals; want %s, %s, %d ms, %d intervals",
+							tc.name, gotTenant, gotBucket, activeMS, intervals,
+							tenant, bucket, tc.activeMS[bucketIndex], tc.intervals[bucketIndex])
+					}
+				}
+			}
+			if rows.Next() {
+				t.Errorf("%s backfill has extra rows", tc.name)
+			}
+			if err := rows.Err(); err != nil {
+				t.Fatalf("iterate %s backfill: %v", tc.name, err)
+			}
+		})
+	}
+}
+
+func applyMigrationsThrough(t *testing.T, dsn string, version int) {
+	t.Helper()
+	ctx := t.Context()
+	migs, err := loadMigrations()
+	if err != nil {
+		t.Fatalf("load migrations: %v", err)
+	}
+	pool, err := pgxpool.New(ctx, dsn)
+	if err != nil {
+		t.Fatalf("connect to apply migrations: %v", err)
+	}
+	defer pool.Close()
+	conn, err := pool.Acquire(ctx)
+	if err != nil {
+		t.Fatalf("acquire to apply migrations: %v", err)
+	}
+	defer conn.Release()
+	if _, err := conn.Exec(ctx, "SELECT pg_advisory_lock($1)", migrationLockKey); err != nil {
+		t.Fatalf("acquire migration lock: %v", err)
+	}
+	defer func() {
+		if _, err := conn.Exec(context.WithoutCancel(ctx), "SELECT pg_advisory_unlock($1)", migrationLockKey); err != nil {
+			t.Errorf("release migration lock: %v", err)
+		}
+	}()
+	if err := ensureMigrationsTable(ctx, conn); err != nil {
+		t.Fatal(err)
+	}
+	applied, err := appliedChecksums(ctx, conn)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, migration := range migs {
+		if _, done := applied[migration.version]; migration.version > version || done {
+			continue
+		}
+		if err := applyMigration(ctx, conn, migration); err != nil {
+			t.Fatal(err)
+		}
+	}
+}