fix(sessions): stop a session pick from blocking command handling

src/app/managers/summary-aggregation-manager.ts

@@ -361,6 +361,11 @@ function normalizeSnapshotValue(value: unknown): unknown {
   return value;
 }
 
+interface DeferredOutboundTask {
+  key?: string;
+  task: () => void | Promise<void>;
+}
+
 export class SummaryAggregator {
   private currentSessionId: string | null = null;
   private textMessageStates: Map<string, TextMessageState> = new Map();
@@ -392,8 +397,17 @@ export class SummaryAggregator {
   private onSessionDiffCallback: SessionDiffCallback | null = null;
   private onFileChangeCallback: FileChangeCallback | null = null;
   private onClearedCallback: ClearedCallback | null = null;
+  // The hold and its queue outlive a session clear: they belong to the outbound-hold
+  // lifecycle (session picks and lost-run endings), not to the session state. The array and
+  // the coalescing map are only ever changed together.
   private outboundHeld = false;
-  private deferredOutbound: Array<() => void | Promise<void>> = [];
+  private outboundHoldGeneration = 0;
+  private outboundIdleWaiters: Array<() => void> = [];
+  private deferredOutbound: DeferredOutboundTask[] = [];
+  // Coalescable tasks, last-wins: a held stream keeps one partial-text task per message
+  // instead of one per delta.
+  private deferredOutboundByKey = new Map<string, DeferredOutboundTask>();
+  private drainPromise: Promise<void> | null = null;
   private processedToolStates: Set<string> = new Set();
   private thinkingFiredForMessages: Set<string> = new Set();
   private thinkingFinishedForMessages: Set<string> = new Set();
@@ -559,34 +573,99 @@ export class SummaryAggregator {
     this.liveTurnStartedAt ??= Date.now();
   }
 
-  holdOutbound(): void {
+  holdOutbound(): number {
     this.outboundHeld = true;
+    this.outboundHoldGeneration += 1;
+    return this.outboundHoldGeneration;
   }
 
   hasDeferredOutbound(): boolean {
     return this.deferredOutbound.length > 0;
   }
 
-  async drainOutbound(gapMs: number): Promise<void> {
-    if (!this.outboundHeld) {
+  /**
+   * Replays the held outbound queue for a hold generation. Only one drain runs at a time:
+   * a concurrent caller awaits the running drain and then re-checks its own generation.
+   * A drain stops when a newer hold starts, leaving the queue to that hold's own release.
+   */
+  async drainOutbound(gapMs: number, generation: number): Promise<void> {
+    if (!this.outboundHeld || generation !== this.outboundHoldGeneration) {
+      return;
+    }
+
+    if (this.drainPromise) {
+      await this.drainPromise;
+      await this.drainOutbound(gapMs, generation);
       return;
     }
 
+    const drain = this.runDrain(generation, gapMs).finally(() => {
+      if (this.drainPromise === drain) {
+        this.drainPromise = null;
+      }
+    });
+    this.drainPromise = drain;
+    await drain;
+  }
+
+  /**
+   * Resolves once the held queue is empty, whoever drains it: a superseded caller can wait
+   * for its output without taking over a newer hold.
+   */
+  whenOutboundIdle(): Promise<void> {
+    if (!this.outboundHeld) {
+      return Promise.resolve();
+    }
+
+    return new Promise((resolve) => {
+      this.outboundIdleWaiters.push(resolve);
+    });
+  }
+
+  private markOutboundIdle(): void {
+    this.outboundHeld = false;
+    const waiters = this.outboundIdleWaiters;
+    this.outboundIdleWaiters = [];
+    for (const resolve of waiters) {
+      resolve();
+    }
+  }
+
+  private async runDrain(generation: number, gapMs: number): Promise<void> {
     let first = true;
     for (;;) {
-      const task = this.deferredOutbound.shift();
-      if (!task) {
-        this.outboundHeld = false;
+      if (this.outboundHoldGeneration !== generation) {
+        return;
+      }
+
+      const next = this.deferredOutbound[0];
+      if (!next) {
+        this.markOutboundIdle();
         return;
       }
 
       if (!first) {
         await new Promise((resolve) => setTimeout(resolve, gapMs));
       }
+
+      // A newer hold can start during the gap: leave the queue to it.
+      if (this.outboundHoldGeneration !== generation) {
+        return;
+      }
+
+      const entry = this.deferredOutbound.shift();
+      if (!entry) {
+        this.markOutboundIdle();
+        return;
+      }
+
+      if (entry.key) {
+        this.deferredOutboundByKey.delete(entry.key);
+      }
       first = false;
 
       try {
-        await task();
+        await entry.task();
       } catch (err) {
         logger.error("[Aggregator] Error in deferred outbound callback:", err);
       }
@@ -598,22 +677,46 @@ export class SummaryAggregator {
    */
   runOutbound(task: () => void | Promise<void>): Promise<void> {
     if (this.outboundHeld) {
-      this.deferredOutbound.push(task);
+      this.deferOutbound(task);
       return Promise.resolve();
     }
 
     return Promise.resolve(task());
   }
 
-  private scheduleOutbound(task: () => void | Promise<void>, immediate: boolean): void {
+  private scheduleOutbound(
+    task: () => void | Promise<void>,
+    immediate: boolean,
+    coalesceKey?: string,
+  ): void {
     if (this.outboundHeld) {
-      this.deferredOutbound.push(task);
+      this.deferOutbound(task, coalesceKey);
       return;
     }
 
     this.deliverOutbound(task, immediate);
   }
 
+  /**
+   * Defers a task for the hold. A coalescable key keeps only the latest task in the slot
+   * the key first took, so a long stream cannot grow the queue by its delta count.
+   */
+  private deferOutbound(task: () => void | Promise<void>, coalesceKey?: string): void {
+    if (coalesceKey) {
+      const existing = this.deferredOutboundByKey.get(coalesceKey);
+      if (existing) {
+        existing.task = task;
+        return;
+      }
+    }
+
+    const entry: DeferredOutboundTask = coalesceKey ? { key: coalesceKey, task } : { task };
+    if (coalesceKey) {
+      this.deferredOutboundByKey.set(coalesceKey, entry);
+    }
+    this.deferredOutbound.push(entry);
+  }
+
   /**
    * Delivers past the hold: the dashboard's values (tokens, cost, files) are never held, only
    * the redraws and messages that show them.
@@ -2402,9 +2505,13 @@ export class SummaryAggregator {
     }
 
     const callback = this.onThinkingCallback;
-    this.scheduleOutbound(() => {
-      callback({ sessionId, messageId, sections, isFirstUpdate });
-    }, true);
+    this.scheduleOutbound(
+      () => {
+        callback({ sessionId, messageId, sections, isFirstUpdate });
+      },
+      true,
+      `thinking:${sessionId}:${messageId}:${isFirstUpdate ? "first" : "next"}`,
+    );
   }
 
   private emitThinkingFinishedOnce(sessionId: string, messageId: string): void {
@@ -2475,13 +2582,17 @@ export class SummaryAggregator {
     }
 
     const callback = this.onPartialCallback;
-    this.scheduleOutbound(() => {
-      try {
-        callback(sessionId, messageId, messageText);
-      } catch (err) {
-        logger.error("[Aggregator] Error in partial callback:", err);
-      }
-    }, false);
+    this.scheduleOutbound(
+      () => {
+        try {
+          callback(sessionId, messageId, messageText);
+        } catch (err) {
+          logger.error("[Aggregator] Error in partial callback:", err);
+        }
+      },
+      false,
+      `partial:${sessionId}:${messageId}`,
+    );
   }
 
   private getOrCreateTextMessageState(messageID: string): TextMessageState {

src/bot/callbacks/session-callback-handler.ts

@@ -143,11 +143,12 @@ export async function selectSessionById(
 
   const deliverPick = options.postSelectAction === "preview" && Boolean(ctx.chat);
   let pickMessages: SessionPickMessage[] | null = null;
+  let pickGeneration: number | null = null;
   let pickBusy = false;
   if (deliverPick) {
     pickBusy = await readSessionBusy(session.id, currentProject.worktree);
     pickMessages = await loadSessionPickMessages(session.id, currentProject.worktree, pickBusy);
-    deps.summaryAggregator.holdOutbound();
+    pickGeneration = deps.summaryAggregator.holdOutbound();
   }
 
   try {
@@ -223,8 +224,8 @@ export async function selectSessionById(
       }
     }
   } finally {
-    if (deliverPick) {
-      await releaseSessionPickHold(deps.summaryAggregator);
+    if (deliverPick && pickGeneration !== null) {
+      releaseSessionPickHold(deps.summaryAggregator, pickGeneration);
     }
   }
 
@@ -379,13 +380,27 @@ function waitForSessionPickSend(): Promise<void> {
   return new Promise((resolve) => setTimeout(resolve, sessionPickSendGapMs()));
 }
 
-async function releaseSessionPickHold(
+/**
+ * Releases the outbound hold in the background: the deferred live-stream updates are
+ * replayed at a send gap, and awaiting that here would block grammY's sequential update
+ * loop for as long as the picked session keeps streaming.
+ */
+function releaseSessionPickHold(
   aggregator: SessionSelectDeps["summaryAggregator"],
-): Promise<void> {
-  if (aggregator.hasDeferredOutbound()) {
-    await waitForSessionPickSend();
-  }
-  await aggregator.drainOutbound(sessionPickSendGapMs());
+  generation: number,
+): void {
+  safeBackgroundTask({
+    taskName: "sessions.releasePickOutbound",
+    task: async () => {
+      try {
+        if (aggregator.hasDeferredOutbound()) {
+          await waitForSessionPickSend();
+        }
+      } finally {
+        await aggregator.drainOutbound(sessionPickSendGapMs(), generation);
+      }
+    },
+  });
 }
 
 async function readSessionBusy(sessionId: string, directory: string): Promise<boolean> {

src/bot/services/event-subscription-service.ts

@@ -206,29 +206,40 @@ class EventSubscriptionService implements BotEventSubscriptionService {
     const { summaryAggregator } = deps;
     summaryAggregator.forgetLiveTurn();
 
-    summaryAggregator.holdOutbound();
-    // Every subagent card still running ends with the server and is no longer followed; one
-    // the server resumes gets a new card.
-    summaryAggregator.interruptSubagents(true);
-
-    const sessionId = getCurrentSession()?.id;
-    const runningTools = sessionId ? runtime.getRunningToolInfos(sessionId) : [];
-    if (sessionId && (deps.assistantRunState.hasRun(sessionId) || runningTools.length > 0)) {
-      logger.info(
-        `[Bot] Ending the run lost with the OpenCode server: session=${sessionId}, runningTools=${runningTools.length}, reason=${reason}`,
-      );
-      deps.assistantRunState.clearRun(sessionId, reason);
-
-      const now = Date.now();
-      for (const tool of runningTools) {
-        summaryAggregator.processEvent(buildFailedToolEvent(tool, now));
+    const holdGeneration = summaryAggregator.holdOutbound();
+    try {
+      // Every subagent card still running ends with the server and is no longer followed; one
+      // the server resumes gets a new card.
+      summaryAggregator.interruptSubagents(true);
+
+      const sessionId = getCurrentSession()?.id;
+      const runningTools = sessionId ? runtime.getRunningToolInfos(sessionId) : [];
+      if (sessionId && (deps.assistantRunState.hasRun(sessionId) || runningTools.length > 0)) {
+        logger.info(
+          `[Bot] Ending the run lost with the OpenCode server: session=${sessionId}, runningTools=${runningTools.length}, reason=${reason}`,
+        );
+        deps.assistantRunState.clearRun(sessionId, reason);
+
+        const now = Date.now();
+        for (const tool of runningTools) {
+          summaryAggregator.processEvent(buildFailedToolEvent(tool, now));
+        }
+        for (const event of buildInterruptedIdleEvents(sessionId)) {
+          summaryAggregator.processEvent(event);
+        }
       }
-      for (const event of buildInterruptedIdleEvents(sessionId)) {
-        summaryAggregator.processEvent(event);
-      }
-    }
 
-    await Promise.all([pollClosed, summaryAggregator.drainOutbound(0)]);
+      await Promise.all([
+        pollClosed,
+        summaryAggregator.drainOutbound(0, holdGeneration),
+        summaryAggregator.whenOutboundIdle(),
+      ]);
+    } finally {
+      // Best-effort release: a throw before the drain must not leave the hold set.
+      void summaryAggregator.drainOutbound(0, holdGeneration).catch((error) => {
+        logger.warn("[Bot] Failed to release the ending outbound:", error);
+      });
+    }
   };
 
   private restoreAfterReconnect({ serverRestarted }: ReconnectInfo): void {