perf(db): spool only changed usage facts

.github/workflows/ci.yml

@@ -544,7 +544,7 @@ jobs:
 
       - name: Upload ${{ matrix.name }} test events
         if: failure()
-        uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9  # v7.0.2
+        uses: actions/upload-artifact@v7.0.2
         with:
           name: go-test-events-windows-${{ matrix.name }}-attempt-${{ github.run_attempt }}
           path: go-test-events.jsonl

docs/agents/storage.md

@@ -646,6 +646,14 @@ of a few seconds is expected and intended. Do not reintroduce a whole-snapshot
 recheck against the archive: validating a snapshot against a source that changes
 one session at a time livelocks the request.
 
+Refills still extract complete session histories, but compare facts in bounded
+batches against one read-only cache snapshot and spool only changed or new
+rows. Comparison requires the expected install revision; installation checks
+that revision again. Complete per-session fact counts remove obsolete tails,
+and identity invalidation compares the complete cache before and after install.
+An append can still shift later activity or event indexes and spool those rows.
+This reduces temporary writes; it does not make source extraction incremental.
+
 Timezone rollup identity includes both the resolved zone name and its rule
 fingerprint. Cache-generation retirement cancels detached work immediately but
 keeps immutable coordinator pointers and the cache database alive until active

internal/db/usage_cache_fill.go

@@ -258,7 +258,7 @@ func (c *usageFillCoordinator) fillSessionsDetached(
 		// version they belong to, so the install never has to consult the
 		// archive again. The cache expectation prevents an older concurrent
 		// extraction from replacing facts installed after this attempt began.
-		spool, extracted, err := c.extractSessions(ctx, pending)
+		spool, extracted, err := c.extractSessions(ctx, pending, expected)
 		if err != nil {
 			return nil, err
 		}
@@ -429,6 +429,11 @@ func newUsageFactSpool(ctx context.Context) (*usageFactSpool, error) {
 			source_uuid TEXT NOT NULL, usage_dedup_key TEXT NOT NULL,
 			token_eligible INTEGER NOT NULL, activity_eligible INTEGER NOT NULL,
 			PRIMARY KEY(session_id, fact_index)
+		) WITHOUT ROWID;
+		CREATE TABLE sessions (
+			session_id TEXT PRIMARY KEY,
+			install_revision INTEGER,
+			fact_count INTEGER NOT NULL DEFAULT 0
 		) WITHOUT ROWID`); err != nil {
 		_ = database.Close()
 		_ = removeUsageCacheFiles(path)
@@ -442,12 +447,13 @@ func (s *usageFactSpool) Close() error {
 }
 
 // extractSessions reads the requested sessions' usage facts and their source
-// versions inside one archive read transaction. WAL gives that transaction a
-// consistent snapshot, so the returned versions describe exactly the facts in
-// the spool. Sessions missing from the returned map were deleted as of the
-// snapshot.
+// versions inside one archive read transaction. Equal cached facts are omitted
+// from the spool only when the cache still matches the installation expectation.
+// The complete snapshot's fact counts let installation remove obsolete tails.
+// Sessions missing from the returned map were deleted as of the snapshot.
 func (c *usageFillCoordinator) extractSessions(
 	ctx context.Context, versions []usageSourceVersion,
+	expected map[string]usageFillInstallExpectation,
 ) (*usageFactSpool, map[string]usageSourceVersion, error) {
 	release, err := c.archive.acquireBulkRead(ctx)
 	if err != nil {
@@ -519,13 +525,31 @@ func (c *usageFillCoordinator) extractSessions(
 		return nil, nil, err
 	}
 
+	// Pin one cache snapshot while comparing bounded batches. This connection
+	// writes only the disposable spool; WAL lets cache writers keep progressing.
+	if _, err := spool.db.ExecContext(ctx,
+		`ATTACH DATABASE ? AS comparison`, makeDSN(c.cache.path, true),
+	); err != nil {
+		return nil, nil, fmt.Errorf("attaching usage fact comparison: %w", err)
+	}
 	spoolTx, err := spool.db.BeginTx(ctx, nil)
 	if err != nil {
 		return nil, nil, err
 	}
 	defer func() { _ = spoolTx.Rollback() }()
+	for id := range extracted {
+		var revision any
+		if expectation := expected[id]; expectation.Exists {
+			revision = expectation.InstallRevision
+		}
+		if _, err := spoolTx.ExecContext(ctx,
+			`INSERT INTO sessions(session_id, install_revision) VALUES (?, ?)`,
+			id, revision); err != nil {
+			return nil, nil, err
+		}
+	}
 	indexes := make(map[string]int, len(versions))
-	spoolWriter := usageFactSpoolWriter{ctx: ctx, tx: spoolTx}
+	spoolWriter := usageFactSpoolWriter{ctx: ctx, tx: spoolTx, compare: true}
 	if err := extractUsageMessageFacts(ctx, tx, &spoolWriter, indexes); err != nil {
 		return nil, nil, err
 	}
@@ -538,9 +562,19 @@ func (c *usageFillCoordinator) extractSessions(
 	if err := spoolWriter.Flush(); err != nil {
 		return nil, nil, err
 	}
+	for id, count := range indexes {
+		if _, err := spoolTx.ExecContext(ctx,
+			`UPDATE sessions SET fact_count = ? WHERE session_id = ?`, count, id,
+		); err != nil {
+			return nil, nil, err
+		}
+	}
 	if err := spoolTx.Commit(); err != nil {
 		return nil, nil, fmt.Errorf("committing usage fact spool: %w", err)
 	}
+	if _, err := spool.db.ExecContext(ctx, `DETACH DATABASE comparison`); err != nil {
+		return nil, nil, fmt.Errorf("detaching usage fact comparison: %w", err)
+	}
 	if err := tx.Commit(); err != nil {
 		return nil, nil, fmt.Errorf("closing archive usage fill snapshot: %w", err)
 	}
@@ -665,6 +699,9 @@ type usageFactSpoolWriter struct {
 	ctx  context.Context
 	tx   *sql.Tx
 	rows []usageFactSpoolRow
+
+	compare   bool
+	fullBatch *sql.Stmt
 }
 
 func (w *usageFactSpoolWriter) Add(
@@ -683,14 +720,18 @@ func (w *usageFactSpoolWriter) Flush() error {
 	if len(w.rows) == 0 {
 		return nil
 	}
-	const prefix = `INSERT INTO facts(
+	const columns = `
 		session_id, fact_index, source, message_ordinal, timestamp_ms, timestamp_ns,
 		raw_timestamp, uses_session_start, model, input_tokens, output_tokens,
 		reasoning_tokens, provider_id, cache_creation_tokens, cache_creation_1h_tokens,
 		cache_read_tokens, web_search_requests, reported_cost_microdollars,
 		cost_source, request_scoped, claude_message_id, claude_request_id,
 		source_uuid, usage_dedup_key, token_eligible, activity_eligible
-	) VALUES `
+	`
+	prefix := `INSERT INTO facts(` + columns + `) VALUES `
+	if w.compare {
+		prefix = `WITH incoming(` + columns + `) AS (VALUES `
+	}
 	const placeholders = `(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`
 	var query strings.Builder
 	query.Grow(len(prefix) + len(w.rows)*(len(placeholders)+1))
@@ -715,7 +756,48 @@ func (w *usageFactSpoolWriter) Flush() error {
 			boolInt(fact.TokenEligible), boolInt(fact.ActivityEligible),
 		)
 	}
-	_, err := w.tx.ExecContext(w.ctx, query.String(), args...)
+	if w.compare {
+		// Never omit rows based on a different cache installation. The final
+		// install fence rejects a race even if it occurred before this snapshot.
+		query.WriteString(`) INSERT INTO facts(` + columns + `) SELECT * FROM incoming
+			WHERE NOT EXISTS (
+				SELECT 1 FROM comparison.usage_cached_sessions cached
+				JOIN sessions expected ON expected.session_id = cached.session_id
+				  AND expected.install_revision = cached.install_revision
+				JOIN comparison.usage_facts prior ON prior.cached_session_id = cached.id
+				  AND prior.fact_index = incoming.fact_index
+				WHERE cached.session_id = incoming.session_id
+				  AND (prior.source, prior.message_ordinal, prior.timestamp_ms,
+					prior.timestamp_ns, prior.raw_timestamp, prior.uses_session_start,
+					prior.model, prior.input_tokens, prior.output_tokens, prior.reasoning_tokens,
+					prior.provider_id, prior.cache_creation_tokens, prior.cache_creation_1h_tokens,
+					prior.cache_read_tokens, prior.web_search_requests, prior.reported_cost_microdollars,
+					prior.cost_source, prior.request_scoped, prior.claude_message_id,
+					prior.claude_request_id, prior.source_uuid, prior.usage_dedup_key,
+					prior.token_eligible, prior.activity_eligible) IS
+					(incoming.source, incoming.message_ordinal, incoming.timestamp_ms,
+					incoming.timestamp_ns, incoming.raw_timestamp, incoming.uses_session_start,
+					incoming.model, incoming.input_tokens, incoming.output_tokens, incoming.reasoning_tokens,
+					incoming.provider_id, incoming.cache_creation_tokens, incoming.cache_creation_1h_tokens,
+					incoming.cache_read_tokens, incoming.web_search_requests, incoming.reported_cost_microdollars,
+					incoming.cost_source, incoming.request_scoped, incoming.claude_message_id,
+					incoming.claude_request_id, incoming.source_uuid, incoming.usage_dedup_key,
+					incoming.token_eligible, incoming.activity_eligible)
+			)`)
+	}
+	var err error
+	if len(w.rows) == usageFactSpoolBatchSize {
+		// Preparing the large VALUES expression costs more than comparing a
+		// batch. Reuse it within this extraction; the transaction closes it.
+		if w.fullBatch == nil {
+			w.fullBatch, err = w.tx.PrepareContext(w.ctx, query.String())
+		}
+		if err == nil {
+			_, err = w.fullBatch.ExecContext(w.ctx, args...)
+		}
+	} else {
+		_, err = w.tx.ExecContext(w.ctx, query.String(), args...)
+	}
 	if err != nil {
 		return fmt.Errorf("spooling %d usage facts: %w", len(w.rows), err)
 	}
@@ -914,7 +996,7 @@ func (c *usageFillCoordinator) invalidateChangedIdentitySharers(
 	for _, version := range stable {
 		stableIDs = append(stableIDs, version.SessionID)
 	}
-	newIdentities, err := usageSpoolIdentitiesForSessions(ctx, conn, stableIDs)
+	newIdentities, err := usageFactIdentitiesForSessions(ctx, conn, stableIDs)
 	if err != nil {
 		return nil, err
 	}
@@ -1081,6 +1163,18 @@ func installSpoolSessions(
 	if _, err := cacheConn.ExecContext(ctx, statement.String(), args...); err != nil {
 		return nil, err
 	}
+	var incomplete int
+	if err := cacheConn.QueryRowContext(ctx, `
+		SELECT COUNT(*) FROM usage_install_sessions install
+		LEFT JOIN usage_fill_spool.sessions complete
+		  ON complete.session_id = install.session_id
+		WHERE complete.session_id IS NULL OR complete.fact_count < 0`,
+	).Scan(&incomplete); err != nil {
+		return nil, err
+	}
+	if incomplete != 0 {
+		return nil, errors.New("usage fact spool is missing complete session counts")
+	}
 	if _, err := cacheConn.ExecContext(ctx, `
 		INSERT INTO usage_cached_sessions(
 			session_id, source_sync_marker, source_transcript_rev,
@@ -1101,17 +1195,17 @@ func installSpoolSessions(
 			SELECT cached.id FROM usage_cached_sessions cached
 			JOIN usage_install_sessions install
 			  ON install.session_id = cached.session_id
-		) AND NOT EXISTS (
-			SELECT 1 FROM usage_fill_spool.facts facts
+		) AND EXISTS (
+			SELECT 1 FROM usage_fill_spool.sessions complete
 			JOIN usage_cached_sessions cached
-			  ON cached.session_id = facts.session_id
+			  ON cached.session_id = complete.session_id
 			WHERE cached.id = usage_facts.cached_session_id
-			  AND facts.fact_index = usage_facts.fact_index
+			  AND usage_facts.fact_index >= complete.fact_count
 		)`); err != nil {
 		return nil, err
 	}
-	// Extraction still supplies a complete snapshot. Retain equal facts so a
-	// small append does not rewrite the session's history and identity indexes.
+	// Extraction assigns dense indexes to the complete snapshot, but spools
+	// only changed/new facts. Retain equal facts and their identity indexes.
 	// IS NOT compares nullable fields without treating NULL as a change.
 	if _, err := cacheConn.ExecContext(ctx, `INSERT INTO usage_facts(
 		cached_session_id, fact_index, source, message_ordinal, timestamp_ms,