From 78393bd90819a16f89410feecb482425cc069c81 Mon Sep 17 00:00:00 2001 From: Anton Gladkov Date: Wed, 21 Oct 2026 10:03:44 +0300 Subject: [PATCH] fix(sessions): stop a session pick from blocking command handling Picking a session held the aggregator's outbound queue, deferred one task per streaming delta and replayed them at a one-second gap inside grammY's sequential update loop, so commands sent while the picked session kept streaming waited until the run ended and were then dropped as stale with the outage notice. Coalesce deferred partial-text and thinking tasks per message (last-wins) and run the pick's drain in the background, tied to its hold generation so an older release never drains a newer hold. The lost-run ending keeps its completion barrier through the idle wait. --- .../managers/summary-aggregation-manager.ts | 153 +++++++- src/bot/callbacks/session-callback-handler.ts | 33 +- .../services/event-subscription-service.ts | 53 ++- .../summary-aggregation-outbound-hold.test.ts | 360 +++++++++++++++++- tests/bot/commands/sessions.test.ts | 186 ++++++++- .../events/handlers/dashboard-handler.test.ts | 4 +- ...vent-subscription-service.lost-run.test.ts | 41 ++ 7 files changed, 771 insertions(+), 59 deletions(-) diff --git a/src/app/managers/summary-aggregation-manager.ts b/src/app/managers/summary-aggregation-manager.ts index 13af3f743..91ef0bf86 100644 --- a/src/app/managers/summary-aggregation-manager.ts +++ b/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; +} + export class SummaryAggregator { private currentSessionId: string | null = null; private textMessageStates: Map = 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> = []; + 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(); + private drainPromise: Promise | null = null; private processedToolStates: Set = new Set(); private thinkingFiredForMessages: Set = new Set(); private thinkingFinishedForMessages: Set = 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 { - 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 { + 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 { + 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 { 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): Promise { if (this.outboundHeld) { - this.deferredOutbound.push(task); + this.deferOutbound(task); return Promise.resolve(); } return Promise.resolve(task()); } - private scheduleOutbound(task: () => void | Promise, immediate: boolean): void { + private scheduleOutbound( + task: () => void | Promise, + 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, 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 { diff --git a/src/bot/callbacks/session-callback-handler.ts b/src/bot/callbacks/session-callback-handler.ts index 8d90b5758..78a01c3b0 100644 --- a/src/bot/callbacks/session-callback-handler.ts +++ b/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 { 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 { - 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 { diff --git a/src/bot/services/event-subscription-service.ts b/src/bot/services/event-subscription-service.ts index 1ecde07b4..1dcd262f4 100644 --- a/src/bot/services/event-subscription-service.ts +++ b/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 { diff --git a/tests/app/managers/summary-aggregation-outbound-hold.test.ts b/tests/app/managers/summary-aggregation-outbound-hold.test.ts index bf25cadc8..c4d19534c 100644 --- a/tests/app/managers/summary-aggregation-outbound-hold.test.ts +++ b/tests/app/managers/summary-aggregation-outbound-hold.test.ts @@ -29,6 +29,22 @@ function emitPartial(aggregator: SummaryAggregator, messageId: string, text: str } as unknown as Event); } +function emitDelta( + aggregator: SummaryAggregator, + messageId: string, + partId: string, + delta: string, + type: "text" | "reasoning" = "text", +): void { + aggregator.processEvent({ + type: "message.part.delta", + properties: { + part: { id: partId, sessionID: "session-1", messageID: messageId, type }, + delta, + }, + } as unknown as Event); +} + describe("summary aggregator outbound hold", () => { afterEach(() => { vi.useRealTimers(); @@ -53,13 +69,13 @@ describe("summary aggregator outbound hold", () => { aggregator.setOnPartial((_sessionId, _messageId, text) => { texts.push(text); }); - aggregator.holdOutbound(); + const generation = aggregator.holdOutbound(); emitPartial(aggregator, "message-1", "first"); emitPartial(aggregator, "message-2", "second"); expect(texts).toEqual([]); - const draining = aggregator.drainOutbound(1000); + const draining = aggregator.drainOutbound(1000, generation); await vi.advanceTimersByTimeAsync(0); expect(texts).toEqual(["first"]); @@ -117,11 +133,347 @@ describe("summary aggregator outbound hold", () => { await aggregator.runOutbound(task); expect(task).toHaveBeenCalledOnce(); - aggregator.holdOutbound(); + const generation = aggregator.holdOutbound(); await aggregator.runOutbound(task); expect(task).toHaveBeenCalledOnce(); - await aggregator.drainOutbound(0); + await aggregator.drainOutbound(0, generation); expect(task).toHaveBeenCalledTimes(2); }); + + it("coalesces deferred partial text into the latest payload", async () => { + const aggregator = new SummaryAggregator(); + const onPartial = vi.fn(); + aggregator.setSession("session-1"); + aggregator.setOnPartial(onPartial); + const generation = aggregator.holdOutbound(); + + emitPartial(aggregator, "message-1", "Hel"); + emitDelta(aggregator, "message-1", "message-1-part", "lo"); + emitDelta(aggregator, "message-1", "message-1-part", " world"); + + await aggregator.drainOutbound(0, generation); + + expect(onPartial).toHaveBeenCalledOnce(); + expect(onPartial).toHaveBeenCalledWith("session-1", "message-1", "Hello world"); + }); + + it("keeps the order of a coalesced partial and a plain deferred task", async () => { + const aggregator = new SummaryAggregator(); + const order: string[] = []; + aggregator.setSession("session-1"); + aggregator.setOnPartial(() => { + order.push("partial"); + }); + const generation = aggregator.holdOutbound(); + + emitPartial(aggregator, "message-1", "A"); + emitDelta(aggregator, "message-1", "message-1-part", "B"); + await aggregator.runOutbound(() => { + order.push("plain"); + }); + + await aggregator.drainOutbound(0, generation); + + expect(order).toEqual(["partial", "plain"]); + }); + + it("awaits the running drain instead of starting a second one", async () => { + const aggregator = new SummaryAggregator(); + const calls: number[] = []; + let releaseFirst!: () => void; + let secondSettled = false; + const gate = new Promise((resolve) => { + releaseFirst = resolve; + }); + + const generation = aggregator.holdOutbound(); + aggregator.runOutbound(() => gate); + aggregator.runOutbound(() => { + calls.push(2); + }); + + const firstDrain = aggregator.drainOutbound(0, generation); + await Promise.resolve(); + const secondDrain = aggregator.drainOutbound(0, generation); + void secondDrain.then(() => { + secondSettled = true; + }); + await Promise.resolve(); + expect(calls).toEqual([]); + expect(secondSettled).toBe(false); + + releaseFirst(); + await firstDrain; + await secondDrain; + expect(calls).toEqual([2]); + expect(secondSettled).toBe(true); + }); + + it("does not let a stale generation drain a newer hold", async () => { + const aggregator = new SummaryAggregator(); + const calls: string[] = []; + + const generationA = aggregator.holdOutbound(); + aggregator.runOutbound(() => { + calls.push("a"); + }); + const generationB = aggregator.holdOutbound(); + aggregator.runOutbound(() => { + calls.push("b"); + }); + + await aggregator.drainOutbound(0, generationA); + expect(calls).toEqual([]); + + await aggregator.drainOutbound(0, generationB); + expect(calls).toEqual(["a", "b"]); + }); + + it("stops a running drain when a newer hold starts", async () => { + const aggregator = new SummaryAggregator(); + const calls: string[] = []; + let releaseFirst!: () => void; + const gate = new Promise((resolve) => { + releaseFirst = resolve; + }); + + const generationA = aggregator.holdOutbound(); + aggregator.runOutbound(() => gate); + aggregator.runOutbound(() => { + calls.push("a"); + }); + + const drainA = aggregator.drainOutbound(0, generationA); + await Promise.resolve(); + + const generationB = aggregator.holdOutbound(); + aggregator.runOutbound(() => { + calls.push("b"); + }); + + releaseFirst(); + await drainA; + expect(calls).toEqual([]); + + await aggregator.drainOutbound(0, generationB); + expect(calls).toEqual(["a", "b"]); + }); + + it("resolves the idle barrier after a newer hold drains the queue", async () => { + const aggregator = new SummaryAggregator(); + let releaseFirst!: () => void; + let idleResolved = false; + const gate = new Promise((resolve) => { + releaseFirst = resolve; + }); + + const generationA = aggregator.holdOutbound(); + aggregator.runOutbound(() => gate); + const drainA = aggregator.drainOutbound(0, generationA); + await Promise.resolve(); + + const generationB = aggregator.holdOutbound(); + aggregator.runOutbound(() => {}); + const idle = aggregator.whenOutboundIdle(); + void idle.then(() => { + idleResolved = true; + }); + + releaseFirst(); + await drainA; + await Promise.resolve(); + expect(idleResolved).toBe(false); + + await aggregator.drainOutbound(0, generationB); + await idle; + expect(idleResolved).toBe(true); + }); + + it("delivers a newer payload that arrives while the keyed entry is already draining", async () => { + vi.useFakeTimers(); + const aggregator = new SummaryAggregator(); + const texts: string[] = []; + aggregator.setSession("session-1"); + aggregator.setOnPartial((_sessionId, _messageId, text) => { + texts.push(text); + }); + const generation = aggregator.holdOutbound(); + + emitPartial(aggregator, "message-1", "first"); + emitPartial(aggregator, "message-1", "second"); + await aggregator.runOutbound(() => {}); + + const draining = aggregator.drainOutbound(1000, generation); + await vi.advanceTimersByTimeAsync(0); + expect(texts).toEqual(["second"]); + + emitDelta(aggregator, "message-1", "message-1-part", " third"); + await vi.advanceTimersByTimeAsync(1000); + await vi.advanceTimersByTimeAsync(1000); + await draining; + + expect(texts).toEqual(["second", "second third"]); + }); + + it("coalesces deferred thinking updates while keeping the first update", async () => { + const aggregator = new SummaryAggregator(); + const onThinking = vi.fn(); + aggregator.setSession("session-1"); + aggregator.setOnThinking(onThinking); + const generation = aggregator.holdOutbound(); + + aggregator.processEvent({ + type: "message.updated", + properties: { + info: { + id: "message-1", + sessionID: "session-1", + role: "assistant", + time: { created: 1 }, + }, + }, + } as unknown as Event); + aggregator.processEvent({ + type: "message.part.updated", + properties: { + part: { + id: "reasoning-1", + sessionID: "session-1", + messageID: "message-1", + type: "reasoning", + title: "Analysis", + text: "first ", + }, + }, + } as unknown as Event); + emitDelta(aggregator, "message-1", "reasoning-1", "second", "reasoning"); + emitDelta(aggregator, "message-1", "reasoning-1", " third", "reasoning"); + + await aggregator.drainOutbound(0, generation); + + expect(onThinking).toHaveBeenCalledTimes(2); + expect(onThinking.mock.calls[0]?.[0]).toMatchObject({ + isFirstUpdate: true, + sections: [{ id: "reasoning-1", title: "Analysis", text: "first " }], + }); + expect(onThinking.mock.calls[1]?.[0]).toMatchObject({ + isFirstUpdate: false, + sections: [{ id: "reasoning-1", title: "Analysis", text: "first second third" }], + }); + }); + + it("runs the latest payload when a delta replaces the head during the gap", async () => { + vi.useFakeTimers(); + const aggregator = new SummaryAggregator(); + const texts: string[] = []; + aggregator.setSession("session-1"); + aggregator.setOnPartial((_sessionId, _messageId, text) => { + texts.push(text); + }); + const generation = aggregator.holdOutbound(); + + await aggregator.runOutbound(() => {}); + emitPartial(aggregator, "message-1", "first"); + + const draining = aggregator.drainOutbound(1000, generation); + await vi.advanceTimersByTimeAsync(0); + expect(texts).toEqual([]); + + emitDelta(aggregator, "message-1", "message-1-part", " second"); + await vi.advanceTimersByTimeAsync(1000); + await draining; + + expect(texts).toEqual(["first second"]); + expect(aggregator.hasDeferredOutbound()).toBe(false); + }); + + it("leaves the queue to a newer hold that starts during the gap", async () => { + vi.useFakeTimers(); + const aggregator = new SummaryAggregator(); + const calls: string[] = []; + const generationA = aggregator.holdOutbound(); + aggregator.runOutbound(() => { + calls.push("first"); + }); + aggregator.runOutbound(() => { + calls.push("second"); + }); + + const drainA = aggregator.drainOutbound(1000, generationA); + await vi.advanceTimersByTimeAsync(0); + expect(calls).toEqual(["first"]); + + const generationB = aggregator.holdOutbound(); + await vi.advanceTimersByTimeAsync(1000); + await drainA; + expect(calls).toEqual(["first"]); + + await aggregator.drainOutbound(0, generationB); + expect(calls).toEqual(["first", "second"]); + }); + + it("returns a caller whose generation went stale while it awaited the drain", async () => { + const aggregator = new SummaryAggregator(); + const calls: string[] = []; + let releaseFirst!: () => void; + const gate = new Promise((resolve) => { + releaseFirst = resolve; + }); + + const generationA = aggregator.holdOutbound(); + aggregator.runOutbound(() => gate); + const drainA = aggregator.drainOutbound(0, generationA); + await Promise.resolve(); + + // Still the current generation: this caller enters the await-the-drain branch. + const secondCallA = aggregator.drainOutbound(0, generationA); + let secondSettled = false; + void secondCallA.then(() => { + secondSettled = true; + }); + await Promise.resolve(); + expect(secondSettled).toBe(false); + + const generationB = aggregator.holdOutbound(); + aggregator.runOutbound(() => { + calls.push("b"); + }); + + releaseFirst(); + await drainA; + await secondCallA; + expect(calls).toEqual([]); + + await aggregator.drainOutbound(0, generationB); + expect(calls).toEqual(["b"]); + }); + + it("returns a stale caller without draining a newer hold", async () => { + const aggregator = new SummaryAggregator(); + const calls: string[] = []; + let releaseFirst!: () => void; + const gate = new Promise((resolve) => { + releaseFirst = resolve; + }); + + const generationA = aggregator.holdOutbound(); + aggregator.runOutbound(() => gate); + const drainA = aggregator.drainOutbound(0, generationA); + await Promise.resolve(); + + const generationB = aggregator.holdOutbound(); + aggregator.runOutbound(() => { + calls.push("b"); + }); + const staleCall = aggregator.drainOutbound(0, generationA); + const drainB = aggregator.drainOutbound(0, generationB); + + releaseFirst(); + await drainA; + await staleCall; + await drainB; + + expect(calls).toEqual(["b"]); + }); }); diff --git a/tests/bot/commands/sessions.test.ts b/tests/bot/commands/sessions.test.ts index 8531fa353..6af941130 100644 --- a/tests/bot/commands/sessions.test.ts +++ b/tests/bot/commands/sessions.test.ts @@ -510,7 +510,9 @@ describe("bot/commands/sessions", () => { reply_markup: { inline_keyboard: [] }, }), ]); - expect(safeBackgroundTaskMock).not.toHaveBeenCalled(); + expect(safeBackgroundTaskMock).toHaveBeenCalledWith( + expect.objectContaining({ taskName: "sessions.releasePickOutbound" }), + ); expect((ctx.api.sendMessage as ReturnType).mock.calls[2]?.[1]).toBe( t("sessions.preview.empty"), ); @@ -585,7 +587,9 @@ describe("bot/commands/sessions", () => { const replyIndex = sent.findIndex((text) => text.includes("finished reply")); const waitingIndex = sent.findIndex((text) => text.includes("newer prompt")); expect(waitingIndex).toBeGreaterThan(replyIndex); - expect(safeBackgroundTaskMock).not.toHaveBeenCalled(); + expect(safeBackgroundTaskMock).toHaveBeenCalledWith( + expect.objectContaining({ taskName: "sessions.releasePickOutbound" }), + ); const sendCalls = (ctx.api.sendMessage as ReturnType).mock.calls; const replyCall = sendCalls.find((call) => String(call[1]).includes("finished reply")); @@ -594,6 +598,184 @@ describe("bot/commands/sessions", () => { expect(quoteCall?.[2]).not.toHaveProperty("link_preview_options"); }); + it("does not wait for the deferred outbound drain when picking a session", async () => { + mocked.sessionGetMock.mockResolvedValueOnce({ + data: createSession(0), + error: null, + }); + mocked.sessionMessagesMock.mockReset(); + mocked.sessionMessagesMock.mockResolvedValueOnce({ + data: [ + { + info: { + id: "user-1", + role: "user", + time: { created: 1 }, + }, + parts: [{ type: "text", text: "the prompt" }], + }, + { + info: { + id: "assistant-1", + role: "assistant", + parentID: "user-1", + time: { created: 2, completed: 3 }, + }, + parts: [{ type: "text", text: "the reply" }], + }, + ], + error: null, + }); + + startInteractionForTest(container.interactionManager, { + kind: "inline", + expectedInput: "callback", + metadata: { + menuKind: "session", + messageId: 456, + }, + }); + + const actualSafeBackgroundTask = await vi.importActual< + typeof import("../../../src/utils/safe-background-task.js") + >("../../../src/utils/safe-background-task.js"); + safeBackgroundTaskMock.mockImplementation(actualSafeBackgroundTask.safeBackgroundTask); + + const deps = createDeps(); + const holdSpy = vi.spyOn(deps.summaryAggregator, "holdOutbound").mockReturnValue(42); + const drainSpy = vi + .spyOn(deps.summaryAggregator, "drainOutbound") + .mockImplementation(() => new Promise(() => {})); + + const ctx = createCallbackContext("session:session-1", 456); + await handleSessionSelect(ctx, deps); + + expect(safeBackgroundTaskMock).toHaveBeenCalledWith( + expect.objectContaining({ taskName: "sessions.releasePickOutbound" }), + ); + expect(holdSpy).toHaveBeenCalledTimes(1); + await Promise.resolve(); + expect(drainSpy).toHaveBeenCalledTimes(1); + expect(drainSpy.mock.calls[0]?.[1]).toBe(42); + }); + + it("drains the pick hold even when the pre-drain check throws", async () => { + mocked.sessionGetMock.mockResolvedValueOnce({ + data: createSession(0), + error: null, + }); + mocked.sessionMessagesMock.mockReset(); + mocked.sessionMessagesMock.mockResolvedValueOnce({ + data: [ + { + info: { + id: "user-1", + role: "user", + time: { created: 1 }, + }, + parts: [{ type: "text", text: "the prompt" }], + }, + { + info: { + id: "assistant-1", + role: "assistant", + parentID: "user-1", + time: { created: 2, completed: 3 }, + }, + parts: [{ type: "text", text: "the reply" }], + }, + ], + error: null, + }); + + startInteractionForTest(container.interactionManager, { + kind: "inline", + expectedInput: "callback", + metadata: { + menuKind: "session", + messageId: 456, + }, + }); + + const actualSafeBackgroundTask = await vi.importActual< + typeof import("../../../src/utils/safe-background-task.js") + >("../../../src/utils/safe-background-task.js"); + safeBackgroundTaskMock.mockImplementation(actualSafeBackgroundTask.safeBackgroundTask); + + const deps = createDeps(); + vi.spyOn(deps.summaryAggregator, "holdOutbound").mockReturnValue(7); + vi.spyOn(deps.summaryAggregator, "hasDeferredOutbound").mockImplementation(() => { + throw new Error("check failed"); + }); + const drainSpy = vi + .spyOn(deps.summaryAggregator, "drainOutbound") + .mockImplementation(() => new Promise(() => {})); + + const ctx = createCallbackContext("session:session-1", 456); + await handleSessionSelect(ctx, deps); + await Promise.resolve(); + + expect(drainSpy).toHaveBeenCalledWith(expect.any(Number), 7); + }); + + it("drains a deferred task in the background after a pick", async () => { + mocked.sessionGetMock.mockResolvedValueOnce({ + data: createSession(0), + error: null, + }); + mocked.sessionMessagesMock.mockReset(); + mocked.sessionMessagesMock.mockResolvedValueOnce({ + data: [ + { + info: { + id: "user-1", + role: "user", + time: { created: 1 }, + }, + parts: [{ type: "text", text: "the prompt" }], + }, + { + info: { + id: "assistant-1", + role: "assistant", + parentID: "user-1", + time: { created: 2, completed: 3 }, + }, + parts: [{ type: "text", text: "the reply" }], + }, + ], + error: null, + }); + + startInteractionForTest(container.interactionManager, { + kind: "inline", + expectedInput: "callback", + metadata: { + menuKind: "session", + messageId: 456, + }, + }); + + const actualSafeBackgroundTask = await vi.importActual< + typeof import("../../../src/utils/safe-background-task.js") + >("../../../src/utils/safe-background-task.js"); + safeBackgroundTaskMock.mockImplementation(actualSafeBackgroundTask.safeBackgroundTask); + + const deps = createDeps(); + const task = vi.fn(); + deps.summaryAggregator.holdOutbound(); + await deps.summaryAggregator.runOutbound(task); + expect(task).not.toHaveBeenCalled(); + + const ctx = createCallbackContext("session:session-1", 456); + await handleSessionSelect(ctx, deps); + + await vi.waitFor(() => { + expect(task).toHaveBeenCalledTimes(1); + }); + expect(deps.summaryAggregator.hasDeferredOutbound()).toBe(false); + }); + it("still sends the empty notice when the first history page fails", async () => { mocked.sessionGetMock.mockResolvedValueOnce({ data: createSession(0), diff --git a/tests/bot/events/handlers/dashboard-handler.test.ts b/tests/bot/events/handlers/dashboard-handler.test.ts index 890eb039c..c736e3e0c 100644 --- a/tests/bot/events/handlers/dashboard-handler.test.ts +++ b/tests/bot/events/handlers/dashboard-handler.test.ts @@ -64,7 +64,7 @@ describe("bot/events/dashboard-handler", () => { }); it("applies values at once during a hold and queues the redraw and the card's files in order", async () => { - aggregator.holdOutbound(); + const generation = aggregator.holdOutbound(); emitSessionDiff(aggregator, "src/a.ts"); emitSessionDiff(aggregator, "src/b.ts"); @@ -72,7 +72,7 @@ describe("bot/events/dashboard-handler", () => { expect(calls).toEqual(["value:src/a.ts", "value:src/b.ts"]); - await aggregator.drainOutbound(0); + await aggregator.drainOutbound(0, generation); expect(calls).toEqual([ "value:src/a.ts", diff --git a/tests/bot/services/event-subscription-service.lost-run.test.ts b/tests/bot/services/event-subscription-service.lost-run.test.ts index 88dd792e7..f3b5507ed 100644 --- a/tests/bot/services/event-subscription-service.lost-run.test.ts +++ b/tests/bot/services/event-subscription-service.lost-run.test.ts @@ -272,6 +272,47 @@ describe("bot/services/event-subscription-service lost run", () => { expect(last).not.toContain("⏳"); }); + it("releases the ending hold when the ending throws", async () => { + const aggregator = container.summaryAggregator; + const interruptSpy = vi.spyOn(aggregator, "interruptSubagents").mockImplementation(() => { + throw new Error("interrupt failed"); + }); + + try { + const ending = service!.endRunLostWithServer("opencode_stop"); + + await expect(ending).rejects.toThrow("interrupt failed"); + await expect(aggregator.whenOutboundIdle()).resolves.toBeUndefined(); + } finally { + interruptSpy.mockRestore(); + } + }); + + it("stays pending when a newer hold takes over the ending queue", async () => { + emitTool(container, "bash", "running", "call-bash", { command: "npm test" }); + await settle(); + + const aggregator = container.summaryAggregator; + const ending = service!.endRunLostWithServer("opencode_stop"); + const newerGeneration = aggregator.holdOutbound(); + + let settled = false; + void ending.then(() => { + settled = true; + }); + await settle(); + expect(settled).toBe(false); + + const draining = aggregator.drainOutbound(0, newerGeneration); + await settle(); + await draining; + await ending; + expect(settled).toBe(true); + + const last = collectSentTexts(api).filter((text) => text.includes("npm test")).at(-1) ?? ""; + expect(last).not.toContain("⏳"); + }); + it("closes the poll on screen as not answered instead of deleting it", async () => { emitTool(container, "question", "running", "call-question", {}); await settle();