diff --git a/internal/capture/store/outbox.go b/internal/capture/store/outbox.go index f4943dd..d9fcc6b 100644 --- a/internal/capture/store/outbox.go +++ b/internal/capture/store/outbox.go @@ -242,10 +242,19 @@ func (s *Store) SetDrainState(ctx context.Context, state string, now time.Time) } // OutboxCounts counts rows by sync_state across the uploaded tables. +// +// Pending leaves out rows of aborted turns, which Pending never reads: counting +// them reported work as waiting to sync that never would, so the pending count +// could not reach zero. func (s *Store) OutboxCounts(ctx context.Context) (map[string]int64, error) { out := map[string]int64{} for _, t := range syncTables { - rows, err := s.db.QueryContext(ctx, `SELECT sync_state, COUNT(*) FROM `+t+` GROUP BY sync_state`) //nolint:gosec // G202: fixed table list + q := `SELECT sync_state, COUNT(*) FROM ` + t + ` x GROUP BY sync_state` //nolint:gosec // G202: fixed table list + if t == "agent_message" || t == "tool_call" { + q = `SELECT sync_state, COUNT(*) FROM ` + t + ` x WHERE sync_state != 'pending'` + //nolint:gosec // G202: fixed table list + ` OR (1` + notAborted + `) GROUP BY sync_state` + } + rows, err := s.db.QueryContext(ctx, q) if err != nil { return nil, err } diff --git a/internal/capture/sync/sync_test.go b/internal/capture/sync/sync_test.go index 99fac8f..4bcab4a 100644 --- a/internal/capture/sync/sync_test.go +++ b/internal/capture/sync/sync_test.go @@ -172,6 +172,25 @@ func TestArtifactsExactShape(t *testing.T) { func itoa(n int64) string { b, _ := json.Marshal(n); return string(b) } +func TestAbortedTurnRowsAreNotCountedAsPending(t *testing.T) { + st, sid := cloudStore(t) + ctx := context.Background() + // The desktop marks a turn aborted when the process died inside it. + if _, err := st.DB().ExecContext(ctx, `UPDATE turn SET state = 'aborted' WHERE session_id = ?`, sid); err != nil { + t.Fatal(err) + } + o, err := st.Pending(ctx, 10) + if err != nil { + t.Fatal(err) + } + if got, want := counts(t, st)[store.SyncPending], int64(o.Len()); got != want { + t.Fatalf("pending count %d, outbox holds %d", got, want) + } + if o.Len() != 2 { + t.Fatalf("outbox holds %d rows, want the session and the checkpoint", o.Len()) + } +} + func TestHashFormulasNullsAreEmpty(t *testing.T) { r := store.OutSession{ID: "as-1", TokenTotals: "{}"} if SessionHash(r) != hashOf("as-1::::{}") {