From 056f55fc53abcc52b5b97e4ad92b0b1e7c740a25 Mon Sep 17 00:00:00 2001 From: cat0825 <1759138827@qq.com> Date: Mon, 24 Aug 2026 20:42:02 +0800 Subject: [PATCH] fix(cli): dispatch turn cancellation ahead of the queue barrier (3698) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Double-Escape and Ctrl-C recognized the interrupt gesture but did not cancel anything until the client-side queue work had settled. The interrupt path awaited `settlePendingEnqueues()` and `retractQueued()` before it ever reached `driver.stop()`, so a pending `turn.message.submit` round trip that hung on transport, Session admission, storage, or a fallback retry put an unbounded wait in front of the abort. The TUI meanwhile kept rendering `Working…`, leaving the user with no evidence the keypress had registered. Reverse the order and give the runtime the authority. `MakaSessionDriver` gains an optional `interruptTurn()`; the Runtime Host driver implements it with the atomic `turn.interrupt` operation, which commits the queue stop fence, retracts, and aborts the owning turn as one control-mode call. Cancellation now goes out first, and ordering is still exact because the fence — not client-side sequencing — decides each message's fate: an enqueue that committed before the fence returns in `retracted`, and one that lost the race rejects and restores its own text through the existing enqueue catch. Drivers without `interruptTurn()` compose `retractQueued()` then `stop()`, preserving today's semantics. Acceptance is also now visible immediately. `interruptRequestedAt` is stamped in the same tick as gesture recognition and rendered by the activity strip as `Cancelling… `, which outranks both `Working…` and a scheduled provider retry; the counter keeps a slow cleanup, such as a tool held through its process termination grace, legible as progress rather than a hang. Two existing runner fakes modelled cancellation as edge-triggered — a bare `resolve` callback, and a flag reset at async-generator body entry — so an abort landing before the drain pulled its first event was dropped. An async generator does not run its body until the first `next()` call, which the reordering exposed. The real Host channel buffers durable events from `eventsForTurn()` at turn creation, so it is level-triggered; the fakes now arm their abort state in `preparePrompt()` to match. Fixes #3698 --- .../cli/src/__tests__/pi-transcript.test.ts | 57 +++- .../cli/src/__tests__/pi-tui-runner.test.ts | 269 +++++++++++++++++- .../runtime-host-session-driver.test.ts | 97 +++++++ packages/cli/src/pi-transcript.ts | 22 +- packages/cli/src/pi-tui-runner.ts | 41 ++- .../cli/src/runtime-host-session-driver.ts | 15 + packages/cli/src/session-driver.ts | 10 + 7 files changed, 497 insertions(+), 14 deletions(-) diff --git a/packages/cli/src/__tests__/pi-transcript.test.ts b/packages/cli/src/__tests__/pi-transcript.test.ts index b57e43775c..5f4e435d95 100644 --- a/packages/cli/src/__tests__/pi-transcript.test.ts +++ b/packages/cli/src/__tests__/pi-transcript.test.ts @@ -22,7 +22,7 @@ import { describe, test } from 'node:test'; import { visibleWidth } from '@earendil-works/pi-tui'; import type { PipeShellOutput, PtyShellOutput } from '@maka/core/shell-run'; import type { ShellRunToolResult } from '@maka/core/shell-run-result'; -import type { SessionEvent, ToolResultContent } from '@maka/core/events'; +import type { ProviderRetryEvent, SessionEvent, ToolResultContent } from '@maka/core/events'; import type { StoredMessage } from '@maka/core/session'; import { appendUserPrompt, @@ -42,6 +42,7 @@ import { toggleAllThinkingExpansion, toggleAllToolExpansion, type MakaPiToolEntry, + type MakaPiTranscriptMetadata, } from '../pi-transcript.js'; function toolStatus(entry: MakaPiToolEntry | undefined): string | undefined { @@ -3650,6 +3651,46 @@ describe('Maka Pi TUI transcript', () => { assert.doesNotMatch(expanded, /progress-0\b/); assert.match(expanded, /progress-512\b/); }); + + test('activity strip reports cancellation ahead of working and retry states', () => { + const strip = (extra: Partial): string => + stripAnsi(renderMakaPiActivityStrip({ ...meta(), ...extra }, 80)); + + // A running turn with nothing else to say reports elapsed work. + assert.equal(strip({ turnElapsedMs: 3_000 }), 'Working… 3s'); + + // Cancellation is announced from the moment the gesture is accepted, so the + // first render after recognition already reads `Cancelling…` — zero elapsed + // is the common case, not an edge case, and must not fall back to `Working…`. + assert.equal(strip({ turnElapsedMs: 3_000, interruptElapsedMs: 0 }), 'Cancelling… 0s'); + + // A cleanup that outlives the gesture (a tool held through the process + // termination grace) stays legible as progress rather than looking hung. + assert.equal(strip({ turnElapsedMs: 9_000, interruptElapsedMs: 2_000 }), 'Cancelling… 2s'); + + // A retry scheduled before the interrupt is superseded by it: the turn is no + // longer working towards anything the user asked for. + assert.equal( + strip({ + turnElapsedMs: 9_000, + interruptElapsedMs: 1_000, + providerRetry: scheduledRetry(), + }), + 'Cancelling… 1s', + ); + + // Without an interrupt the retry still wins over `Working…`, unchanged. + assert.equal( + strip({ + turnElapsedMs: 9_000, + providerRetry: scheduledRetry(), + }), + 'Retrying in 30s (2/5)', + ); + + // An idle transcript stays silent even though a previous interrupt happened. + assert.equal(strip({}), ''); + }); }); describe('transcript entry render memoization', () => { @@ -4045,6 +4086,20 @@ function subagentResult( }; } +function scheduledRetry(): ProviderRetryEvent { + return { + type: 'provider_retry', + phase: 'scheduled', + id: 'event-retry', + turnId: 'turn-1', + ts: 1, + attempt: 2, + maxAttempts: 5, + delayMs: 30_000, + reason: 'rate_limit', + }; +} + function stripAnsi(text: string): string { return text.replace(/\x1b\[[0-9;]*m/g, ''); } diff --git a/packages/cli/src/__tests__/pi-tui-runner.test.ts b/packages/cli/src/__tests__/pi-tui-runner.test.ts index 84d559537f..9f3061cf58 100644 --- a/packages/cli/src/__tests__/pi-tui-runner.test.ts +++ b/packages/cli/src/__tests__/pi-tui-runner.test.ts @@ -4836,6 +4836,120 @@ describe('Maka Pi TUI runner', () => { await run; }); + test('dispatches the interrupt ahead of a never-settling enqueue', async () => { + const terminal = new FakeTerminal(); + const driver = new StuckEnqueueDriver(); + const run = runMakaPiTui({ + title: 'Maka', + driver, + cwd: '/repo', + model: 'claude-sonnet-4-5', + connectionSlug: 'claude-subscription', + permissionMode: 'ask', + terminal, + }); + + terminal.input('run'); + terminal.input('\r'); + await waitFor(() => terminal.progressStates.at(-1) === true); + + // Mid-turn Enter steers. This driver's steer RPC never settles, standing in + // for a `turn.message.submit` delayed by transport, Session admission, + // storage, or a fallback retry — so the enqueue task stays pending for the + // rest of the turn. + terminal.input('unfinished idea'); + terminal.input('\r'); + await waitFor(() => driver.steerCalls === 1); + + terminal.input('\x1b'); + terminal.input('\x1b'); + // The cancellation authority must be reached without waiting on that RPC. + await waitFor(() => driver.stopCalls === 1, 'the stop authority to be reached'); + // ...and the turn must actually converge, not merely be asked to. + await waitFor(() => terminal.progressStates.at(-1) === false); + + exitMaka(terminal); + await run; + }); + + test('reports Cancelling while the stop authority is still converging', async () => { + const terminal = new FakeTerminal(); + const driver = new SlowStopDriver(); + const run = runMakaPiTui({ + title: 'Maka', + driver, + cwd: '/repo', + model: 'claude-sonnet-4-5', + connectionSlug: 'claude-subscription', + permissionMode: 'ask', + terminal, + }); + + terminal.input('run'); + terminal.input('\r'); + await waitFor(() => plainTerminalOutput(terminal.screenOutput()).includes('Working…')); + + terminal.input('\x1b'); + terminal.input('\x1b'); + // This driver records the stop but leaves the turn parked, standing in for a + // tool held through its process termination grace. Acceptance is a local + // fact, so the strip flips now rather than after that cleanup lands. + await waitFor(() => plainTerminalOutput(terminal.screenOutput()).includes('Cancelling…')); + assert.doesNotMatch(plainTerminalOutput(terminal.screenOutput()), /Working…/); + // Fast acknowledgement must not fake completion: the turn is still running, + // and durable terminal convergence is still owed. + assert.equal(terminal.progressStates.at(-1), true); + assert.equal(driver.stopCalls, 1); + + driver.endTurn(); + await waitFor(() => terminal.progressStates.at(-1) === false); + // A frame painted after convergence no longer claims cancellation is in + // progress, and the editor takes input again. + terminal.input('next'); + await waitFor(() => plainTerminalOutput(terminal.screenOutput()).includes('next')); + assert.doesNotMatch(plainTerminalOutput(terminal.screenOutput()), /Cancelling…/); + + terminal.input('\x03'); + exitMaka(terminal); + await run; + }); + + test('interrupts through the driver authority instead of composing retract and stop', async () => { + const terminal = new FakeTerminal(); + const driver = new InterruptAuthorityDriver(); + const run = runMakaPiTui({ + title: 'Maka', + driver, + cwd: '/repo', + model: 'claude-sonnet-4-5', + connectionSlug: 'claude-subscription', + permissionMode: 'ask', + terminal, + }); + + terminal.input('run'); + terminal.input('\r'); + await waitFor(() => terminal.progressStates.at(-1) === true); + + terminal.input('\x1b'); + terminal.input('\x1b'); + await waitFor(() => terminal.progressStates.at(-1) === false); + + // One authoritative operation owns the queue fence, the retraction, and the + // abort — the CLI no longer sequences `queue.retract` then `turn.stop`. + assert.equal(driver.interruptCalls, 1); + assert.equal(driver.retractCalls, 0); + assert.equal(driver.stopCalls, 0); + // Its retracted entries are what comes back for re-editing. + await waitFor(() => + plainTerminalOutput(terminal.screenOutput()).includes('idea the authority gave back'), + ); + + terminal.input('\x03'); + exitMaka(terminal); + await run; + }); + test('exits on a second Ctrl-C while a turn interrupt is still in flight', async () => { const terminal = new FakeTerminal(); const driver = new SlowStopDriver(); @@ -6614,6 +6728,7 @@ class InterruptibleTurnDriver implements MakaSessionDriver { stopCalls = 0; readonly prompts: string[] = []; private releaseTurn: (() => void) | null = null; + private stopped = false; async listSessions(): Promise { return []; @@ -6621,16 +6736,24 @@ class InterruptibleTurnDriver implements MakaSessionDriver { preparePrompt(prompt: string): Promise { this.prompts.push(prompt); + // The turn exists from here on, so its abort state is armed here — not on + // first pull. A Host turn's event buffer is created by preparePrompt too. + this.stopped = false; return prepareTestPrompt(this, prompt); } async *compactSession(): AsyncIterable {} async *promptEvents(_prompt: string): AsyncIterable { - // The turn parks like a real long-running provider call until stop() aborts it. - await new Promise((resolve) => { - this.releaseTurn = resolve; - }); + // The turn parks like a real long-running provider call until stop() aborts + // it. Cancellation is level-triggered, matching the Host channel: an abort + // that lands before the drain pulls the first event is still observed, + // rather than being dropped because nobody was parked to receive it. + if (!this.stopped) { + await new Promise((resolve) => { + this.releaseTurn = resolve; + }); + } yield { type: 'abort', id: 'event-abort', @@ -6642,6 +6765,7 @@ class InterruptibleTurnDriver implements MakaSessionDriver { async stop(): Promise { this.stopCalls += 1; + this.stopped = true; this.releaseTurn?.(); this.releaseTurn = null; } @@ -6695,6 +6819,9 @@ class SteeringTurnDriver implements MakaSessionDriver { ): Promise { const turnId = options.turnId ?? 'turn-1'; this.turnOrchestrations.push(options.turnOrchestration); + // Armed here, not on first pull: a stop between turn creation and the first + // event pull must still end the turn (see InterruptibleTurnDriver). + this.turnEnded = false; return Promise.resolve({ sessionId: this.getSessionId(), turnId, @@ -6726,7 +6853,6 @@ class SteeringTurnDriver implements MakaSessionDriver { } async *promptEvents(_prompt: string, turnId: string): AsyncIterable { - this.turnEnded = false; for (;;) { while (this.pendingEvents.length > 0) yield this.pendingEvents.shift()!; if (this.turnEnded) break; @@ -7015,6 +7141,139 @@ class DeferredRetryDriver extends FallbackSteeringDriver { } } +// A turn that parks like InterruptibleTurnDriver, plus a steer RPC that never +// settles — the enqueue barrier the interrupt used to wait behind. +class StuckEnqueueDriver implements MakaSessionDriver { + stopCalls = 0; + steerCalls = 0; + private releaseTurn: (() => void) | null = null; + private stopped = false; + + async listSessions(): Promise { + return []; + } + + preparePrompt(prompt: string): Promise { + this.stopped = false; + return prepareTestPrompt(this, prompt); + } + + async *compactSession(): AsyncIterable {} + + async *promptEvents(): AsyncIterable { + if (!this.stopped) { + await new Promise((resolve) => { + this.releaseTurn = resolve; + }); + } + yield { type: 'abort', id: 'event-abort', turnId: 'turn-1', ts: 1, reason: 'user_stop' }; + } + + steer(_text: string): Promise { + this.steerCalls += 1; + // Never settles. The text stays owned by this request, so it is restored by + // the enqueue's own failure path — never by the interrupt waiting on it. + return new Promise(() => {}); + } + + async stop(): Promise { + this.stopCalls += 1; + this.stopped = true; + this.releaseTurn?.(); + this.releaseTurn = null; + } + + async respondToSandboxBoundary(_response: SandboxBoundaryResponse): Promise {} + async renameSession(): Promise {} + async setModel(): Promise {} + async setPermissionMode(): Promise {} + async setThinkingLevel(): Promise {} + async switchSession(sessionId: string): Promise { + return switchResult(fakeSessionSummary(sessionId)); + } + async listRewindTargets(): Promise { + return []; + } + async rewindToTurn(): Promise { + throw new Error('rewind not supported in this fake'); + } + startNewSession(): void {} + getSessionId(): string { + return 'session-1'; + } +} + +// A driver that owns cancellation as one operation, the way the Runtime Host's +// `turn.interrupt` does: it commits the queue fence, returns what it retracted, +// and aborts the turn. `stop()`/`retractQueued()` stay here only to prove the +// runner stops composing them once the authority exists. +class InterruptAuthorityDriver implements MakaSessionDriver { + stopCalls = 0; + retractCalls = 0; + interruptCalls = 0; + private releaseTurn: (() => void) | null = null; + private stopped = false; + + async listSessions(): Promise { + return []; + } + + preparePrompt(prompt: string): Promise { + this.stopped = false; + return prepareTestPrompt(this, prompt); + } + + async *compactSession(): AsyncIterable {} + + async *promptEvents(): AsyncIterable { + if (!this.stopped) { + await new Promise((resolve) => { + this.releaseTurn = resolve; + }); + } + yield { type: 'abort', id: 'event-abort', turnId: 'turn-1', ts: 1, reason: 'user_stop' }; + } + + async retractQueued(): Promise { + this.retractCalls += 1; + return ''; + } + + async interruptTurn(): Promise { + this.interruptCalls += 1; + this.stopped = true; + this.releaseTurn?.(); + this.releaseTurn = null; + return 'idea the authority gave back'; + } + + async stop(): Promise { + this.stopCalls += 1; + this.stopped = true; + this.releaseTurn?.(); + this.releaseTurn = null; + } + + async respondToSandboxBoundary(_response: SandboxBoundaryResponse): Promise {} + async renameSession(): Promise {} + async setModel(): Promise {} + async setPermissionMode(): Promise {} + async setThinkingLevel(): Promise {} + async switchSession(sessionId: string): Promise { + return switchResult(fakeSessionSummary(sessionId)); + } + async listRewindTargets(): Promise { + return []; + } + async rewindToTurn(): Promise { + throw new Error('rewind not supported in this fake'); + } + startNewSession(): void {} + getSessionId(): string { + return 'session-1'; + } +} + class SlowStopDriver implements MakaSessionDriver { stopCalls = 0; readonly prompts: string[] = []; diff --git a/packages/cli/src/__tests__/runtime-host-session-driver.test.ts b/packages/cli/src/__tests__/runtime-host-session-driver.test.ts index 1b9dc7abb1..4a5a42b72e 100644 --- a/packages/cli/src/__tests__/runtime-host-session-driver.test.ts +++ b/packages/cli/src/__tests__/runtime-host-session-driver.test.ts @@ -1130,6 +1130,75 @@ describe('Runtime Host Maka Session driver', () => { ); }); + test('cancels a running turn as one Host operation that returns the retracted queue', async () => { + const subscription = new FakeSubscription(continuitySnapshot(), Promise.resolve([])); + const connection = new FakeConnection([subscription]); + const driver = createRuntimeHostMakaSessionDriver({ + connection: connection.value, + cwd: '/tmp', + llmConnectionSlug: 'openai-main', + model: 'gpt-5', + newId: sequenceIds('interrupt-1'), + }); + await driver.switchSession('session-1'); + + // One operation, not a retract followed by a stop: the Host commits the queue + // stop fence, retracts, and aborts the owning turn atomically, so no message + // can be consumed in a gap between two client calls. + assert.equal(await driver.interruptTurn!(), 'Still queued\n\nAlso queued'); + assert.deepEqual( + connection.requests.filter( + (request) => + request.operation === 'turn.interrupt' || + request.operation === 'queue.retract' || + request.operation === 'turn.stop', + ), + [ + { + operation: 'turn.interrupt', + input: { + originHostEpoch: 'host-1', + sessionId: 'session-1', + interruptId: 'interrupt-1', + turnId: 'turn-1', + runId: 'run-1', + }, + }, + ], + ); + }); + + test('retracts without interrupting when no turn owns the cancellation', async () => { + const subscription = new FakeSubscription( + continuitySnapshot({ rootTurn: completedTurn('turn-1', 'run-1') }), + Promise.resolve([]), + ); + const connection = new FakeConnection([subscription]); + const driver = createRuntimeHostMakaSessionDriver({ + connection: connection.value, + cwd: '/tmp', + llmConnectionSlug: 'openai-main', + model: 'gpt-5', + newId: sequenceIds('retract-1'), + }); + await driver.switchSession('session-1'); + + // A terminal turn has nothing left to abort, but the queue can still hold + // entries the user wants back — retracting alone beats reporting nothing. + assert.equal(await driver.interruptTurn!(), 'Later'); + assert.deepEqual( + connection.requests + .filter( + (request) => + request.operation === 'turn.interrupt' || + request.operation === 'queue.retract' || + request.operation === 'turn.stop', + ) + .map((request) => request.operation), + ['queue.retract'], + ); + }); + test('projects the acknowledgement that releases a question answered through the Host', async () => { const subscription = new FakeSubscription( continuitySnapshot({ interactions: { pending: [pendingQuestion()] } }), @@ -1657,6 +1726,34 @@ class FakeConnection { goal: this.goalQueryResults.shift() ?? null, } as OperationOutput; } + if (operation === 'turn.interrupt') { + const interrupt = input as OperationInput<'turn.interrupt'>; + return { + queueRevision: 4, + retracted: [ + { + entryId: 'entry-1', + messageId: 'message-1', + content: { text: 'Still queued' }, + placement: 'current_turn', + }, + { + entryId: 'entry-2', + messageId: 'message-2', + content: { text: 'Also queued' }, + placement: 'next_turn', + }, + ], + turn: { + sessionId: interrupt.sessionId, + turnId: interrupt.turnId, + runId: interrupt.runId, + status: 'aborted', + completedAt: 90, + terminalEventId: `terminal-${interrupt.turnId}`, + }, + } as OperationOutput; + } if (operation === 'session.configuration.update') { const update = input as OperationInput<'session.configuration.update'>; return { diff --git a/packages/cli/src/pi-transcript.ts b/packages/cli/src/pi-transcript.ts index 35de5bbd80..dd842ecfb1 100644 --- a/packages/cli/src/pi-transcript.ts +++ b/packages/cli/src/pi-transcript.ts @@ -196,6 +196,13 @@ export interface MakaPiTranscriptMetadata { modelContextWindow?: number; /** Elapsed milliseconds of the running agent turn, for the activity strip. */ turnElapsedMs?: number; + /** + * Elapsed milliseconds since a turn interrupt was accepted, for the activity + * strip's `Cancelling…` counter. Set from local gesture recognition, not from + * the runtime's terminal convergence — the counter is what makes a slow + * cleanup readable instead of looking like an ignored keypress. + */ + interruptElapsedMs?: number; providerRetry?: ProviderRetryEvent; /** Resolved locale for primary TUI guidance. Defaults to English for direct embeddings. */ uiLocale?: UiLocale; @@ -1400,14 +1407,25 @@ export function renderMakaPiStatusLine(metadata: MakaPiTranscriptMetadata, width /** * One-line activity strip shown between the transcript and the editor. - * Renders `Working… ` while a turn runs, or a blank reserved row when idle - * so the layout does not jump when a turn starts or ends. + * Renders `Cancelling… ` once an interrupt is accepted, `Working… ` + * while a turn runs, or a blank reserved row when idle so the layout does not + * jump when a turn starts or ends. */ export function renderMakaPiActivityStrip( metadata: MakaPiTranscriptMetadata, width: number, ): string { const safeWidth = Math.max(1, width); + // Cancellation outranks both other states: the abort supersedes a scheduled + // provider retry, and the turn is no longer working towards anything the user + // asked for. The elapsed counter keeps a slow cleanup — a tool holding a + // process through its termination grace — legible as progress. + if (metadata.interruptElapsedMs !== undefined) { + return fitLine( + ansi.dim(`Cancelling… ${formatElapsedDuration(metadata.interruptElapsedMs)}`), + safeWidth, + ); + } if (metadata.providerRetry) { const retry = metadata.providerRetry; const text = diff --git a/packages/cli/src/pi-tui-runner.ts b/packages/cli/src/pi-tui-runner.ts index 61d335b547..e5ece0c247 100644 --- a/packages/cli/src/pi-tui-runner.ts +++ b/packages/cli/src/pi-tui-runner.ts @@ -358,6 +358,11 @@ export async function runMakaPiTui(input: MakaPiTuiInput): Promise { let turnEpoch = 0; let turnStartedAt: number | undefined; let interruptRequested = false; + // When the interrupt gesture was accepted, for the activity strip's + // `Cancelling…` counter. Set in the same tick as recognition so acceptance is + // visible without waiting on the cancellation authority, and cleared with the + // rest of the turn's UI state. + let interruptRequestedAt: number | undefined; // True while a mid-turn detach-switch is in flight: an interrupt issued in // that window would target the freshly attached Session instead of the Turn // being left behind. @@ -453,6 +458,8 @@ export async function runMakaPiTui(input: MakaPiTuiInput): Promise { usage: state.usage, modelContextWindow, turnElapsedMs: turnStartedAt !== undefined ? Date.now() - turnStartedAt : undefined, + interruptElapsedMs: + interruptRequestedAt !== undefined ? Date.now() - interruptRequestedAt : undefined, providerRetry: state.providerRetry, uiLocale: locale, goal: input.driver.getGoal?.() ?? null, @@ -820,31 +827,51 @@ export async function runMakaPiTui(input: MakaPiTuiInput): Promise { } }; + // Cancellation authority. A driver with `interruptTurn` commits the queue stop + // fence, retracts, and aborts the owning turn as one Host operation; without + // it, compose the two calls in that same order, since a message consumed + // between them would otherwise be lost rather than returned for re-editing. + const interruptTurnThroughDriver = async (): Promise => { + const interrupt = input.driver.interruptTurn; + if (interrupt) return interrupt.call(input.driver); + const retracted = (await input.driver.retractQueued?.()) ?? ''; + await input.driver.stop(); + return retracted; + }; + const requestTurnInterrupt = () => { // A detach in flight is not the running Turn's owner acting on it — the // driver already points at the next Session, so a stop here would abort // whatever that Session has attached. Swallow until the handoff settles. if (interruptRequested || detaching) return; interruptRequested = true; + interruptRequestedAt = Date.now(); // The convergence window (stop issued, turn not yet terminal) accepts no // new input: submits would race the abort and could open work the user // just cancelled. The normal turn finally restores submit; a rejected // stop restores it here. editor.disableSubmit = true; + // Renders `Cancelling…` in this tick. Acceptance is a local fact and must + // not wait on the authority: backend abort, tool cleanup, process + // termination grace, and durable terminal publication all land after this. requestRender(); - // The authority retracts before stop: only messages still queued come back - // for re-editing, while anything already consumed stays in the transcript. - // Serializing these operations also preserves that ordering over a Host - // connection where both calls are asynchronous. void (async () => { + // Cancellation goes out before any client-side queue barrier. Pending + // enqueue Promises are `turn.message.submit` round trips, which can be + // delayed by transport, Session admission, storage, or a fallback retry; + // settling them first put an unbounded wait in front of the abort. + // Ordering is still exact, because the authority serializes against its + // own fence: an enqueue that committed before the fence comes back in + // `retracted`, and one that lost the race rejects and restores its own + // text through the enqueue catch — each message survives exactly once. + const retracted = await interruptTurnThroughDriver(); await settlePendingEnqueues(); - const retracted = (await input.driver.retractQueued?.()) ?? ''; const fallback = await takePendingFallbackSettled(); refillEditorFromQueues([fallback, retracted].filter(Boolean).join('\n\n')); requestRender(); - await input.driver.stop(); })().catch((error) => { interruptRequested = false; + interruptRequestedAt = undefined; editor.disableSubmit = false; reportError(error); }); @@ -1199,6 +1226,7 @@ export async function runMakaPiTui(input: MakaPiTuiInput): Promise { turnStartedAt = Date.now(); startTurnElapsedTicker(); interruptRequested = false; + interruptRequestedAt = undefined; lastTurnEscapeAt = 0; editor.disableSubmit = false; setTaskbarProgress(true); @@ -1212,6 +1240,7 @@ export async function runMakaPiTui(input: MakaPiTuiInput): Promise { turnStartedAt = undefined; stopTurnElapsedTicker(); interruptRequested = false; + interruptRequestedAt = undefined; editor.disableSubmit = false; setTaskbarProgress(false); attention.promptTurnEnded(); diff --git a/packages/cli/src/runtime-host-session-driver.ts b/packages/cli/src/runtime-host-session-driver.ts index 6b7cf0f2a8..b94ee4654c 100644 --- a/packages/cli/src/runtime-host-session-driver.ts +++ b/packages/cli/src/runtime-host-session-driver.ts @@ -375,6 +375,21 @@ class RuntimeHostMakaSessionDriverImpl implements RuntimeHostMakaSessionDriver { return result.retracted.map((entry) => entry.content.text).join('\n\n'); } + async interruptTurn(): Promise { + const turn = this.#channel?.snapshot.rootTurn; + // No owning turn to interrupt: the queue can still hold entries the user + // wants back, so retract on its own rather than reporting nothing. + if (!turn || isTerminalTurn(turn)) return this.retractQueued(); + const result = await this.#request('turn.interrupt', { + originHostEpoch: this.#connection.hostEpoch, + sessionId: turn.sessionId, + interruptId: this.#newId(), + turnId: turn.turnId, + runId: turn.runId, + }); + return result.retracted.map((entry) => entry.content.text).join('\n\n'); + } + async respondToSandboxBoundary(response: SandboxBoundaryResponse): Promise { const sessionId = this.#requireSession('respond to permission'); const pending = this.#channel?.pendingInteraction(response.requestId); diff --git a/packages/cli/src/session-driver.ts b/packages/cli/src/session-driver.ts index 42e179efab..60ec6e31aa 100644 --- a/packages/cli/src/session-driver.ts +++ b/packages/cli/src/session-driver.ts @@ -101,6 +101,16 @@ export interface MakaSessionDriver { queueMessage?(text: string): Promise; takePendingFollowup?(): Promise; retractQueued?(): Promise; + /** + * Cancel the running turn as one authoritative step: commit the queue stop + * fence, retract what was still queued, and abort the owning turn. Returns + * the retracted text in `retractQueued()`'s joined form. + * + * A driver exposing this owns the ordering itself, so the caller never has to + * land a queue mutation before it can ask for cancellation. Callers fall back + * to `retractQueued()` followed by `stop()` when it is absent. + */ + interruptTurn?(): Promise; respondToSandboxBoundary(response: SandboxBoundaryResponse): Promise; respondToUserQuestion?(response: UserQuestionResponse): Promise; setModel(model: string, connectionSlug?: string): Promise;