fix: stabilize snapshot attribution and scheduler retries

Cross-session snapshot selection can take complete token and cost data from a later transcript, but report materialization still credited that later session and equal snapshots depended on scan order. Preserve earliest-session attribution through every report backend and use session identity to settle ranking ties.

An exhausted embedding backstop retry could also be restarted by every periodic tick, keeping a detached daemon alive and making its regression timing-dependent. Latch that exhausted reconciliation until a real new mutation arrives.

Author: wesm

cmd/agentsview/embed_scheduler.go

@@ -166,6 +166,7 @@ func (s *embedScheduler) Run(ctx context.Context) {
 	// instead of retrying on the debounce interval. It is read and written only
 	// from this single goroutine, so it needs no synchronization of its own.
 	var pendingBackstop bool
+	var backstopRetryExhausted bool
 	var pendingRelease func()
 	var buildErrorRetries int
 	defer func() {
@@ -192,6 +193,7 @@ func (s *embedScheduler) Run(ctx context.Context) {
 			// A new mutation is a new work item, even if an earlier attempt is
 			// still waiting to retry.
 			buildErrorRetries = 0
+			backstopRetryExhausted = false
 			if pendingRelease == nil {
 				pendingRelease = release
 			} else {
@@ -217,6 +219,7 @@ func (s *embedScheduler) Run(ctx context.Context) {
 					// retries and release the lease, but let the next fresh
 					// notification carry Backstop: true instead of deferring the
 					// full pass until the next periodic tick.
+					backstopRetryExhausted = pendingBackstop
 					buildErrorRetries = 0
 					continue
 				}
@@ -236,8 +239,15 @@ func (s *embedScheduler) Run(ctx context.Context) {
 				pendingRelease = nil
 			}
 			pendingBackstop = false
+			backstopRetryExhausted = false
 			buildErrorRetries = 0
 		case <-backstopC:
+			if backstopRetryExhausted {
+				// This full reconciliation already consumed its bounded retry.
+				// Keep its intent for the next mutation without letting periodic
+				// ticks restart the same failed work and retain the daemon forever.
+				continue
+			}
 			if buildErrorRetries > 0 {
 				// A failed work item already owns the retained lease and
 				// retry timer. Fold the periodic reconciliation into it
@@ -276,6 +286,7 @@ func (s *embedScheduler) Run(ctx context.Context) {
 
 			release()
 			pendingBackstop = false
+			backstopRetryExhausted = false
 			if pendingRelease != nil {
 				// This successful full reconciliation also satisfies work
 				// that was already pending before the backstop began. A

cmd/agentsview/embed_scheduler_test.go

@@ -426,27 +426,15 @@ func TestEmbedSchedulerBackstopTicksDoNotRestartPendingRetry(t *testing.T) {
 	fake := &fakeEmbedManager{results: []fakeTryBuildResult{
 		{started: true, err: buildErr},
 	}}
-	idled := make(chan struct{})
-	ctx, cancel := context.WithCancel(t.Context())
-	tracker := server.NewIdleTracker(5*time.Millisecond, func() {
-		close(idled)
-		cancel()
-	})
 	s := newEmbedScheduler(
-		fake, 70*time.Millisecond, 20*time.Millisecond, false, tracker,
+		fake, 20*time.Millisecond, 5*time.Millisecond, false, nil,
 	)
-	go s.Run(ctx)
+	go s.Run(t.Context())
 	defer s.Stop()
 
-	waitForSchedulerCondition(t, func() bool { return fake.callCount() >= 1 },
-		"expected the initial backstop attempt")
-	go tracker.Run(ctx)
-	select {
-	case <-idled:
-	case <-time.After(500 * time.Millisecond):
-		require.Fail(t,
-			"frequent backstop ticks restarted the retry lifecycle and retained the idle lease")
-	}
+	waitForSchedulerCondition(t, func() bool { return fake.callCount() >= 2 },
+		"expected the failed backstop and its bounded retry")
+	time.Sleep(50 * time.Millisecond)
 	assert.Equal(t, []vector.BuildRequest{
 		{Backstop: true}, {Backstop: true},
 	}, fake.callsSnapshot(), "one backstop work item should get one bounded retry")

docs/internal/session-format-sources.md

@@ -134,9 +134,11 @@ Grok section and remove the explicit registry exception in the coverage test.
   can persist several assistant records with one `(message.id, requestId)` pair
   while `usage.output_tokens` grows from an early partial count to the final
   billed count (observed examples included `5` then `631` and `6` then `798`).
-  The session-usage and activity-report paths therefore keep the greatest
-  output-token snapshot within each session before applying cross-session
-  replay deduplication; equal snapshots retain their first occurrence.
+  Usage reporting therefore keeps the greatest output-token snapshot for each
+  message/request identity across the included sessions, attributes it to the
+  earliest transcript, and then applies cross-session replay deduplication.
+  Equal snapshots are selected deterministically by timestamp, session id, and
+  message ordinal.
   Replaying the three captured sessions after this correction matched all
   transcript-visible output; each full-wire total remained 15 output tokens
   higher because Claude Code's separate session-title request is not persisted.

internal/activity/activity.go

@@ -777,10 +777,19 @@ func claudeSnapshotSurvivorSelection(
 // report's range, effective-end, complete-snapshot, and cross-session dedup
 // filters.
 func UsageSurvivorMask(start, end, effEnd time.Time, usage []UsageRow) []bool {
-	mask, _ := usageSurvivorSelection(start, end, effEnd, usage)
+	mask, _ := UsageSurvivorSelection(start, end, effEnd, usage)
 	return mask
 }
 
+// UsageSurvivorSelection returns both the survivor mask and the session that
+// should receive each surviving row. A complete snapshot can come from a
+// later transcript while retaining the earliest transcript's attribution.
+func UsageSurvivorSelection(
+	start, end, effEnd time.Time, usage []UsageRow,
+) (mask []bool, attribution []string) {
+	return usageSurvivorSelection(start, end, effEnd, usage)
+}
+
 func usageSurvivorSelection(
 	start, end, effEnd time.Time, usage []UsageRow,
 ) (mask []bool, attribution []string) {

internal/activity/parity_pgtest_test.go

@@ -486,23 +486,25 @@ func assertParityForCase(
 
 // assertDayMinuteFixtureSanity checks the day-minute report actually exercises
 // the fixture: a full day with peak concurrency 2, nine sessions, non-zero
-// cost, and exactly 14050 output tokens. The token total proves the
+// cost, and exactly 22550 output tokens. The token total proves the
 // synthetic-model usage row (9999 tokens) is excluded, the dedup pair
-// collapses to its earlier 500-token row, the subagent's and unique fork's
-// tokens count, and the replaying fork's do not -- not merely that the
-// backends agree on a wrong number -- so the deep-compare above extends those
-// guarantees, plus the zero-cost primary-model fallback, to PG and DuckDB.
+// keeps its complete 9000-token snapshot with earlier attribution, the
+// subagent's and unique fork's tokens count, and the replaying fork's do not --
+// not merely that the backends agree on a wrong number. The deep-compare above
+// extends those guarantees, plus the zero-cost primary-model fallback, to PG
+// and DuckDB.
 func assertDayMinuteFixtureSanity(t *testing.T, r activity.Report) {
 	t.Helper()
 	require.False(t, r.Partial, "fixture day must be a full day")
 	require.Equal(t, 2, r.Peak.Agents, "fixture must reach peak concurrency 2")
 	require.Equal(t, 9, r.Totals.Sessions, "fixture session count")
 	require.Positive(t, r.Totals.Cost.Microdollars, "fixture must exercise cost")
 	// 2400 (parity-a) + 1600 (parity-b) + 300 (parity-c; synthetic 9999 row
-	// excluded) + 500 (parity-d wins the dedup) + 0 (parity-e deduped away;
+	// excluded) + 9000 (parity-d receives parity-e's complete snapshot) +
+	// 0 (parity-e deduped away;
 	// parity-f zero-cost) + 250 (parity-sub) + 9000 (parity-fork, unique)
-	// + 0 (parity-fork-replay deduped away) = 14050.
-	require.Equal(t, 14050, r.Totals.OutputTokens,
+	// + 0 (parity-fork-replay deduped away) = 22550.
+	require.Equal(t, 22550, r.Totals.OutputTokens,
 		"synthetic row excluded, dedup collapses, subagent and unique fork count")
 
 	bySession := map[string]activity.SessionRow{}
@@ -522,8 +524,8 @@ func assertDayMinuteFixtureSanity(t *testing.T, r activity.Report) {
 	require.Contains(t, bySession, "parity-d")
 	require.Contains(t, bySession, "parity-e")
 	require.Contains(t, bySession, "parity-f")
-	require.Equal(t, 500, bySession["parity-d"].OutputTokens,
-		"dedup keeps the earlier whole-second duplicate's tokens")
+	require.Equal(t, 9000, bySession["parity-d"].OutputTokens,
+		"complete snapshot is attributed to the earlier session")
 	require.Equal(t, 0, bySession["parity-e"].OutputTokens,
 		"the later fractional duplicate is dropped")
 	require.Equal(t, "model-x", bySession["parity-f"].PrimaryModel,

internal/db/activityreport.go

@@ -521,11 +521,11 @@ func (db *DB) activityReportUsageFrom(
 		_, row.OutputTokens, _, _, _ = dailyUsageRowTokens(candidate.scan)
 		baseRows[i] = row
 	}
-	mask := activity.UsageSurvivorMask(
+	mask, attribution := activity.UsageSurvivorSelection(
 		q.RangeStart, q.RangeEnd, q.EffectiveEnd, baseRows,
 	)
 	return materializeActivityReportUsageCandidates(
-		candidates, mask, rateResolver,
+		candidates, mask, attribution, rateResolver,
 	)
 }
 
@@ -667,13 +667,14 @@ func (db *DB) activityReportUsageCandidatesFrom(
 		return nil, nil, err
 	}
 	return materializeActivityReportUsageCandidates(
-		candidates, nil, rateResolver,
+		candidates, nil, nil, rateResolver,
 	)
 }
 
 func materializeActivityReportUsageCandidates(
 	candidates []activityReportUsageCandidate,
 	mask []bool,
+	attribution []string,
 	rateResolver *export.PricingResolver,
 ) ([]activity.UsageRow, *export.PricingBlock, error) {
 	out := make([]activity.UsageRow, 0, len(candidates))
@@ -702,6 +703,9 @@ func materializeActivityReportUsageCandidates(
 			costSource = export.CostSourceReported
 		}
 		row := candidate.row
+		if attribution != nil {
+			row.SessionID = attribution[i]
+		}
 		row.InputTokens = inputTok
 		row.OutputTokens = outputTok
 		row.CacheCreationTokens = cacheCrTok

internal/db/activityreport_test.go

@@ -367,6 +367,7 @@ func TestGetActivityReport_PricingModelsOnlyIncludeDedupSurvivors(t *testing.T)
 		s.Agent = "claude"
 		s.StartedAt = Ptr("2026-06-16T10:31:00Z")
 		s.EndedAt = Ptr("2026-06-16T10:31:00Z")
+		s.IsAutomated = true
 	})
 	insertMessages(t, d, Message{
 		SessionID: "later", Ordinal: 0, Role: "assistant", Content: "x",
@@ -380,6 +381,16 @@ func TestGetActivityReport_PricingModelsOnlyIncludeDedupSurvivors(t *testing.T)
 		dayQuery(t, "2026-06-16", "UTC"))
 	require.NoError(t, err)
 	assert.Equal(t, 900, r.Totals.OutputTokens)
+	assert.Equal(t, r.Totals.Cost, r.Totals.InteractiveCost)
+	assert.Zero(t, r.Totals.AutomatedCost.Microdollars)
+	bySession := make(map[string]activity.SessionRow, len(r.BySession))
+	for _, session := range r.BySession {
+		bySession[session.SessionID] = session
+	}
+	require.Contains(t, bySession, "earlier")
+	require.Contains(t, bySession, "later")
+	assert.Equal(t, 900, bySession["earlier"].OutputTokens)
+	assert.Zero(t, bySession["later"].OutputTokens)
 	require.NotNil(t, r.Pricing)
 	assert.Contains(t, r.Pricing.Models, "complete-model")
 	assert.NotContains(t, r.Pricing.Models, "partial-model")

internal/db/reporting_export.go

@@ -422,13 +422,15 @@ func finalizeReportingUsage(
 	rows []activity.UsageRow,
 ) ([]activity.UsageRow, error) {
 	sortReportingUsage(rows)
-	mask := activity.UsageSurvivorMask(
+	mask, attribution := activity.UsageSurvivorSelection(
 		query.RangeStart, query.RangeEnd, query.EffectiveEnd, rows,
 	)
 	survivors := make([]activity.UsageRow, 0, len(rows))
 	for i, keep := range mask {
 		if keep {
-			survivors = append(survivors, rows[i])
+			row := rows[i]
+			row.SessionID = attribution[i]
+			survivors = append(survivors, row)
 		}
 	}
 	return allocateReportingUsageCosts(survivors)

internal/db/reporting_export_test.go

@@ -869,6 +869,51 @@ func TestFinalizeReportingUsageOrdering(t *testing.T) {
 	}
 }
 
+func TestFinalizeReportingUsageAttributesCompleteSnapshotToEarliestSession(t *testing.T) {
+	query := activity.Query{
+		RangeStart:   time.Date(2026, 7, 28, 9, 0, 0, 0, time.UTC),
+		RangeEnd:     time.Date(2026, 7, 28, 10, 0, 0, 0, time.UTC),
+		EffectiveEnd: time.Date(2026, 7, 28, 10, 0, 0, 0, time.UTC),
+	}
+	rows := []activity.UsageRow{
+		{
+			SessionID:       "earlier-parent",
+			MessageOrdinal:  1,
+			UsageSource:     "message",
+			Timestamp:       "2026-07-28T09:05:00Z",
+			Model:           "model-a",
+			OutputTokens:    100,
+			Cost:            money.Money{Microdollars: 1000},
+			CostSource:      export.CostSourceReported,
+			Priced:          true,
+			Contributes:     true,
+			ClaudeMessageID: "shared-message",
+			ClaudeRequestID: "shared-request",
+		},
+		{
+			SessionID:       "later-child",
+			MessageOrdinal:  1,
+			UsageSource:     "message",
+			Timestamp:       "2026-07-28T09:06:00Z",
+			Model:           "model-a",
+			OutputTokens:    900,
+			Cost:            money.Money{Microdollars: 9000},
+			CostSource:      export.CostSourceReported,
+			Priced:          true,
+			Contributes:     true,
+			ClaudeMessageID: "shared-message",
+			ClaudeRequestID: "shared-request",
+		},
+	}
+
+	survivors, err := finalizeReportingUsage(query, rows)
+	require.NoError(t, err)
+	require.Len(t, survivors, 1)
+	assert.Equal(t, "earlier-parent", survivors[0].SessionID)
+	assert.Equal(t, 900, survivors[0].OutputTokens)
+	assert.Equal(t, money.Money{Microdollars: 9000}, survivors[0].Cost)
+}
+
 func reportingUsageBreakdownKeys(
 	rows []export.ReportingUsageBreakdown,
 ) []string {

internal/db/usage.go

@@ -1139,7 +1139,7 @@ func snapshotRankedDailyUsageRowsSQL(
 				ROW_NUMBER() OVER (
 					PARTITION BY claude_message_id, claude_request_id
 					ORDER BY snapshot_output_tokens DESC, ts ASC,
-						COALESCE(message_ordinal, -1) ASC
+						session_id ASC, COALESCE(message_ordinal, -1) ASC
 				) AS snapshot_rank
 			FROM usage_snapshot_window
 			WHERE claude_message_id != '' AND claude_request_id != ''