diff --git a/products/desktop/docs/TROUBLESHOOTING.md b/products/desktop/docs/TROUBLESHOOTING.md index 352f3bf3b47a..d577a2bd4b1d 100644 --- a/products/desktop/docs/TROUBLESHOOTING.md +++ b/products/desktop/docs/TROUBLESHOOTING.md @@ -9,6 +9,16 @@ Restarting the session does not reduce the request size. Start a new task with a short text summary of the work. Do not copy the old images or full tool output into the new task. +## Completion sounds when resuming a task + +Resuming an idle sandbox does not trigger a completion sound or notification. +For Pi sessions, completion notifications require an active turn and stay silent during restart setup. +After a cloud restart, the server must confirm the resent message before a completion can notify. +Returning to the task while the message is pending does not trigger a completion notification. +Delayed completion events from restart setup stay silent while the message is pending. +If you stop a pending message, a successful cancellation clears the running state even if the message never appears in the conversation. +A completed reply or a request for input still follows your notification settings. + ## Codex asks for the same permissions again Codex Auto keeps approvals for actions outside its allowed scope. diff --git a/products/desktop/packages/agent/e2e/README.md b/products/desktop/packages/agent/e2e/README.md index 69c9d05aa7ef..48ea2b1edd97 100644 --- a/products/desktop/packages/agent/e2e/README.md +++ b/products/desktop/packages/agent/e2e/README.md @@ -33,7 +33,7 @@ codex arm only, and `guard.e2e.test.ts` always runs: Codex-only (advertised codex capabilities; registered as skipped on the claude arm so the gap is visible): - **mode switch** → `current_mode_update`. -- **steering** — a mid-turn prompt folds into the running turn via `turn/steer`. +- **steering** — the host holds a question open while it submits a second prompt, which folds into the running turn via `turn/steer`. - **list + fork** — `listSessions` finds the session; `forkSession` branches it. The command/file approval `{decision}` round-trip is **not** covered here: codex diff --git a/products/desktop/packages/agent/e2e/driver.e2e.test.ts b/products/desktop/packages/agent/e2e/driver.e2e.test.ts index d4d58560fb82..f0bd33505ae9 100644 --- a/products/desktop/packages/agent/e2e/driver.e2e.test.ts +++ b/products/desktop/packages/agent/e2e/driver.e2e.test.ts @@ -21,6 +21,37 @@ vi.mock("../src/adapters/acp-connection", () => ({ })); describe("live e2e host permissions", () => { + it("holds a question open until the host releases its answer", async () => { + let releaseAnswer: () => void = () => {}; + const answer = new Promise((resolve) => { + releaseAnswer = resolve; + }); + const onQuestion = vi.fn(() => answer); + openConnection({ adapter: "claude", cwd: "/tmp", onQuestion }); + const request = { + sessionId: "session", + toolCall: { + toolCallId: "question", + title: "Choose an approach", + kind: "other", + _meta: { codeToolKind: "question" }, + }, + options: [{ kind: "allow_once", optionId: "option_0", name: "A" }], + } as unknown as RequestPermissionRequest; + const answered = vi.fn(); + const response = Promise.resolve( + host.client?.requestPermission(request), + ).then(answered); + + await vi.waitFor(() => expect(onQuestion).toHaveBeenCalledOnce()); + expect(answered).not.toHaveBeenCalled(); + releaseAnswer(); + await response; + expect(answered).toHaveBeenCalledWith({ + outcome: { outcome: "selected", optionId: "option_0" }, + }); + }); + it.each([ { kind: "switch_mode", diff --git a/products/desktop/packages/agent/e2e/driver.ts b/products/desktop/packages/agent/e2e/driver.ts index 637bf1b1ee02..02d2fd28809e 100644 --- a/products/desktop/packages/agent/e2e/driver.ts +++ b/products/desktop/packages/agent/e2e/driver.ts @@ -92,6 +92,7 @@ export function openConnection(opts: { cwd: string; codexOptions?: Record; onStructuredOutput?: (output: Record) => Promise; + onQuestion?: () => Promise; }): E2EConnection { const { adapter, cwd } = opts; // Sweep before every codex spawn so one leaked process (holding the @@ -119,6 +120,9 @@ export function openConnection(opts: { codeToolKind: p?.toolCall?._meta?.codeToolKind, }, }); + if (p?.toolCall?._meta?.codeToolKind === "question") { + await opts.onQuestion?.(); + } // Accepting a plan starts extra model turns outside the test's scenario. if (p?.toolCall?.kind === "switch_mode") { return { outcome: { outcome: "cancelled" } }; @@ -206,6 +210,7 @@ export async function openSession(opts: { cwd: string; codexOptions?: Record; onStructuredOutput?: (output: Record) => Promise; + onQuestion?: () => Promise; meta: Record; }): Promise { const c = openConnection(opts); diff --git a/products/desktop/packages/agent/e2e/session-lifecycle.e2e.test.ts b/products/desktop/packages/agent/e2e/session-lifecycle.e2e.test.ts index 2d18fe22d842..0515efde3a3f 100644 --- a/products/desktop/packages/agent/e2e/session-lifecycle.e2e.test.ts +++ b/products/desktop/packages/agent/e2e/session-lifecycle.e2e.test.ts @@ -382,40 +382,59 @@ for (const adapter of ADAPTERS) { itCodex( "folds a mid-turn prompt into the running turn via steering", async () => { + let questionReady: () => void = () => {}; + const question = new Promise((resolve) => { + questionReady = resolve; + }); + let releaseAnswer: () => void = () => {}; + const answer = new Promise((resolve) => { + releaseAnswer = resolve; + }); const s = await openSession({ adapter, cwd: repo, codexOptions: codexOptions(), meta: meta(), + onQuestion: async () => { + questionReady(); + await answer; + }, }); try { + await s.conn.setSessionConfigOption({ + sessionId: s.sessionId, + configId: "mode", + value: "plan", + }); const p1 = s.conn.prompt({ sessionId: s.sessionId, prompt: [ { type: "text", - text: "Count up from 1, one number per line, and keep going.", + text: + "Call request_user_input once to ask whether to use approach A or B. " + + "Wait for the answer, then acknowledge it in one sentence and stop. " + + "Do not call other tools or propose an implementation plan.", }, ], }); - await waitFor( - () => - s.capture.updates("agent_message_chunk").length > 0 - ? true - : undefined, - 20_000, - ); - const p2 = s.conn.prompt({ + await Promise.race([ + question, + p1.then(() => { + throw new Error("The first turn ended before asking a question"); + }), + ]); + const r2 = await s.conn.prompt({ sessionId: s.sessionId, prompt: [{ type: "text", text: "Now stop and say DONE." }], }); - const [r1] = await Promise.all([p1, p2]); + expect(r2).toMatchObject({ _meta: { steer: true } }); + releaseAnswer(); + const r1 = await p1; expect(r1.stopReason).toBe("end_turn"); expect( s.capture.updates("user_message_chunk").length, ).toBeGreaterThanOrEqual(2); - // The steer proof: folded into a SINGLE turn (one turn_complete). Two would - // mean the steer didn't take and p2 ran as its own turn. const turnCompletes = s.capture.events.filter( (e) => e.kind === "extNotification" && @@ -427,6 +446,7 @@ for (const adapter of ADAPTERS) { "turn_complete); 2 means the steer didn't take", ).toBe(1); } finally { + releaseAnswer(); await s.cleanup(); } }, diff --git a/products/desktop/packages/core/src/pi-runtime/piSessionController.test.ts b/products/desktop/packages/core/src/pi-runtime/piSessionController.test.ts index aef3c8e39b59..09f75ddb93f3 100644 --- a/products/desktop/packages/core/src/pi-runtime/piSessionController.test.ts +++ b/products/desktop/packages/core/src/pi-runtime/piSessionController.test.ts @@ -143,7 +143,24 @@ describe("PiSessionController", () => { ); }); - it("keeps a cancelled turn cancelled when a buffered chunk is redelivered", async () => { + it.each<[string, (chunk: AgentConversationEvent) => AgentConversationEvent]>([ + // The transport redelivers the chunk while it still sits in the batch. + ["a buffered chunk is redelivered", (chunk) => chunk], + // A polled batch can carry activity produced before the abort landed. + [ + "activity produced before the stop arrives late", + (chunk) => ({ ...chunk, sourceId: "chunk-1", timestamp: 2 }), + ], + [ + "the sent message echoes back late", + () => ({ + type: "user_message", + id: "message-1", + timestamp: 2, + content: [{ type: "text", text: "continue" }], + }), + ], + ])("keeps a cancelled turn cancelled when %s", async (_case, lateEvent) => { vi.useFakeTimers(); const session = createSession(); let receive: ( @@ -175,10 +192,10 @@ describe("PiSessionController", () => { receive(chunk, { isLive: true }); // The user cancels while the chunk is still buffered. await controller.abort("task-1"); - // A redelivery of the same still-buffered chunk must not reopen the turn: - // if it reaches applyTurnEvent it discards the "cancelled" stop reason. - receive(chunk, { isLive: true }); - receive({ type: "turn_completed", timestamp: 2 }, { isLive: true }); + // Whatever lands after the stop must not clear the "cancelled" reason, or + // the completion below notifies the user who just pressed stop. + receive(lateEvent(chunk), { isLive: true }); + receive({ type: "turn_completed", timestamp: 3 }, { isLive: true }); vi.advanceTimersByTime(16); expect(notifier.notify).toHaveBeenCalledWith( @@ -713,55 +730,70 @@ describe("PiSessionController", () => { ); }); - it("notifies when a live completion arrives before history hydration", async () => { - let onEvent: ( - event: AgentConversationEvent, - context?: PiConversationEventContext, - ) => void = () => {}; - let resolveConversation: (events: AgentConversationEvent[]) => void = - () => {}; - const conversation = new Promise((resolve) => { - resolveConversation = resolve; - }); - const session = createSession(); - vi.mocked(session.getConversation).mockReturnValue(conversation); - vi.mocked(session.onConversationEvent).mockImplementation((handler) => { - onEvent = handler; - return () => {}; - }); - const notifier = { notify: vi.fn() }; - const controller = createController( - session, - undefined, - undefined, - notifier, - ); - controller.setNotificationContext("task-1", { - taskTitle: "Fix notifications", - }); - - const connection = controller.connect("task-1"); - await vi.waitFor(() => { - expect(session.onConversationEvent).toHaveBeenCalledOnce(); - }); - onEvent( - { type: "turn_completed", timestamp: 52, stopReason: "stop" }, - { isLive: true }, - ); + it.each([false, true])( + "handles a live completion before history loads with turn activity: %s", + async (hasTurnActivity) => { + let onEvent: ( + event: AgentConversationEvent, + context?: PiConversationEventContext, + ) => void = () => {}; + let resolveConversation: (events: AgentConversationEvent[]) => void = + () => {}; + const conversation = new Promise((resolve) => { + resolveConversation = resolve; + }); + const session = createSession(); + vi.mocked(session.getConversation).mockReturnValue(conversation); + vi.mocked(session.onConversationEvent).mockImplementation((handler) => { + onEvent = handler; + return () => {}; + }); + const notifier = { notify: vi.fn() }; + const controller = createController( + session, + undefined, + undefined, + notifier, + ); + controller.setNotificationContext("task-1", { + taskTitle: "Fix notifications", + }); - expect(notifier.notify).toHaveBeenCalledWith({ - kind: "turn_completed", - trigger: "pi_turn_completed", - taskId: "task-1", - taskTitle: "Fix notifications", - stopReason: "end_turn", - durationMs: undefined, - isTaskAuthor: undefined, - }); + const connection = controller.connect("task-1"); + await vi.waitFor(() => { + expect(session.onConversationEvent).toHaveBeenCalledOnce(); + }); + if (hasTurnActivity) { + onEvent( + { + type: "assistant_message_chunk", + timestamp: 51, + content: { type: "text", text: "Done" }, + }, + { isLive: true }, + ); + } + onEvent( + { type: "turn_completed", timestamp: 52, stopReason: "stop" }, + { isLive: true }, + ); - resolveConversation([]); - await connection; - }); + expect(notifier.notify).toHaveBeenCalledTimes(hasTurnActivity ? 1 : 0); + if (hasTurnActivity) + expect(notifier.notify).toHaveBeenCalledWith({ + kind: "turn_completed", + trigger: "pi_turn_completed", + taskId: "task-1", + taskTitle: "Fix notifications", + stopReason: "end_turn", + durationMs: undefined, + isTaskAuthor: undefined, + }); + + resolveConversation([]); + await connection; + }, + ); it("notifies once for a live completed turn without replaying historical completions", async () => { let onEvent: ( @@ -885,6 +917,77 @@ describe("PiSessionController", () => { expect(unsubscribe).toHaveBeenCalledOnce(); }); + it.each([false, true])( + "releases an aborted pending turn with completion event: %s", + async (receivesCompletion) => { + const staleSession = { + ...createSession(), + resumeRequired: true, + taskRunId: "run-1", + sendUserMessage: vi.fn(async () => {}), + }; + const session = { + ...createSession(), + sendUserMessage: vi.fn(async () => {}), + }; + let onEvent: (event: AgentConversationEvent) => void = () => {}; + const unsubscribe = vi.fn(); + vi.mocked(session.onConversationEvent).mockImplementation((handler) => { + onEvent = handler; + return unsubscribe; + }); + const provider = { + get: vi + .fn() + .mockResolvedValueOnce(staleSession) + .mockResolvedValue(session), + }; + const taskService = { + resumeCloudPiRun: vi.fn(async () => ({ id: "run-2" })), + prepareCloudPiMessage: vi.fn(async () => ({ + content: "continue", + artifactIds: [], + })), + } as unknown as TaskService; + const notifier = { notify: vi.fn() }; + const controller = new PiSessionController( + provider, + taskService, + undefined, + notifier, + ); + controller.setNotificationContext("task-1", { taskTitle: "Continue" }); + + await controller.connect("task-1"); + await controller.submit("task-1", "continue", false, "steer"); + vi.mocked(session.client.abort).mockRejectedValueOnce( + new Error("Unable to stop"), + ); + await expect(controller.abort("task-1")).rejects.toThrow( + PiOperationError, + ); + expect( + controller.store.getState().sessions["task-1"].status?.isStreaming, + ).toBe(true); + + await controller.abort("task-1"); + expect( + controller.store.getState().sessions["task-1"].status?.isStreaming, + ).toBe(false); + if (receivesCompletion) { + onEvent({ type: "turn_completed", timestamp: Date.now() }); + expect(notifier.notify).toHaveBeenCalledWith( + expect.objectContaining({ stopReason: "cancelled" }), + ); + } else { + expect(notifier.notify).not.toHaveBeenCalled(); + } + controller.release("task-1"); + expect(unsubscribe).toHaveBeenCalledOnce(); + expect(controller.store.getState().sessions["task-1"].events).toEqual([]); + }, + ); + it("keeps a background cloud session until the run ends", async () => { const session: PiSession = { ...createSession(), @@ -910,50 +1013,56 @@ describe("PiSessionController", () => { expect(controller.store.getState().sessions["task-1"].events).toEqual([]); }); - it("notifies a background cloud completion before releasing the session", async () => { - const session: PiSession = { - ...createSession(), - cloudStatus: "in_progress", - }; - const unsubscribe = vi.fn(); - let onEvent: ( - event: AgentConversationEvent, - context?: PiConversationEventContext, - ) => void = () => {}; - let onCloudStatus: Parameters[2]; - vi.mocked(session.onConversationEvent).mockImplementation( - (handler, _error, onStatus) => { - onEvent = handler; - onCloudStatus = onStatus; - return unsubscribe; - }, - ); - const notifier = { notify: vi.fn() } as unknown as AgentSessionNotifier; - const controller = createController( - session, - undefined, - undefined, - notifier, - ); - controller.setNotificationContext("task-1", { - taskTitle: "Fix notifications", - isTaskAuthor: true, - }); - - await controller.connect("task-1"); - controller.release("task-1"); + it.each([false, true])( + "handles a background cloud completion with turn activity: %s", + async (hasTurnActivity) => { + const session: PiSession = { + ...createSession(), + cloudStatus: "in_progress", + }; + const unsubscribe = vi.fn(); + let onEvent: ( + event: AgentConversationEvent, + context?: PiConversationEventContext, + ) => void = () => {}; + let onCloudStatus: Parameters[2]; + vi.mocked(session.onConversationEvent).mockImplementation( + (handler, _error, onStatus) => { + onEvent = handler; + onCloudStatus = onStatus; + return unsubscribe; + }, + ); + const notifier = { notify: vi.fn() } as unknown as AgentSessionNotifier; + const controller = createController( + session, + undefined, + undefined, + notifier, + ); + controller.setNotificationContext("task-1", { + taskTitle: "Fix notifications", + isTaskAuthor: true, + }); - // The cloud client reports the terminal status first, then emits the turn event. - onCloudStatus?.("completed"); - onEvent( - { type: "turn_completed", timestamp: 5, stopReason: "stop" }, - { isLive: true }, - ); - await Promise.resolve(); + await controller.connect("task-1"); + if (hasTurnActivity) { + await controller.submit("task-1", "continue", false, "steer"); + } + controller.release("task-1"); + + // The cloud client reports the terminal status first, then emits the turn event. + onCloudStatus?.("completed"); + onEvent( + { type: "turn_completed", timestamp: 5, stopReason: "stop" }, + { isLive: true }, + ); + await Promise.resolve(); - expect(notifier.notify).toHaveBeenCalledOnce(); - expect(unsubscribe).toHaveBeenCalledOnce(); - }); + expect(notifier.notify).toHaveBeenCalledTimes(hasTurnActivity ? 1 : 0); + expect(unsubscribe).toHaveBeenCalledOnce(); + }, + ); it("releases a finished cloud run holding an unanswered permission request", async () => { const session: PiSession = { @@ -1445,6 +1554,14 @@ describe("PiSessionController", () => { taskRunId: "run-1", }; const resumedSession = createSession(); + let onEvent: (event: AgentConversationEvent) => void = () => {}; + vi.mocked(resumedSession.onConversationEvent).mockImplementation( + (handler) => { + onEvent = handler; + handler({ type: "turn_completed", timestamp: 1, stopReason: "stop" }); + return () => {}; + }, + ); const provider = { get: vi .fn() @@ -1453,7 +1570,17 @@ describe("PiSessionController", () => { } as PiSessionProvider; const resumeCloudPiRun = vi.fn(async () => ({ id: "run-1" })); const taskService = { resumeCloudPiRun } as unknown as TaskService; - const controller = new PiSessionController(provider, taskService); + const notifier = { notify: vi.fn() }; + const controller = new PiSessionController( + provider, + taskService, + undefined, + notifier, + ); + controller.setNotificationContext("task-1", { + taskTitle: "Continue the task", + isTaskAuthor: true, + }); await controller.connect("task-1"); @@ -1463,6 +1590,13 @@ describe("PiSessionController", () => { expect(resumeCloudPiRun).toHaveBeenCalledWith("task-1", "run-1"); expect(resumedSession.client.prompt).toHaveBeenCalledWith("continue"); + expect(notifier.notify).not.toHaveBeenCalled(); + onEvent({ + type: "turn_completed", + timestamp: Date.now(), + stopReason: "stop", + }); + expect(notifier.notify).toHaveBeenCalledOnce(); }); it("applies deferred Pi config before the first resumed prompt", async () => { @@ -1503,22 +1637,54 @@ describe("PiSessionController", () => { }); it.each([ - ["the prior sandbox is gone", "No active sandbox for this task run"], - ["the prior workflow has ended", "Task run workflow has ended"], + ["the prior run is terminal", "", true], + ["the prior sandbox is gone", "No active sandbox for this task run", false], + ["the prior workflow has ended", "Task run workflow has ended", false], + [ + "a pending prompt reconnects", + "", + true, + { isStreaming: false, confirmFromHistory: false }, + ], + [ + "a pending steer reconnects", + "", + true, + { isStreaming: true, confirmFromHistory: true }, + ], ])( - "resumes and retries a message when %s", - async (_condition, errorMessage) => { + "resumes and sends a message when %s", + async (_condition, errorMessage, resumeRequired, reconnect = undefined) => { const staleSession = { ...createSession(), taskRunId: "run-1", + resumeRequired, sendUserMessage: vi.fn(async () => { throw new Error(errorMessage); }), }; - const resumedSession = { + let resentMessageId = ""; + const resumedSession: PiSession = { ...createSession(), - sendUserMessage: vi.fn(async () => {}), + sendUserMessage: vi.fn( + async (_type, _content, _artifacts, messageId) => { + resentMessageId = messageId; + onEvent({ + type: "turn_completed", + timestamp: 2, + stopReason: "stop", + }); + }, + ), }; + let onEvent: (event: AgentConversationEvent) => void = () => {}; + vi.mocked(resumedSession.onConversationEvent).mockImplementation( + (handler) => { + onEvent = handler; + handler({ type: "turn_completed", timestamp: 1, stopReason: "stop" }); + return () => {}; + }, + ); const provider = { get: vi .fn() @@ -1533,18 +1699,77 @@ describe("PiSessionController", () => { })), resumeCloudPiRun, } as unknown as TaskService; - const controller = new PiSessionController(provider, taskService); + const notifier = { notify: vi.fn() }; + const controller = new PiSessionController( + provider, + taskService, + undefined, + notifier, + ); + controller.setNotificationContext("task-1", { + taskTitle: "Continue the task", + isTaskAuthor: true, + }); await controller.connect("task-1"); - await controller.submit("task-1", "continue", false, "steer"); + await controller.submit( + "task-1", + "continue", + reconnect?.isStreaming ?? false, + "steer", + ); expect(resumeCloudPiRun).toHaveBeenCalledWith("task-1", "run-1"); expect(resumedSession.sendUserMessage).toHaveBeenCalledWith( - "prompt", + reconnect?.isStreaming ? "steer" : "prompt", "continue", [], expect.any(String), ); + expect(notifier.notify).not.toHaveBeenCalled(); + onEvent({ + type: "user_message", + id: "previous-message", + timestamp: 3, + content: [{ type: "text", text: "old prompt" }], + }); + onEvent({ type: "turn_completed", timestamp: 4, stopReason: "stop" }); + expect(notifier.notify).not.toHaveBeenCalled(); + expect( + controller.store.getState().sessions["task-1"].status?.isStreaming, + ).toBe(true); + if (reconnect) { + const status = await resumedSession.client.getState(); + vi.mocked(resumedSession.client.getState).mockResolvedValue({ + ...status, + isStreaming: reconnect.isStreaming, + }); + controller.release("task-1"); + await controller.ensureConnected("task-1", "run-2"); + onEvent({ type: "turn_completed", timestamp: 5, stopReason: "stop" }); + expect(notifier.notify).not.toHaveBeenCalled(); + } + const confirmedMessage: AgentConversationEvent = { + type: "user_message", + id: resentMessageId, + sourceId: `server:${resentMessageId}`, + timestamp: 6, + content: [{ type: "text", text: "continue" }], + }; + if (reconnect?.confirmFromHistory) { + vi.mocked(resumedSession.getConversation).mockResolvedValue([ + confirmedMessage, + ]); + await controller.ensureConnected("task-1", "run-2"); + } else { + onEvent(confirmedMessage); + } + onEvent({ + type: "turn_completed", + timestamp: Date.now(), + stopReason: "stop", + }); + expect(notifier.notify).toHaveBeenCalledOnce(); }, ); diff --git a/products/desktop/packages/core/src/pi-runtime/piSessionController.ts b/products/desktop/packages/core/src/pi-runtime/piSessionController.ts index 67b445687e9f..91acd4316742 100644 --- a/products/desktop/packages/core/src/pi-runtime/piSessionController.ts +++ b/products/desktop/packages/core/src/pi-runtime/piSessionController.ts @@ -126,6 +126,7 @@ type TextEvent = Extract< const STREAM_BATCH_MS = 16; type PiTurnState = + | { phase: "pending"; messageId: string; startedAt: number } | { phase: "active"; startedAt?: number; stopReason?: string } | { phase: "completed" }; @@ -590,7 +591,7 @@ export class PiSessionController { }); } const previousTurnState = this.turnStates.get(taskId); - this.markTurnPending(taskId); + this.setTurnStreaming(taskId, true); if (currentSession.resumeRequired) { this.updateSession(taskId, { connectionState: "connecting" }); } @@ -598,7 +599,10 @@ export class PiSessionController { try { const session = await this.getWritablePiSession(taskId); await this.applyDeferredConfig(session, deferredConfig); - this.markTurnPending(taskId); + this.markTurnPending( + taskId, + currentSession.resumeRequired ? messageId : undefined, + ); if (session.sendUserMessage && messageId) { const taskRunId = this.taskRunIds.get(taskId); const prepared = taskRunId @@ -702,6 +706,15 @@ export class PiSessionController { try { const session = await this.getPiSession(taskId); await session.client.abort(); + const currentTurn = this.turnStates.get(taskId); + if (currentTurn?.phase === "pending" || currentTurn?.phase === "active") { + this.turnStates.set(taskId, { + phase: "active", + startedAt: currentTurn.startedAt, + stopReason: "cancelled", + }); + } + this.setTurnStreaming(taskId, false); await this.refreshStatus(taskId); } catch (error) { throw this.recordOperationFailure(taskId, "cancel", error); @@ -1102,7 +1115,11 @@ export class PiSessionController { if (status && hasTurnActivity) { status = { ...status, isStreaming: true }; } - if (status && event.type === "turn_completed") { + if ( + status && + event.type === "turn_completed" && + this.turnStates.get(taskId)?.phase !== "pending" + ) { status = { ...status, isStreaming: false }; } @@ -1157,11 +1174,14 @@ export class PiSessionController { events: AgentConversationEvent[], isStreaming: boolean, ): void { - this.turnStates.delete(taskId); + if (this.turnStates.get(taskId)?.phase !== "pending") { + this.turnStates.delete(taskId); + } for (const event of events) { this.applyTurnEvent(taskId, event, false); } - if (isStreaming && this.turnStates.get(taskId)?.phase !== "active") { + const phase = this.turnStates.get(taskId)?.phase; + if (isStreaming && phase !== "active" && phase !== "pending") { this.turnStates.set(taskId, { phase: "active" }); } } @@ -1172,6 +1192,19 @@ export class PiSessionController { isLive: boolean, ): void { const current = this.turnStates.get(taskId); + if (current?.phase === "pending") { + if ( + event.type === "user_message" && + event.id === current.messageId && + !event.sourceId?.startsWith("optimistic:") + ) { + this.turnStates.set(taskId, { + phase: "active", + startedAt: current.startedAt, + }); + } + return; + } const activeTurn = current?.phase === "active" ? current : undefined; const isDirectBash = (event.type === "tool_call_started" || @@ -1191,6 +1224,13 @@ export class PiSessionController { startedAt: activeTurn?.startedAt ?? (event.type === "user_message" ? event.timestamp : undefined), + // A cancel is final for its turn. Activity that was already in flight + // when the abort landed must keep the reason, because a completion that + // carries no reason of its own then reads as "end_turn" and notifies + // the user who just pressed stop. A failure still clears here, because + // a recovering turn continues. + stopReason: + activeTurn?.stopReason === "cancelled" ? "cancelled" : undefined, }); return; } @@ -1209,7 +1249,7 @@ export class PiSessionController { } this.turnStates.set(taskId, { phase: "completed" }); - if (!isLive || current?.phase === "completed") { + if (!isLive || !activeTurn) { return; } @@ -1524,11 +1564,16 @@ export class PiSessionController { ); } - private markTurnPending(taskId: string): void { + private markTurnPending(taskId: string, messageId?: string): void { const current = this.turnStates.get(taskId); const startedAt = current?.phase === "active" ? current.startedAt : Date.now(); - this.turnStates.set(taskId, { phase: "active", startedAt }); + this.turnStates.set( + taskId, + messageId + ? { phase: "pending", messageId, startedAt: startedAt ?? Date.now() } + : { phase: "active", startedAt }, + ); this.setTurnStreaming(taskId, true); } @@ -1622,11 +1667,13 @@ export class PiSessionController { taskRunId, ); this.resetTransport(taskId); + this.turnStates.delete(taskId); await this.ensureConnected(taskId, resumedRun.id); const resumedSession = await this.getPiSession(taskId); if (!resumedSession.sendUserMessage) { throw new Error("Resumed cloud Pi session cannot send messages"); } + this.markTurnPending(taskId, messageId); await resumedSession.sendUserMessage( type, content, @@ -1689,6 +1736,7 @@ export class PiSessionController { this.submissionsInFlight.has(taskId) || session.isBashRunning || session.authRestoring || + turnState?.phase === "pending" || // A turn that already recorded a failure receives no turn_completed, so it // is finished, not in flight. A recovering turn clears the failure on its // next activity event.