Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 10 additions & 1 deletion internal/capture/store/outbox.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
}
Expand Down
19 changes: 19 additions & 0 deletions internal/capture/sync/sync_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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::::{}") {
Expand Down
Loading