Retry finished-session uploads from desktop and terminal activity

src/db.ts

@@ -206,12 +206,16 @@ export async function claimSession(
   });
 }
 
-export async function updateSession(id: string, updates: Partial<Session>): Promise<void> {
+export async function updateSession(id: string, updates: Partial<Session>, expected?: Session[]): Promise<void> {
   await withLock(() => {
     const data = readDb();
-    const session = data.sessions.find((s) => s.id === id);
-    if (session) {
-      Object.assign(session, updates);
+    const matches = data.sessions.filter((s) => s.id === id);
+    // An upload settles the complete unchanged group, including duplicates
+    // from older CLIs. Never let an old response settle a corrected version.
+    if (expected && JSON.stringify(matches) !== JSON.stringify(expected)) return;
+    const targets = expected ? matches : matches.slice(0, 1);
+    if (targets.length) {
+      for (const session of targets) Object.assign(session, updates);
       writeDb(data);
     }
   });
@@ -236,11 +240,12 @@ export async function deleteSession(id: string): Promise<void> {
 // an idle desktop session alive.
 const HEARTBEAT_MS = Math.max(3 * TUNABLES.pollIntervalMs, 2 * 60_000);
 
-export async function reapOrphanedSessions(): Promise<void> {
-  await withLock(() => {
+// Returns how many sessions it finalized, so a caller can send them on.
+export async function reapOrphanedSessions(): Promise<number> {
+  return withLock(() => {
     const data = readDb();
     const now = Date.now();
-    let changed = false;
+    let reaped = 0;
 
     for (const s of data.sessions) {
       if (s.exitCode !== -1) continue;
@@ -268,10 +273,11 @@ export async function reapOrphanedSessions(): Promise<void> {
         s.reapedAt = s.endedAt;
         s.momentum = scoreReaped(s, readConfig());
       }
-      changed = true;
+      reaped++;
     }
 
-    if (changed) writeDb(data);
+    if (reaped) writeDb(data);
+    return reaped;
   });
 }
 

src/hook.ts

@@ -159,7 +159,10 @@ async function onSessionStart(sessionId: string, cwd: string, tool: HookTool): P
   // activity events: those fire on every tool call.
   reconcileInstall();
   await refreshTunables(HOOK_NETWORK_BUDGET_MS).catch(() => {});
-  await refreshAndReap();
+  // A desktop session left open is only ever finalized here, by the reaper, and
+  // it would otherwise wait for some other session's clean end to be sent.
+  // Only when something settled, so a quiet start spends no time on the network.
+  if (await refreshAndReap()) await flushPendingSubmissions(HOOK_NETWORK_BUDGET_MS).catch(() => {});
 
   // SessionStart can also fire when an existing conversation is resumed — key
   // off the provider's session id so a session is only opened once.
@@ -252,10 +255,7 @@ async function onActivity(sessionId: string, cwd: string): Promise<void> {
 // marked before sending so a failure waits for the next window instead of
 // looping; the signature only once the server has it, so a failure is resent.
 async function submitProgress(session: Session): Promise<void> {
-  if (session.exitCode !== -1 || session.momentum !== 'shipped') return;
-
-  const signature = progressSignature(session);
-  if (signature === session.acceptedProgressSignature) return;
+  if (session.exitCode !== -1) return;
 
   const last = session.lastProgressSubmitAt ? Date.parse(session.lastProgressSubmitAt) : 0;
   if (last && Date.now() - last < IN_PROGRESS_SUBMIT_INTERVAL_MS) return;
@@ -264,6 +264,16 @@ async function submitProgress(session: Session): Promise<void> {
     await updateSession(session.id, { lastProgressSubmitAt: new Date().toISOString() });
   } catch { return; }
 
+  // A settled session normally waits for a clean end, but when every endcard
+  // flush of a day fails — one rate-limit window is enough — nothing ever
+  // retriggers: desktop sessions can stay open for days. This one is live and
+  // already on the network, so it carries the settled queue with it.
+  await flushPendingSubmissions(HOOK_NETWORK_BUDGET_MS).catch(() => {});
+
+  if (session.momentum !== 'shipped') return;
+  const signature = progressSignature(session);
+  if (signature === session.acceptedProgressSignature) return;
+
   if (await submitInProgress(session, HOOK_NETWORK_BUDGET_MS).catch(() => false)) {
     try { await updateSession(session.id, { acceptedProgressSignature: signature }); } catch {}
   }

src/rescore.ts

@@ -47,11 +47,15 @@ function ownedByAnotherSession(session: Session, all: Session[]): boolean {
 // and the commit is never read: the session is submitted with zero. Refreshing
 // before the reaper runs means the work is recorded whether or not you came back
 // to end it cleanly.
-export async function refreshRecentSessions(): Promise<void> {
+//
+// Returns how many closed sessions it changed: those are back in the submit
+// queue, while an open one waits for its own end.
+export async function refreshRecentSessions(): Promise<number> {
   const now = Date.now();
   const all = getSessions();
   const config = readConfig();
   const identities = commitIdentities();
+  let reopened = 0;
 
   for (const session of all) {
     // Sessions from an older CLI have no repo baselines to measure against.
@@ -76,12 +80,16 @@ export async function refreshRecentSessions(): Promise<void> {
       // resubmitting is safe.
       submittedAt: undefined,
     });
+    if (!open) reopened++;
   }
+  return reopened;
 }
 
 // Refresh first, then reap: a session the reaper is about to close carries the
 // work it actually did rather than whatever the last hook event happened to see.
-export async function refreshAndReap(): Promise<void> {
-  await refreshRecentSessions();
-  await reapOrphanedSessions();
+// Returns how many sessions it closed or corrected after close, all of them now
+// waiting to be submitted.
+export async function refreshAndReap(): Promise<number> {
+  const refreshed = await refreshRecentSessions();
+  return refreshed + (await reapOrphanedSessions());
 }

src/submit.ts

@@ -1,3 +1,6 @@
+import { mkdirSync, rmdirSync, readFileSync, writeFileSync, statSync, unlinkSync } from 'node:fs';
+import { join } from 'node:path';
+import { VIBE_DIR, ensureVibeDir } from './config.js';
 import { createHash, createHmac } from 'node:crypto';
 import { getSessions, updateSession, type Session } from './db.js';
 import { readAuth, markSignedOut, refreshAuth, jwtExpiresAtMs, rememberCommitKey, type AuthRecord } from './auth.js';
@@ -141,9 +144,9 @@ export async function submitInProgress(session: Session, budgetMs = 1500): Promi
 // Deduped by id: parallel hook processes could both pass the existence check
 // and write the same session twice, and a flush would then upload the empty
 // copy over the real one. More commits wins, longer session breaks the tie.
-function pendingSessions(): Session[] {
+function pendingSessions(sessions = getSessions()): Session[] {
   const byId = new Map<string, Session>();
-  for (const s of getSessions()) {
+  for (const s of sessions) {
     if (s.submittedAt || s.exitCode === -1 || s.durationSeconds < 60) continue;
     const prev = byId.get(s.id);
     if (!prev || s.commits > prev.commits || (s.commits === prev.commits && s.durationSeconds > prev.durationSeconds)) {
@@ -162,54 +165,139 @@ export function pendingSubmissionCount(): number {
   return pendingSessions().length;
 }
 
+const FLUSH_LOCK = join(VIBE_DIR, 'submit.lock');
+const FLUSH_BREAK = FLUSH_LOCK + '.break';
+
+function staleFlushLock(): boolean {
+  try {
+    const pid = Number(readFileSync(join(FLUSH_LOCK, 'pid'), 'utf8'));
+    if (Number.isInteger(pid) && pid > 0) {
+      try { process.kill(pid, 0); return false; }
+      catch (err) { return (err as NodeJS.ErrnoException).code === 'ESRCH'; }
+    }
+  } catch {}
+  try { return Date.now() - statSync(FLUSH_LOCK).mtimeMs > 10_000; }
+  catch { return false; }
+}
+
+function releaseFlushLock(): void {
+  try { unlinkSync(join(FLUSH_LOCK, 'pid')); } catch {}
+  try { rmdirSync(FLUSH_LOCK); } catch {}
+}
+
+function claimFlushLock(retried = false): boolean {
+  try { mkdirSync(FLUSH_LOCK); }
+  catch {
+    if (retried || !staleFlushLock()) return false;
+    try { mkdirSync(FLUSH_BREAK); } catch { return false; }
+    try {
+      if (staleFlushLock()) releaseFlushLock();
+      return claimFlushLock(true);
+    } finally {
+      try { rmdirSync(FLUSH_BREAK); } catch {}
+    }
+  }
+  try {
+    writeFileSync(join(FLUSH_LOCK, 'pid'), String(process.pid));
+    return true;
+  } catch {
+    releaseFlushLock();
+    return false;
+  }
+}
+
+// When each queued session was last sent, kept beside the queue rather than in
+// sessions.json so recording it never changes the snapshot an upload settles.
+// Only the submit.lock owner reads or writes it.
+const ATTEMPTS_PATH = join(VIBE_DIR, 'submit-attempts.json');
+
+function readAttempts(): Record<string, number> {
+  try {
+    const value: unknown = JSON.parse(readFileSync(ATTEMPTS_PATH, 'utf8'));
+    if (value && typeof value === 'object' && !Array.isArray(value)) return value as Record<string, number>;
+  } catch {}
+  return {};
+}
+
+function writeAttempts(attempts: Record<string, number>): void {
+  try { writeFileSync(ATTEMPTS_PATH, JSON.stringify(attempts)); } catch {}
+}
+
 export async function flushPendingSubmissions(budgetMs: number): Promise<void> {
   const deadline = Date.now() + budgetMs;
+  if (budgetMs <= 0 || pendingSessions().length === 0 || !readAuth()) return;
+  ensureVibeDir();
+  // Nonblocking: another agent is draining the same queue. Its database writes
+  // remain short; never hold sessions.lock across a network request.
+  if (!claimFlushLock()) return;
+  try {
+    await flushPendingLocked(deadline);
+  } finally {
+    releaseFlushLock();
+  }
+}
+
+async function flushPendingLocked(deadline: number): Promise<void> {
+  // Read after claiming the lock, so a waiter never sends an already-drained
+  // snapshot. Empty queues do not renew auth or contact the network.
+  // Least recently attempted first: a session whose upload always outlasts the
+  // hook budget would otherwise stop every drain at the front of the queue, and
+  // nothing behind it would ever go out.
+  const attempts = readAttempts();
+  const pending = pendingSessions().sort((a, b) =>
+    (attempts[a.id] ?? 0) - (attempts[b.id] ?? 0) || a.startedAt.localeCompare(b.startedAt));
+  const budgetMs = deadline - Date.now();
+  if (pending.length === 0 || budgetMs <= 0) return;
+  // Drop entries for sessions that have left the queue.
+  for (const id of Object.keys(attempts)) if (!pending.some((p) => p.id === id)) delete attempts[id];
 
-  // Renewed here when near expiry; a failed renewal falls through with the
-  // current jwt, which may still be valid.
   let auth: AuthRecord | null = await currentAuth(budgetMs);
   if (!auth) return;
 
-  const pending = pendingSessions().sort((a, b) => a.startedAt.localeCompare(b.startedAt));
-
   let renewedOnce = false;
   for (let i = 0; i < pending.length; i++) {
-    const session = pending[i];
+    // A hook can reopen, correct or delete a queued session while an earlier
+    // request is in flight. Send its current settled version, if it still has one.
+    const snapshots = getSessions().filter((s) => s.id === pending[i].id);
+    const session = pendingSessions(snapshots)[0];
+    if (!session) continue;
     const remaining = deadline - Date.now();
     if (remaining <= 0) return;
     const perRequest = Math.min(remaining, 1500);
+    // Recorded before sending: a hook process can be cut off mid-request.
+    attempts[session.id] = Date.now();
+    writeAttempts(attempts);
     try {
       await postSession(session, auth, perRequest);
-      await updateSession(session.id, { submittedAt: new Date().toISOString() });
+      await updateSession(session.id, { submittedAt: new Date().toISOString() }, snapshots);
     } catch (e) {
       if (e instanceof ApiError) {
         if (e.status === 401) {
           if (auth.refreshToken && !renewedOnce) {
             renewedOnce = true;
-            const renewed = await refreshAuth(auth, Math.min(deadline - Date.now(), 3000));
-            if (!renewed) return; // refresh rejected: local auth already cleared
+            const renewalBudget = deadline - Date.now();
+            if (renewalBudget <= 0) return;
+            const renewed = await refreshAuth(auth, Math.min(renewalBudget, 3000));
+            if (!renewed) return;
             if (renewed.jwt !== auth.jwt) {
               auth = renewed;
-              i--; // retry this session with the fresh token
+              i--;
               continue;
             }
-            return; // couldn't reach the server to renew: retry next flush
+            return;
           }
-          // no refresh token (pre-v0.7 login) or still 401 after renewing:
-          // drop the token, leave sessions unsubmitted so a fresh login retries them
           markSignedOut();
           return;
         }
         if (e.status === 400) {
-          // validation failure: nothing about waiting will fix this. stop retrying this one.
-          await updateSession(session.id, { submittedAt: new Date().toISOString() });
+          // Settle only the rejected snapshot, never a correction made while
+          // it was in flight. The corrected payload may be valid.
+          await updateSession(session.id, { submittedAt: new Date().toISOString() }, snapshots);
           continue;
         }
-        // 403 (account too new), 429 (rate limit), 5xx: retry later
-        return;
+        return; // 403, 429 and 5xx remain pending for the next attempt.
       }
-      // network error: retry later
-      return;
+      return; // network or local persistence failure: leave work pending.
     }
   }
 }

src/wrap.ts

@@ -120,8 +120,10 @@ export async function wrapTool(tool: string, args: string[]): Promise<void> {
   // repo is still owed the grace-window work, and an open session here would
   // claim it instead. Never rejects — a fast-exiting tool must still find its
   // session to finalize.
+  let startupFlush: Promise<void> = Promise.resolve();
   const ready: Promise<Session> = (async () => {
-    await refreshAndReap().catch(() => {});
+    const settled = await refreshAndReap().catch(() => 0);
+    if (settled) startupFlush = flushPendingSubmissions(IN_PROGRESS_SUBMIT_BUDGET_MS).catch(() => {});
     let initial = snapshot(-1);
     let session: Session = {
       id: sessionId, tool, project, branch, startedAt,
@@ -163,7 +165,9 @@ export async function wrapTool(tool: string, args: string[]): Promise<void> {
   // periodic update while the tool runs — track activity for duration accuracy
   let lastInProgressSubmitAt = 0;
   let acceptedSignature = '';
-  const poll = setInterval(async () => {
+  let cleaned = false;
+  let pollInFlight: Promise<void> | undefined;
+  async function pollOnce(): Promise<void> {
     try {
       const session = await ready;
       const now = Date.now();
@@ -197,34 +201,37 @@ export async function wrapTool(tool: string, args: string[]): Promise<void> {
       // reopen would keep the flush from ever sending what came after.
       await updateSession(sessionId, { ...snap, submittedAt: undefined });
 
-      if (snap.momentum === 'shipped') {
-        // Only resubmit when the payload actually changed. An idle open session
-        // otherwise reposts an identical row every 5 minutes forever, and a
-        // developer running a dozen sessions at once burned the server's rate
-        // guard on nothing, losing the submissions that did matter.
-        // A failed send leaves the signature unaccepted, so the next window
-        // resends it; the throttle alone keeps that from looping.
+      const due = lastInProgressSubmitAt === 0 || Date.now() - lastInProgressSubmitAt >= IN_PROGRESS_SUBMIT_INTERVAL_MS;
+      if (due) {
+        lastInProgressSubmitAt = Date.now();
+        // Finished work retries even while this tool has no new ship to send.
+        const deadline = Date.now() + IN_PROGRESS_SUBMIT_BUDGET_MS;
+        await flushPendingSubmissions(IN_PROGRESS_SUBMIT_BUDGET_MS).catch(() => {});
+        if (cleaned) return;
         const signature = progressSignature(snap);
-        const due = lastInProgressSubmitAt === 0 || Date.now() - lastInProgressSubmitAt >= IN_PROGRESS_SUBMIT_INTERVAL_MS;
-        if (due && signature !== acceptedSignature) {
-          lastInProgressSubmitAt = Date.now();
-          submitInProgress({ ...session, ...snap }, IN_PROGRESS_SUBMIT_BUDGET_MS).then(
-            (settled) => { if (settled) acceptedSignature = signature; },
-            () => {},
-          );
+        const remaining = deadline - Date.now();
+        if (remaining > 0 && snap.momentum === 'shipped' && signature !== acceptedSignature) {
+          const settled = await submitInProgress({ ...session, ...snap }, remaining).catch(() => false);
+          if (settled) acceptedSignature = signature;
         }
       }
     } catch {}
+  }
+  const poll = setInterval(() => {
+    if (cleaned || pollInFlight) return;
+    pollInFlight = pollOnce().finally(() => { pollInFlight = undefined; });
   }, POLL_INTERVAL_MS);
   poll.unref();
 
   // single cleanup path for all exit scenarios
-  let cleaned = false;
   async function finalize(exitCode: number, showEndcard: boolean): Promise<void> {
     clearInterval(poll);
-    if (cleaned) process.exit(exitCode);
+    if (cleaned) return;
     cleaned = true;
     const session = await ready;
+    // A queue drain may hold submit.lock as the child exits. Let our own
+    // bounded background work finish before the final snapshot and flush.
+    await Promise.all([startupFlush, pollInFlight]);
 
     // accumulate any trailing idle gap (only when git provides activity signals)
     const now = Date.now();