Fix stale transport busy recovery

Author: IM.codes

src/agent/session-manager.ts

@@ -70,7 +70,15 @@ function isStoredTransportSession(record: Pick<SessionRecord, 'runtimeType' | 'a
 function shouldAutoRelaunchTransportRuntimeAfterError(
   providerError: TransportSessionRuntime['lastProviderError'],
 ): boolean {
-  return providerError?.code === PROVIDER_ERROR_CODES.CONNECTION_LOST;
+  if (!providerError) return false;
+  if (providerError.code === PROVIDER_ERROR_CODES.CONNECTION_LOST) return true;
+  // Codex SDK can occasionally keep an internal "running turn" marker even
+  // though the daemon has no active work left. Manual daemon restart clears the
+  // stale provider state; treat repeated recoverable "already busy" failures as
+  // the same relaunchable provider-wedged condition instead of leaving the UI in
+  // a bare error state forever.
+  return providerError.code === PROVIDER_ERROR_CODES.PROVIDER_ERROR
+    && /already busy|session is busy|provider is busy/i.test(providerError.message);
 }
 
 function sanitizeCodexSdkStartupModel(value: string | null | undefined): string | undefined {
@@ -1156,7 +1164,7 @@ async function recoverTransportRuntimeAfterError(
           pendingCount: runtime.pendingCount,
           activeDispatchCount: runtime.activeDispatchEntries.length,
         },
-        'Transport runtime error did not indicate provider connection loss; skipping provider relaunch',
+        'Transport runtime error did not indicate a relaunchable provider failure; skipping provider relaunch',
       );
       return false;
     }
@@ -1169,7 +1177,7 @@ async function recoverTransportRuntimeAfterError(
         providerError,
         ...preservation,
       },
-      'Transport provider connection lost — preserving queues and relaunching provider runtime',
+      'Transport provider failure — preserving queues and relaunching provider runtime',
     );
 
     const now = Date.now();
@@ -1198,8 +1206,11 @@ async function recoverTransportRuntimeAfterError(
 
     if (pendingCount > 0) {
       const queued = getResendEntries(sessionName);
+      const recoveryReason = providerError?.code === PROVIDER_ERROR_CODES.CONNECTION_LOST
+        ? 'Provider connection lost'
+        : 'Provider became stuck busy';
       timelineEmitter.emit(sessionName, 'assistant.text', {
-        text: `⏳ Provider connection lost — auto-resending ${pendingCount} queued message${pendingCount === 1 ? '' : 's'} after recovery.`,
+        text: `⏳ ${recoveryReason} — auto-resending ${pendingCount} queued message${pendingCount === 1 ? '' : 's'} after recovery.`,
         streaming: false,
         memoryExcluded: true,
       }, { source: 'daemon', confidence: 'high' });
@@ -1304,22 +1315,44 @@ async function drainTransportResendQueueIntoRuntime(
                 entry.commandId,
                 attachments.length > 0 ? attachments : undefined,
                 entry.messagePreamble,
-                sharedMetadata,
+                {
+                  ...sharedMetadata,
+                  ...(entry.timelineCommitted ? { timelineCommitted: true } : {}),
+                  ...(entry.historyCommitted ? { historyCommitted: true } : {}),
+                },
               )
               : runtime.send(
                 entry.text,
                 entry.commandId,
                 attachments.length > 0 ? attachments : undefined,
                 entry.messagePreamble,
+                {
+                  ...(entry.timelineCommitted ? { timelineCommitted: true } : {}),
+                  ...(entry.historyCommitted ? { historyCommitted: true } : {}),
+                },
               ))
           : (attachments.length > 0
               ? (sharedMetadata
-                  ? runtime.send(entry.text, entry.commandId, attachments, undefined, sharedMetadata)
-                  : runtime.send(entry.text, entry.commandId, attachments))
+                  ? runtime.send(entry.text, entry.commandId, attachments, undefined, {
+                    ...sharedMetadata,
+                    ...(entry.timelineCommitted ? { timelineCommitted: true } : {}),
+                    ...(entry.historyCommitted ? { historyCommitted: true } : {}),
+                  })
+                  : runtime.send(entry.text, entry.commandId, attachments, undefined, {
+                    ...(entry.timelineCommitted ? { timelineCommitted: true } : {}),
+                    ...(entry.historyCommitted ? { historyCommitted: true } : {}),
+                  }))
               : (sharedMetadata
-                  ? runtime.send(entry.text, entry.commandId, undefined, undefined, sharedMetadata)
-                  : runtime.send(entry.text, entry.commandId)));
-        if (result === 'sent') {
+                  ? runtime.send(entry.text, entry.commandId, undefined, undefined, {
+                    ...sharedMetadata,
+                    ...(entry.timelineCommitted ? { timelineCommitted: true } : {}),
+                    ...(entry.historyCommitted ? { historyCommitted: true } : {}),
+                  })
+                  : runtime.send(entry.text, entry.commandId, undefined, undefined, {
+                    ...(entry.timelineCommitted ? { timelineCommitted: true } : {}),
+                    ...(entry.historyCommitted ? { historyCommitted: true } : {}),
+                  })));
+        if (result === 'sent' && !entry.timelineCommitted) {
           timelineEmitter.emit(
             sessionName,
             'user.message',
@@ -1341,6 +1374,14 @@ async function drainTransportResendQueueIntoRuntime(
             pendingMessageEntries: runtime.pendingEntries,
             pendingMessageVersion: observeTransportQueueRevision(sessionName, runtime.pendingVersion),
           }, { source: 'daemon', confidence: 'high' });
+        } else if (result === 'sent') {
+          timelineEmitter.emit(sessionName, 'session.state', {
+            state: 'running',
+            pendingCount: runtime.pendingCount,
+            pendingMessages: runtime.pendingMessages,
+            pendingMessageEntries: runtime.pendingEntries,
+            pendingMessageVersion: observeTransportQueueRevision(sessionName, runtime.pendingVersion),
+          }, { source: 'daemon', confidence: 'high' });
         } else if (result === 'queued') {
           timelineEmitter.emit(sessionName, 'session.state', {
             state: 'queued',
@@ -1422,6 +1463,8 @@ function wireTransportCallbacks(runtime: TransportSessionRuntime, sessionName: s
       payload.pendingMessages = runtime.pendingMessages;
       payload.pendingMessageEntries = runtime.pendingEntries;
       payload.pendingMessageVersion = observeTransportQueueRevision(sessionName, runtime.pendingVersion);
+    } else if (mapped === 'error' && runtime.lastProviderError?.message) {
+      payload.error = runtime.lastProviderError.message;
     }
     timelineEmitter.emit(sessionName, 'session.state', payload, { source: 'daemon', confidence: 'high' });
     if (status === 'error') {
@@ -1448,7 +1491,7 @@ function wireTransportCallbacks(runtime: TransportSessionRuntime, sessionName: s
         { source: 'daemon', confidence: 'high', eventId: transportUserEventId(entry.clientMessageId) },
       );
     }
-    if (messages.length === 0) {
+    if (messages.length === 0 && count === 0) {
       timelineEmitter.emit(sessionName, 'user.message', { text: merged, batchedCount: count, allowDuplicate: true, pendingMessageVersion: drainedVersion });
     }
     // Include authoritative pending state after drain. The drained messages have

src/agent/transport-session-runtime.ts

@@ -59,6 +59,15 @@ export interface PendingTransportMessage {
   attachments?: TransportAttachment[];
   /** Server-authored share actor for attribution only; never injected into provider prompts. */
   sharedActor?: SharedActorEnvelope;
+  /** @internal: this logical user event has already been written to the timeline. */
+  timelineCommitted?: boolean;
+  /** @internal: this logical user event has already been written to runtime history. */
+  historyCommitted?: boolean;
+}
+
+function publicPendingEntry(entry: PendingTransportMessage): PendingTransportMessage {
+  const { timelineCommitted: _timelineCommitted, historyCommitted: _historyCommitted, ...publicEntry } = entry;
+  return publicEntry;
 }
 
 export interface TransportSendMetadata {
@@ -71,6 +80,10 @@ export interface TransportSendMetadata {
    * the tail of the FIFO queue.
    */
   queuePlacement?: 'normal' | 'front';
+  /** @internal: set when replaying entries that already have a visible user.message. */
+  timelineCommitted?: boolean;
+  /** @internal: set when replaying entries that already exist in runtime history. */
+  historyCommitted?: boolean;
 }
 
 export interface TransportRuntimeDiagnosticSnapshot {
@@ -115,6 +128,11 @@ const MAX_TRANSPORT_STALE_PENDING_RECOVERY_MS = 30 * 60_000;
 const MIN_TRANSPORT_STALE_PENDING_CANCEL_FALLBACK_MS = 50;
 const MAX_TRANSPORT_STALE_PENDING_CANCEL_FALLBACK_MS = 60_000;
 
+function isRecoverableProviderBusyError(error: ProviderError): boolean {
+  return error.code === PROVIDER_ERROR_CODES.PROVIDER_ERROR
+    && /already busy|session is busy|provider is busy/i.test(error.message);
+}
+
 type TimeoutOutcome<T> =
   | { timedOut: false; value: T }
   | { timedOut: true };
@@ -638,9 +656,13 @@ export class TransportSessionRuntime implements SessionRuntime {
   /** Snapshot of queued messages waiting to be drained (legacy text-only view). */
   get pendingMessages(): string[] { return this._pendingMessages.map((entry) => entry.text); }
   /** Snapshot of queued messages waiting to be drained (stable entity ids for UI/edit/undo). */
-  get pendingEntries(): PendingTransportMessage[] { return this._pendingMessages.map((entry) => ({ ...entry })); }
+  get pendingEntries(): PendingTransportMessage[] { return this._pendingMessages.map(publicPendingEntry); }
+  /** Snapshot of queued messages for internal resend preservation, including idempotency markers. */
+  get pendingEntriesForResend(): PendingTransportMessage[] { return this._pendingMessages.map((entry) => ({ ...entry })); }
   /** Snapshot of the message entries currently being dispatched. */
-  get activeDispatchEntries(): PendingTransportMessage[] { return this._activeDispatchEntries.map((entry) => ({ ...entry })); }
+  get activeDispatchEntries(): PendingTransportMessage[] { return this._activeDispatchEntries.map(publicPendingEntry); }
+  /** Snapshot of active entries for internal resend preservation, including idempotency markers. */
+  get activeDispatchEntriesForResend(): PendingTransportMessage[] { return this._activeDispatchEntries.map((entry) => ({ ...entry })); }
 
   getDiagnosticSnapshot(nowMs: number = Date.now()): TransportRuntimeDiagnosticSnapshot {
     let providerDiagnostics: Record<string, unknown> | null | undefined;
@@ -925,6 +947,8 @@ export class TransportSessionRuntime implements SessionRuntime {
       ...(messagePreamble?.trim() ? { messagePreamble: messagePreamble.trim() } : {}),
       ...(attachments?.length ? { attachments } : {}),
       ...(metadata?.sharedActor ? { sharedActor: metadata.sharedActor } : {}),
+      ...(metadata?.timelineCommitted ? { timelineCommitted: true } : {}),
+      ...(metadata?.historyCommitted ? { historyCommitted: true } : {}),
     };
 
     if (this.hasActiveTurnWork()) {
@@ -937,6 +961,12 @@ export class TransportSessionRuntime implements SessionRuntime {
       return 'queued';
     }
 
+    // Direct sends are rendered by command-handler / resend-drain after
+    // runtime.send() returns 'sent'. If this provider-side send later fails
+    // recoverably and gets retried from the queue, do not render the same
+    // logical clientMessageId a second time during retry drain.
+    entry.timelineCommitted = true;
+
     // N-R8 defense-in-depth (audit 0419d1ac-1f4) — wrap direct dispatch so a
     // synchronous prologue throw inside `_dispatchTurn` (e.g. some future
     // listener regression in `setStatus → _onStatusChange`) cannot leave
@@ -1281,15 +1311,19 @@ export class TransportSessionRuntime implements SessionRuntime {
     void promise.catch(() => {}); // prevent unhandled rejection
     this._activeTurn = { promise, resolve, reject };
 
-    this._history.push({
-      id: randomUUID(),
-      sessionId: this._providerSessionId!,
-      kind: 'text',
-      role: 'user',
-      content: message,
-      timestamp: Date.now(),
-      status: 'complete',
-    });
+    const historyEntries = this._activeDispatchEntries.filter((entry) => !entry.historyCommitted);
+    if (historyEntries.length > 0) {
+      this._history.push({
+        id: randomUUID(),
+        sessionId: this._providerSessionId!,
+        kind: 'text',
+        role: 'user',
+        content: historyEntries.map((entry) => entry.text).join('\n\n'),
+        timestamp: Date.now(),
+        status: 'complete',
+      });
+      for (const entry of historyEntries) entry.historyCommitted = true;
+    }
 
     this.setStatus('thinking');
 
@@ -1498,6 +1532,21 @@ export class TransportSessionRuntime implements SessionRuntime {
           }
           this._recoverableDispatchRetries = 0;
           this.clearRecoverableRetryTimer();
+          if (providerError.recoverable
+            && isRecoverableProviderBusyError(providerError)
+            && this._activeDispatchEntries.length > 0) {
+            // The provider repeatedly claimed "already busy" until the retry
+            // budget exhausted. This is usually a stale provider-side busy
+            // marker, not real daemon work. Keep the logical messages queued so
+            // session-manager can preserve them to resend before relaunching the
+            // provider runtime. Do NOT drain into the same wedged provider and
+            // do NOT drop the active entries.
+            this._pendingMessages.unshift(...this._activeDispatchEntries);
+            this._pendingVersion++;
+            this._activeDispatchEntries = [];
+            this.setStatus('error');
+            return;
+          }
           this._activeDispatchEntries = [];
           if (this._drainPending()) return;
           // Cancellation → idle (the user stopped). Recoverable budget exhausted
@@ -1531,6 +1580,8 @@ export class TransportSessionRuntime implements SessionRuntime {
     this.clearRecoverableRetryTimer();
 
     const messages = this._pendingMessages.splice(0);
+    const timelineMessages = messages.filter((entry) => !entry.timelineCommitted);
+    for (const entry of timelineMessages) entry.timelineCommitted = true;
     // Bump the queue version the moment the queue empties. The onDrain
     // callback below emits this new version on both the per-entry
     // `user.message` events and the cleared `session.state`, so a stale
@@ -1551,7 +1602,7 @@ export class TransportSessionRuntime implements SessionRuntime {
     // `_pendingMessages` already spliced empty — runtime stuck forever,
     // user-visible as bug 2 "bot stays asleep".
     try {
-      this._onDrain?.(messages, merged, messages.length);
+      this._onDrain?.(timelineMessages.map(publicPendingEntry), merged, messages.length);
     } catch (err) {
       logger.warn(
         { err, providerSessionId: this._providerSessionId, count: messages.length },

src/daemon/transport-resend-preservation.ts

@@ -23,6 +23,8 @@ function preserveEntries(
       commandId: entry.clientMessageId,
       ...(entry.attachments?.length ? { attachments: entry.attachments } : {}),
       ...(entry.sharedActor ? { sharedActor: entry.sharedActor } : {}),
+      ...(entry.timelineCommitted ? { timelineCommitted: true } : {}),
+      ...(entry.historyCommitted ? { historyCommitted: true } : {}),
       queuedAt: Date.now(),
     });
     seenCommandIds.add(entry.clientMessageId);
@@ -35,8 +37,8 @@ export function preserveTransportRuntimeQueuesToResend(
   sessionName: string,
   runtime: TransportSessionRuntime,
 ): TransportRuntimeQueuePreservationResult {
-  const activeEntries = runtime.activeDispatchEntries ?? [];
-  const pendingEntries = runtime.pendingEntries ?? [];
+  const activeEntries = runtime.activeDispatchEntriesForResend ?? runtime.activeDispatchEntries ?? [];
+  const pendingEntries = runtime.pendingEntriesForResend ?? runtime.pendingEntries ?? [];
   const beforeCount = getResendCount(sessionName);
   const seenCommandIds = new Set(getResendEntries(sessionName).map((entry) => entry.commandId));
   const preservedActiveCount = preserveEntries(sessionName, activeEntries, seenCommandIds);

src/daemon/transport-resend-queue.ts

@@ -43,6 +43,10 @@ export interface ResendEntry {
   attachments?: TransportAttachment[];
   /** Server-authored share actor for attribution only; never injected into provider prompts. */
   sharedActor?: SharedActorEnvelope;
+  /** @internal: this logical user event has already been written to the timeline. */
+  timelineCommitted?: boolean;
+  /** @internal: this logical user event has already been written to runtime history. */
+  historyCommitted?: boolean;
   /** Enqueue timestamp for expiry calculation. */
   queuedAt: number;
 }