From ad2d5a69cbd79e450688d7c40fa46b22da77c329 Mon Sep 17 00:00:00 2001 From: rbalachandar Date: Sun, 23 Aug 2026 21:01:21 +0200 Subject: [PATCH 1/3] refactor(cli): drop dead fallback retry and followup-takeover paths MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Runtime Host is now the complete queue authority (per-entry queue ops, protocol epoch 40, #3544), which leaves two CLI code paths with no production consumer. Covers sections 1 and 2 of #3556; section 3 (runtime-kernel embedded queue API) is intentionally left for a separate PR pending maintainer confirmation. Section 1 — fallback retry machinery: the only production driver (RuntimeHostMakaSessionDriverImpl) returns `fallback` solely when no sessionId exists, while every fallback producer (steer during a running turn, alt+enter queue) requires a live turn and therefore a session. Removes the retry timer loop, deferred-fallback state, turn-boundary flush, pending-bar merge, and their tests. trackEnqueue / settlePendingEnqueues stay: the interrupt path still needs in-flight submit ordering. Section 2 — takePendingFollowup: the production stub is always null (the Host starts queued follow-ups atomically; returning text would make the TUI double-submit), so the runner's re-queue/nextPrompt fold was unreachable outside test doubles. Removes the interface method, stub, consumer block, and the doubles' implementations. Verified as still needed and kept: Host op queue.retract (CLI interrupt / alt+up), sessions:steer/enqueue IPC, session_busy fallback in sessions:send, and the QueueUpdateEvent steering/followup mirrors the pending bar renders. --- .../cli/src/__tests__/pi-transcript.test.ts | 2 - .../cli/src/__tests__/pi-tui-runner.test.ts | 454 ------------------ packages/cli/src/pi-transcript.ts | 30 +- packages/cli/src/pi-tui-runner.ts | 205 +------- .../cli/src/runtime-host-session-driver.ts | 6 - packages/cli/src/session-driver.ts | 1 - 6 files changed, 20 insertions(+), 678 deletions(-) diff --git a/packages/cli/src/__tests__/pi-transcript.test.ts b/packages/cli/src/__tests__/pi-transcript.test.ts index 72003776aa..9a8abc19fe 100644 --- a/packages/cli/src/__tests__/pi-transcript.test.ts +++ b/packages/cli/src/__tests__/pi-transcript.test.ts @@ -354,7 +354,6 @@ describe('Maka Pi TUI transcript', () => { ); state.entries.push({ kind: 'notice', level: 'error', text: 'Turn failed: provider_error' }); state.steering = ['Keep going']; - state.pendingFallback = [{ text: 'Try again', enqueue: 'steer' }]; assert.equal( hydrateToolsWithStoredMessages(state, 'turn-1', [ @@ -386,7 +385,6 @@ describe('Maka Pi TUI transcript', () => { assert.deepEqual(tool?.input, { path: 'README.md' }); assert.deepEqual(tool?.result, { kind: 'text', text: 'README contents' }); assert.deepEqual(state.steering, ['Keep going']); - assert.deepEqual(state.pendingFallback, [{ text: 'Try again', enqueue: 'steer' }]); assert.equal(state.entries.at(-1)?.kind, 'notice'); }); diff --git a/packages/cli/src/__tests__/pi-tui-runner.test.ts b/packages/cli/src/__tests__/pi-tui-runner.test.ts index 930513d668..0ab4ca2750 100644 --- a/packages/cli/src/__tests__/pi-tui-runner.test.ts +++ b/packages/cli/src/__tests__/pi-tui-runner.test.ts @@ -2200,199 +2200,6 @@ describe('Maka Pi TUI runner', () => { await run; }); - test('a fallback enqueue during a long turn is never dropped and flushes into the next turn', async () => { - const terminal = new FakeTerminal(); - // Every enqueue reports `fallback` — the runtime never has a live owner. - const driver = new FallbackSteeringDriver(); - const run = runMakaPiTui({ - title: 'Maka', - driver, - cwd: '/repo', - model: 'm', - connectionSlug: 'c', - permissionMode: 'bypass', - terminal, - }); - - terminal.input('start the work'); - terminal.input('\r'); - await waitFor(() => terminal.progressStates.at(-1) === true); - - terminal.input('second thought'); - terminal.input('\r'); // steer → fallback → CLI-held pending - terminal.input('and afterwards'); - terminal.input('\x1b\r'); // Alt+Enter → fallback → CLI-held pending - await waitFor(() => { - const screen = plainTerminalOutput(terminal.screenOutput()); - return ( - screen.includes('Steering: second thought') && screen.includes('Queued: and afterwards') - ); - }); - - // The old bounded poll gave up after ~2s of busy (about 20 attempts at the - // 100ms retry cadence) and silently dropped the text. Waiting for the - // driver to observe the retries crossing that budget — instead of guessing - // elapsed time — proves the CLI is still retrying under any scheduler load. - await waitForUpTo(() => driver.steerAttempts > 22 && driver.queueAttempts > 22, 30_000); - const screen = plainTerminalOutput(terminal.screenOutput()); - assert.equal(screen.includes('Steering: second thought'), true); - assert.equal(screen.includes('Queued: and afterwards'), true); - assert.deepEqual(driver.prompts, ['start the work']); - - // The turn boundary flushes the undelivered texts into the next turn. - driver.endTurn(); - await waitFor(() => driver.prompts.length === 2); - assert.equal(driver.prompts[1], 'second thought\n\nand afterwards'); - - await waitForUpTo(() => driver.parked, 1_000); - driver.endTurn(); - await waitFor(() => terminal.progressStates.at(-1) === false); - terminal.input('/exit'); - terminal.input('\r'); - await run; - }); - - test('a fallback steer retries the same enqueue and lands once the owner appears', async () => { - const terminal = new FakeTerminal(); - const driver = new FallbackSteeringDriver(); - driver.steerFallbacks = 2; // the owner appears after ~200ms of retries - const run = runMakaPiTui({ - title: 'Maka', - driver, - cwd: '/repo', - model: 'm', - connectionSlug: 'c', - permissionMode: 'bypass', - terminal, - }); - - terminal.input('start the work'); - terminal.input('\r'); - await waitFor(() => terminal.progressStates.at(-1) === true); - - terminal.input('late owner'); - terminal.input('\r'); // steer → fallback, retried until it lands - await waitForUpTo(() => driver.steered.includes('late owner'), 1_000); - // Landed as a steer of the RUNNING turn — no fresh turn was opened. - assert.deepEqual(driver.prompts, ['start the work']); - await waitFor(() => - plainTerminalOutput(terminal.screenOutput()).includes('Steering: late owner'), - ); - - driver.endTurn(); - await waitFor(() => terminal.progressStates.at(-1) === false); - // Nothing left to flush: the text was delivered mid-turn, not re-queued. - assert.deepEqual(driver.prompts, ['start the work']); - terminal.input('/exit'); - terminal.input('\r'); - await run; - }); - - test('a turn boundary waits for an unresolved enqueue before deciding whether to flush it', async () => { - const terminal = new FakeTerminal(); - const driver = new DeferredAdmissionDriver(); - const run = runMakaPiTui({ - title: 'Maka', - driver, - cwd: '/repo', - model: 'm', - connectionSlug: 'c', - permissionMode: 'bypass', - terminal, - }); - - terminal.input('start'); - terminal.input('\r'); - await waitFor(() => terminal.progressStates.at(-1) === true); - await waitForUpTo(() => driver.parked, 1_000); - terminal.input('late admission'); - terminal.input('\r'); - await waitFor(() => driver.steerCalls === 1); - - driver.endTurn(); - await waitFor(() => driver.completedTurns === 1); - assert.deepEqual(driver.prompts, ['start']); - driver.releaseAdmission({ kind: 'fallback' }); - await waitForUpTo(() => driver.prompts.length === 2, 1_000); - assert.equal(driver.prompts[1], 'late admission'); - - await waitForUpTo(() => driver.parked, 1_000); - driver.endTurn(); - await waitFor(() => terminal.progressStates.at(-1) === false); - terminal.input('/exit'); - terminal.input('\r'); - await run; - }); - - test('a queued retry settling at the turn boundary is not also flushed as a new turn', async () => { - const terminal = new FakeTerminal(); - const driver = new DeferredRetryDriver(); - const run = runMakaPiTui({ - title: 'Maka', - driver, - cwd: '/repo', - model: 'm', - connectionSlug: 'c', - permissionMode: 'bypass', - terminal, - }); - - terminal.input('start'); - terminal.input('\r'); - await waitFor(() => terminal.progressStates.at(-1) === true); - terminal.input('lands on retry'); - terminal.input('\r'); - await waitForUpTo(() => driver.steerCalls === 2, 1_000); - - driver.endTurn(); - driver.releaseRetry(); - await waitFor(() => terminal.progressStates.at(-1) === false); - assert.deepEqual(driver.prompts, ['start']); - assert.deepEqual(driver.delivered, ['lands on retry']); - - terminal.input('/exit'); - terminal.input('\r'); - await run; - }); - - test('interrupt refills CLI-held fallback text into the editor', async () => { - const terminal = new FakeTerminal(); - const driver = new FallbackSteeringDriver(); - const run = runMakaPiTui({ - title: 'Maka', - driver, - cwd: '/repo', - model: 'm', - connectionSlug: 'c', - permissionMode: 'bypass', - terminal, - }); - - terminal.input('start the work'); - terminal.input('\r'); - await waitFor(() => terminal.progressStates.at(-1) === true); - - terminal.input('rescue me'); - terminal.input('\r'); // steer → fallback → CLI-held pending - await waitFor(() => - plainTerminalOutput(terminal.screenOutput()).includes('Steering: rescue me'), - ); - - terminal.input('\x1b'); - terminal.input('\x1b'); // interrupt - await waitFor(() => terminal.progressStates.at(-1) === false); - // The CLI-held text comes back for re-editing; the pending bar clears. - await waitFor(() => { - const screen = plainTerminalOutput(terminal.screenOutput()); - return screen.includes('rescue me') && !screen.includes('Steering: rescue me'); - }); - - terminal.input('\x03'); // clear the refilled draft - terminal.input('/exit'); - terminal.input('\r'); - await run; - }); - test('input during the interrupt convergence window stays in the editor and opens no turn', async () => { const terminal = new FakeTerminal(); const driver = new SlowStopDriver(); // stop() returns but the turn keeps running @@ -2439,49 +2246,6 @@ describe('Maka Pi TUI runner', () => { assert.deepEqual(driver.prompts, ['start the work']); }); - test('an aborted turn never auto-opens the flush turn; undelivered text becomes a draft', async () => { - const terminal = new FakeTerminal(); - const driver = new FallbackSteeringDriver(); // enqueues always fall back - const run = runMakaPiTui({ - title: 'Maka', - driver, - cwd: '/repo', - model: 'm', - connectionSlug: 'c', - permissionMode: 'bypass', - terminal, - }); - - terminal.input('start the work'); - terminal.input('\r'); - await waitFor(() => terminal.progressStates.at(-1) === true); - - terminal.input('next thing'); - terminal.input('\x1b\r'); // Alt+Enter → fallback → CLI-held pending - await waitFor(() => - plainTerminalOutput(terminal.screenOutput()).includes('Queued: next thing'), - ); - - // The turn ends as ABORTED on its own (not via the CLI interrupt path): - // the boundary flush must not open a turn the user just stopped. - driver.abortNextTurn = true; - driver.endTurn(); - await waitFor(() => terminal.progressStates.at(-1) === false); - // The undelivered text is an editable draft, not a queued line. - await waitFor(() => { - const screen = plainTerminalOutput(terminal.screenOutput()); - return screen.includes('next thing') && !screen.includes('Queued: next thing'); - }); - - terminal.input('\x03'); // clear the preserved draft - terminal.input('/exit'); - terminal.input('\r'); - await run; - // Anchored after close: a wrongly-opened flush turn would have landed in - // prompts by the time the TUI has fully shut down. - assert.deepEqual(driver.prompts, ['start the work']); - }); - test('exits on the second Ctrl-C during a control command', async () => { const terminal = new FakeTerminal(); const driver = new DeferredControlDriver(); @@ -6675,13 +6439,6 @@ class SteeringTurnDriver implements MakaSessionDriver { return { kind: 'queued' }; } - async takePendingFollowup(): Promise { - if (this.followup.length === 0) return null; - const joined = this.followup.join('\n\n'); - this.followup = []; - return joined; - } - async retractQueued(): Promise { this.retractCalls += 1; const joined = [...this.steering, ...this.followup].join('\n\n'); @@ -6727,217 +6484,6 @@ class SteeringTurnDriver implements MakaSessionDriver { } } -/** - * A driver whose enqueues hit the no-live-owner `fallback` outcome for the - * first N calls (configurable, default forever) while the turn parks until - * `endTurn()` — the begin-window shape behind review finding N2. - */ -class FallbackSteeringDriver implements MakaSessionDriver { - readonly prompts: string[] = []; - readonly steered: string[] = []; - readonly queuedMessages: string[] = []; - stopCalls = 0; - completedTurns = 0; - /** Enqueue calls that report `fallback` before the owner "appears". */ - steerFallbacks = Number.POSITIVE_INFINITY; - queueFallbacks = Number.POSITIVE_INFINITY; - /** Total enqueue attempts, including rejected ones — the observable retry count. */ - steerAttempts = 0; - queueAttempts = 0; - private steering: string[] = []; - private followup: string[] = []; - private pendingEvents: SessionEvent[] = []; - private wakeTurn: (() => void) | null = null; - private turnOpen = false; - private turnEnded = false; - private eventSeq = 0; - - get parked(): boolean { - return this.turnOpen && !this.turnEnded; - } - - async listSessions(): Promise { - return []; - } - - preparePrompt( - prompt: string, - options: MakaPreparePromptOptions = {}, - ): Promise { - this.prompts.push(options.modelText ?? prompt); - const turnId = options.turnId ?? `turn-${this.prompts.length}`; - return Promise.resolve({ - sessionId: this.getSessionId(), - turnId, - events: this.promptEvents(turnId), - }); - } - - async *compactSession(): AsyncIterable {} - - // Same single-path contract as the runtime: queue contents reach the CLI - // only through `queue_update` events on the turn stream. - private emitQueueUpdate(): void { - this.eventSeq += 1; - this.pendingEvents.push({ - type: 'queue_update', - id: `queue-update-${this.eventSeq}`, - turnId: `turn-${this.prompts.length}`, - ts: this.eventSeq, - steering: [...this.steering], - followup: [...this.followup], - }); - this.wakeTurn?.(); - this.wakeTurn = null; - } - - async *promptEvents(turnId: string): AsyncIterable { - this.turnOpen = true; - this.turnEnded = false; - for (;;) { - while (this.pendingEvents.length > 0) yield this.pendingEvents.shift()!; - if (this.turnEnded) break; - await new Promise((resolve) => { - this.wakeTurn = resolve; - }); - } - this.turnOpen = false; - if (this.abortNextTurn) { - this.abortNextTurn = false; - yield { - type: 'abort', - id: `abort-${this.prompts.length}`, - turnId, - ts: 1, - reason: 'user_stop', - }; - yield { - type: 'complete', - id: `complete-${this.prompts.length}`, - turnId, - ts: 2, - stopReason: 'user_stop', - }; - this.completedTurns += 1; - return; - } - yield { - type: 'complete', - id: `complete-${this.prompts.length}`, - turnId, - ts: 1, - stopReason: 'end_turn', - }; - this.completedTurns += 1; - } - - /** Next endTurn() finishes the turn as aborted instead of end_turn. */ - abortNextTurn = false; - - async steer(text: string): Promise { - this.steerAttempts += 1; - if (this.steerFallbacks > 0) { - this.steerFallbacks -= 1; - return { kind: 'fallback' }; - } - this.steered.push(text); - this.steering.push(text); - this.emitQueueUpdate(); - return { kind: 'queued' }; - } - - async queueMessage(text: string): Promise { - this.queueAttempts += 1; - if (this.queueFallbacks > 0) { - this.queueFallbacks -= 1; - return { kind: 'fallback' }; - } - this.queuedMessages.push(text); - this.followup.push(text); - this.emitQueueUpdate(); - return { kind: 'queued' }; - } - - async takePendingFollowup(): Promise { - if (this.followup.length === 0) return null; - const joined = this.followup.join('\n\n'); - this.followup = []; - return joined; - } - - async retractQueued(): Promise { - const joined = [...this.steering, ...this.followup].join('\n\n'); - this.steering = []; - this.followup = []; - this.emitQueueUpdate(); - return joined; - } - - endTurn(): void { - this.turnEnded = true; - this.wakeTurn?.(); - this.wakeTurn = null; - } - - async stop(): Promise { - this.stopCalls += 1; - this.steering = []; - this.followup = []; - this.endTurn(); - } - - 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 DeferredAdmissionDriver extends FallbackSteeringDriver { - steerCalls = 0; - readonly #admission = deferred(); - - override async steer(_text: string): Promise { - this.steerCalls += 1; - return this.#admission.promise; - } - - releaseAdmission(outcome: QueueEnqueueOutcome): void { - this.#admission.resolve(outcome); - } -} - -class DeferredRetryDriver extends FallbackSteeringDriver { - steerCalls = 0; - readonly delivered: string[] = []; - readonly #retry = deferred(); - - override async steer(text: string): Promise { - this.steerCalls += 1; - if (this.steerCalls === 1) return { kind: 'fallback' }; - await this.#retry.promise; - this.delivered.push(text); - return { kind: 'queued' }; - } - - releaseRetry(): void { - this.#retry.resolve(); - } -} - class SlowStopDriver implements MakaSessionDriver { stopCalls = 0; readonly prompts: string[] = []; diff --git a/packages/cli/src/pi-transcript.ts b/packages/cli/src/pi-transcript.ts index 87433d8e0e..89a957d024 100644 --- a/packages/cli/src/pi-transcript.ts +++ b/packages/cli/src/pi-transcript.ts @@ -104,14 +104,6 @@ export interface MakaPiTranscriptState { */ steering: string[]; followup: string[]; - /** - * Messages whose enqueue hit the no-live-owner fallback while a turn was - * running (the begin window). CLI-owned, NOT a runtime mirror: the runner - * retries the original enqueue until it lands and flushes any remainder - * into the next turn at the turn boundary, so the text is never dropped. - * Rendered in the pending bar alongside the mirror. - */ - pendingFallback: Array<{ text: string; enqueue: 'steer' | 'queue' }>; /** Current non-durable provider retry progress for the activity strip. */ providerRetry?: ProviderRetryEvent; } @@ -216,7 +208,6 @@ export function createMakaPiTranscriptState(): MakaPiTranscriptState { usage: { costUsd: 0, cacheHitInput: 0, cacheMissInput: 0 }, steering: [], followup: [], - pendingFallback: [], }; } @@ -343,7 +334,6 @@ export function replaceTranscriptWithStoredMessages( // Queues are per-active-run; a switched/reset session has none pending. state.steering = []; state.followup = []; - state.pendingFallback = []; for (const msg of messages) { if (msg.type === 'token_usage') accumulateUsage(state.usage, msg); } @@ -1448,26 +1438,12 @@ function formatElapsedDuration(elapsedMs: number): string { * Renders nothing when both queues are empty. */ export function renderMakaPiPendingQueue(state: MakaPiTranscriptState, width: number): string[] { - if ( - state.steering.length === 0 && - state.followup.length === 0 && - state.pendingFallback.length === 0 - ) { + if (state.steering.length === 0 && state.followup.length === 0) { return []; } const safeWidth = Math.max(1, width); - const steering = [ - ...state.steering, - ...state.pendingFallback - .filter((entry) => entry.enqueue === 'steer') - .map((entry) => entry.text), - ]; - const followup = [ - ...state.followup, - ...state.pendingFallback - .filter((entry) => entry.enqueue === 'queue') - .map((entry) => entry.text), - ]; + const steering = state.steering; + const followup = state.followup; const lines: string[] = []; for (const text of steering) { lines.push( diff --git a/packages/cli/src/pi-tui-runner.ts b/packages/cli/src/pi-tui-runner.ts index 79c67bd6ed..b17b8ba7ca 100644 --- a/packages/cli/src/pi-tui-runner.ts +++ b/packages/cli/src/pi-tui-runner.ts @@ -46,7 +46,7 @@ import { slashCommandsForSurface, type SlashCommandIdForSurface, } from '@maka/core/slash-command-catalog'; -import { type QueueEnqueueOutcome, type ShellRunUpdate } from '@maka/core/events'; +import { type ShellRunUpdate } from '@maka/core/events'; import { latestAssistantModelId, type SessionSummary, @@ -719,7 +719,6 @@ export async function runMakaPiTui(input: MakaPiTuiInput): Promise { shellRunHydration.dispose(); shellRunElapsedTicker.dispose(); stopTurnElapsedTicker(); - stopFallbackRetry(); setTaskbarProgress(false); // Drop the busy / attention title marker so the tab is not handed back to // the shell still marked busy when the session exits. @@ -839,8 +838,7 @@ export async function runMakaPiTui(input: MakaPiTuiInput): Promise { void (async () => { await settlePendingEnqueues(); const retracted = (await input.driver.retractQueued?.()) ?? ''; - const fallback = await takePendingFallbackSettled(); - refillEditorFromQueues([fallback, retracted].filter(Boolean).join('\n\n')); + refillEditorFromQueues(retracted); requestRender(); await input.driver.stop(); })().catch((error) => { @@ -867,8 +865,8 @@ export async function runMakaPiTui(input: MakaPiTuiInput): Promise { editor.addToHistory(prompt); if (handleSlashCommand(prompt, idleMs)) return; // First-run has no connection, so the wizard is the only surface. This is - // the single choke point for idle submits (Enter, Alt+Enter, steer - // fallback): reopen the wizard instead of opening a turn against a + // the single choke point for idle submits (Enter and Alt+Enter): reopen + // the wizard instead of opening a turn against a // connection-less driver. Slash commands above already routed to the // command layer (/exit still exits, /help still shows help). if (input.firstRun) { @@ -891,104 +889,7 @@ export async function runMakaPiTui(input: MakaPiTuiInput): Promise { }); }; - // Fallback handoff owner. A `fallback` outcome while the turn is running - // means the runtime has no live steering owner YET (the begin window) or - // just lost it; the runtime keeps no record of the text, so the CLI owns - // delivery: retry the SAME enqueue until the owner appears, and flush any - // remainder into the next turn at the turn boundary. Never a bounded wait — - // a normal turn outlives any fixed budget and the text must not vanish. - const FALLBACK_RETRY_MS = 100; - let fallbackRetryTimer: ReturnType | null = null; - let fallbackRetryInFlight = false; - let fallbackRetryTask: Promise | null = null; - let fallbackRetryGeneration = 0; - - const stopFallbackRetry = () => { - fallbackRetryGeneration += 1; - if (fallbackRetryTimer !== null) clearTimeout(fallbackRetryTimer); - fallbackRetryTimer = null; - }; - - const scheduleFallbackRetry = () => { - if (fallbackRetryTimer !== null || fallbackRetryInFlight) return; - fallbackRetryTimer = setTimeout(() => { - fallbackRetryTimer = null; - const task = retryPendingFallback(); - fallbackRetryTask = task; - void task.finally(() => { - if (fallbackRetryTask === task) fallbackRetryTask = null; - }); - }, FALLBACK_RETRY_MS); - }; - - const retryPendingFallback = async () => { - if (closed || !turnRunning || state.pendingFallback.length === 0) { - stopFallbackRetry(); - return; - } - const generation = fallbackRetryGeneration; - const attempted = [...state.pendingFallback]; - fallbackRetryInFlight = true; - const remaining: typeof state.pendingFallback = []; - let failed = false; - try { - for (const entry of attempted) { - const enqueue = entry.enqueue === 'steer' ? input.driver.steer : input.driver.queueMessage; - let outcome: QueueEnqueueOutcome | undefined; - try { - outcome = enqueue ? await enqueue.call(input.driver, entry.text) : undefined; - } catch (error) { - failed = true; - reportError(error); - } - if (outcome?.kind !== 'queued') remaining.push(entry); - } - } finally { - fallbackRetryInFlight = false; - } - if (generation !== fallbackRetryGeneration) return; - const attemptedEntries = new Set(attempted); - const appended = state.pendingFallback.filter((entry) => !attemptedEntries.has(entry)); - const changed = remaining.length !== attempted.length; - state.pendingFallback = [...remaining, ...appended]; - if (remaining.length === 0) stopFallbackRetry(); - else if (!failed) scheduleFallbackRetry(); - if (!changed) return; - // The queue mirror updates only from `queue_update` events (single path); - // this render just drops the delivered entries from the fallback list. - requestRender(); - }; - - const deferFallback = (text: string, enqueue: 'steer' | 'queue') => { - state.pendingFallback.push({ text, enqueue }); - scheduleFallbackRetry(); - requestRender(); - }; - - /** Drain the CLI-held fallback texts (delivery order), stopping the retry loop. */ - const takePendingFallbackEntries = (): Array<{ text: string; enqueue: 'steer' | 'queue' }> => { - stopFallbackRetry(); - const entries = state.pendingFallback; - state.pendingFallback = []; - return entries; - }; - - const takePendingFallbackEntriesSettled = async (): Promise< - Array<{ text: string; enqueue: 'steer' | 'queue' }> - > => { - if (fallbackRetryTimer !== null) { - clearTimeout(fallbackRetryTimer); - fallbackRetryTimer = null; - } - await fallbackRetryTask; - return takePendingFallbackEntries(); - }; - - const takePendingFallbackSettled = async (): Promise => - (await takePendingFallbackEntriesSettled()).map((entry) => entry.text).join('\n\n'); - - // Enter during a turn steers it (inject at the next step boundary); the - // runtime falls back to a fresh turn if the run already ended. + // Enter during a turn steers it (inject at the next step boundary). const steerRunningTurn = (text: string) => { if (!text.trim()) { requestRender(); @@ -996,19 +897,11 @@ export async function runMakaPiTui(input: MakaPiTuiInput): Promise { } editor.addToHistory(text); const enqueue = input.driver.steer; - if (!enqueue) { - deferFallback(text, 'steer'); - return; - } + if (!enqueue) return; const task = enqueue .call(input.driver, text) - .then((outcome) => { - if (outcome.kind === 'fallback') { - if (turnRunning || busy) deferFallback(text, 'steer'); - else submitPrompt(text); - return; - } - // Queued: the runtime's `queue_update` event refreshes the mirror. + .then(() => { + // The runtime's `queue_update` event refreshes the mirror. requestRender(); }) .catch((error) => { @@ -1037,19 +930,11 @@ export async function runMakaPiTui(input: MakaPiTuiInput): Promise { } editor.addToHistory(text); const enqueue = input.driver.queueMessage; - if (!enqueue) { - deferFallback(text, 'queue'); - return; - } + if (!enqueue) return; const task = enqueue .call(input.driver, text) - .then((outcome) => { - if (outcome.kind === 'fallback') { - if (turnRunning || busy) deferFallback(text, 'queue'); - else submitPrompt(text); - return; - } - // Queued: the runtime's `queue_update` event refreshes the mirror. + .then(() => { + // The runtime's `queue_update` event refreshes the mirror. requestRender(); }) .catch((error) => { @@ -1059,14 +944,13 @@ export async function runMakaPiTui(input: MakaPiTuiInput): Promise { trackEnqueue(task); }; - // Alt+↑: take back every queued message (both queues plus CLI-held fallback - // texts), joined and prepended to the current draft for re-editing. + // Alt+↑: take back every queued message, joined and prepended to the current + // draft for re-editing. const retractQueuedMessages = () => { void (async () => { await settlePendingEnqueues(); const retracted = (await input.driver.retractQueued?.()) ?? ''; - const fallback = await takePendingFallbackSettled(); - refillEditorFromQueues([fallback, retracted].filter(Boolean).join('\n\n')); + refillEditorFromQueues(retracted); requestRender(); })().catch(reportError); }; @@ -1322,9 +1206,8 @@ export async function runMakaPiTui(input: MakaPiTuiInput): Promise { if (superseded()) { // Orphaned by a mid-turn detach (#3380): the Session this turn ran // on is no longer adopted. Skip every continuation that belongs to - // it — queue flushes would steer the NEW Session, fallback texts - // would refill the editor with abandoned-session context, and a - // failure notice would misreport the still-running Host Turn. Only + // it — queue flushes would steer the NEW Session, and a failure notice + // would misreport the still-running Host Turn. Only // release the slot and hand the freshly attached Turn its start; // startPendingAttachedTurn no-ops until applySwitchResult has // installed it and we are idle, and the detach path re-arms it, so @@ -1336,70 +1219,16 @@ export async function runMakaPiTui(input: MakaPiTuiInput): Promise { return outcome; } - // Turn boundary flush: CLI-held fallback texts that never reached the - // runtime (the enqueue retry never found a live owner) are delivered - // FIRST, then queued followups (alt+Enter) — both open the next turn - // before any goal auto-continuation. Consumed here outside the turn - // stream, so clear the local mirror explicitly. + // Wait for enqueue calls already in flight before releasing this turn. await settlePendingEnqueues(); - const fallbackEntries = await takePendingFallbackEntriesSettled(); - const followup = await input.driver.takePendingFollowup?.(); if (outcome.kind === 'completed' && pendingAttachedTurn) { const attached = pendingAttachedTurn; pendingAttachedTurn = undefined; - const undelivered: string[] = []; - for (const entry of fallbackEntries) { - const enqueue = - entry.enqueue === 'steer' ? input.driver.steer : input.driver.queueMessage; - try { - if (!enqueue || (await enqueue.call(input.driver, entry.text)).kind === 'fallback') { - undelivered.push(entry.text); - } - } catch { - undelivered.push(entry.text); - } - } - if (followup) { - try { - if ( - !input.driver.queueMessage || - (await input.driver.queueMessage(followup)).kind === 'fallback' - ) { - undelivered.push(followup); - } - } catch { - undelivered.push(followup); - } - } busy = false; activity.finish(); startAttachedTurn?.(attached); - if (undelivered.length > 0) refillEditorFromQueues(undelivered.join('\n\n')); return outcome; } - const fallbackText = fallbackEntries.map((entry) => entry.text).join('\n\n'); - const nextPrompt = [fallbackText, followup ?? ''].filter(Boolean).join('\n\n'); - if (nextPrompt) { - state.steering = []; - state.followup = []; - if (outcome.kind !== 'completed') { - // The turn was aborted or errored: auto-opening a turn would defeat - // the interrupt (or hammer a failure). Keep the undelivered text as - // an editable draft instead, merged ahead of any current draft. - refillEditorFromQueues(nextPrompt); - } else { - // Install the next local activity before resolving the previous one. - // A Goal admission woken by the old activity therefore observes the - // user follow-up as busy instead of racing it for the session. - void runAgentTurn({ - kind: 'external', - prompt: nextPrompt, - sessionId: input.driver.getSessionId(), - }); - activity.finish(); - return outcome; - } - } busy = false; activity.finish(); diff --git a/packages/cli/src/runtime-host-session-driver.ts b/packages/cli/src/runtime-host-session-driver.ts index ea4e233a80..dd5ae63520 100644 --- a/packages/cli/src/runtime-host-session-driver.ts +++ b/packages/cli/src/runtime-host-session-driver.ts @@ -354,12 +354,6 @@ class RuntimeHostMakaSessionDriverImpl implements RuntimeHostMakaSessionDriver { return this.#enqueue(text, 'next_turn'); } - async takePendingFollowup(): Promise { - // Runtime Host owns the terminal transition and starts the queued follow-up - // atomically. Returning its text here would make the TUI submit it twice. - return null; - } - async retractQueued(): Promise { if (!this.#sessionId) return ''; const result = await this.#request('queue.retract', { diff --git a/packages/cli/src/session-driver.ts b/packages/cli/src/session-driver.ts index 42e179efab..ce58bcd253 100644 --- a/packages/cli/src/session-driver.ts +++ b/packages/cli/src/session-driver.ts @@ -99,7 +99,6 @@ export interface MakaSessionDriver { resumeLatest?(): AsyncIterable; steer?(text: string): Promise; queueMessage?(text: string): Promise; - takePendingFollowup?(): Promise; retractQueued?(): Promise; respondToSandboxBoundary(response: SandboxBoundaryResponse): Promise; respondToUserQuestion?(response: UserQuestionResponse): Promise; From 9bc2587905fb7c4aaba4ddc92af3c9cf9045c274 Mon Sep 17 00:00:00 2001 From: rbalachandar Date: Mon, 24 Aug 2026 21:12:14 +0200 Subject: [PATCH 2/3] fix(cli): hold first-session admission input instead of dropping it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review follow-up on the fallback-path removal. The production driver's #enqueue still returns { kind: 'fallback' } while session.create has not yet assigned a session id, and runAgentTurn sets turnRunning before preparePrompt awaits #ensureSession() — so Enter and Alt+Enter inside that first-session window produced a fallback that the simplified handlers ignored after the editor was cleared, silently dropping the text. Retain a minimal durable handoff for exactly that window: fallback outcomes while a turn is running are held in a CLI-owned pendingAdmission list (rendered in the pending bar), re-enqueued once at the turn boundary, and returned to the editor as an editable draft if still undelivered or the turn aborted. No retry loop — the window is bounded by the first turn. Interrupt exit and alt+up refills merge the held texts, so no path drops them. Held steer-kind text is delivered at the boundary, so it opens the follow-up turn rather than injecting mid-turn into the first turn; the retry loop that could land it mid-turn is deliberately not restored. Adds a delayed-session.create regression test covering the window for both Enter (steer) and Alt+Enter (queue); it fails on the previous head where the fallback outcome was ignored. --- .../cli/src/__tests__/pi-tui-runner.test.ts | 141 ++++++++++++++++++ packages/cli/src/pi-transcript.ts | 30 +++- packages/cli/src/pi-tui-runner.ts | 76 +++++++++- 3 files changed, 240 insertions(+), 7 deletions(-) diff --git a/packages/cli/src/__tests__/pi-tui-runner.test.ts b/packages/cli/src/__tests__/pi-tui-runner.test.ts index 0ab4ca2750..7ecf2dce31 100644 --- a/packages/cli/src/__tests__/pi-tui-runner.test.ts +++ b/packages/cli/src/__tests__/pi-tui-runner.test.ts @@ -1896,6 +1896,53 @@ describe('Maka Pi TUI runner', () => { await run; }); + test('input during the first-session admission window survives to the next turn', async () => { + const terminal = new FakeTerminal(); + // `session.create` is delayed, so the first turn runs before a session id + // exists and enqueues inside the window report `fallback`. + const driver = new AdmissionWindowDriver(); + const run = runMakaPiTui({ + title: 'Maka', + driver, + cwd: '/repo', + model: 'm', + connectionSlug: 'c', + permissionMode: 'bypass', + terminal, + }); + + terminal.input('start'); + terminal.input('\r'); + await waitFor(() => terminal.progressStates.at(-1) === true); + + // Enter and Alt+Enter inside the admission window: both fall back and the + // CLI holds them; nothing is delivered while the session id is missing. + terminal.input('must survive'); + terminal.input('\r'); + terminal.input('and afterwards'); + terminal.input('\x1b\r'); + await waitFor(() => { + const screen = plainTerminalOutput(terminal.screenOutput()); + return ( + screen.includes('Steering: must survive') && screen.includes('Queued: and afterwards') + ); + }); + // The first prepare is still parked on session.create, and nothing is + // delivered while the session id is missing. + assert.deepEqual(driver.prompts, []); + + // session.create resolves and the first turn completes: the turn boundary + // re-enqueues the held texts, and each opens its follow-up turn. + driver.admit(); + await waitFor(() => driver.prompts.length === 3); + assert.deepEqual(driver.prompts, ['start', 'must survive', 'and afterwards']); + await waitFor(() => terminal.progressStates.at(-1) === false); + + terminal.input('/exit'); + terminal.input('\r'); + await run; + }); + test('opens /transcript during a running turn instead of steering it', async () => { const terminal = new FakeTerminal(); const driver = new SteeringTurnDriver(); @@ -6484,6 +6531,100 @@ class SteeringTurnDriver implements MakaSessionDriver { } } +/** + * First-session admission window: `session.create` is slow, so the first turn + * is already running before a session id exists. Enqueues inside the window + * report `fallback` (the production `#enqueue` early-return on a missing + * session id). Once admitted, an enqueue onto an idle session starts the next + * Turn — the same resolution the Host gives `turn.message.submit` on an idle + * session — which this double records as a new prompt. + */ +class AdmissionWindowDriver implements MakaSessionDriver { + stopCalls = 0; + admitted = false; + readonly prompts: string[] = []; + private releaseAdmission: (() => void) | null = null; + private readonly admission: Promise = new Promise((resolve) => { + this.releaseAdmission = resolve; + }); + + async listSessions(): Promise { + return []; + } + + async preparePrompt( + prompt: string, + options: MakaPreparePromptOptions = {}, + ): Promise { + // The first turn's prepare awaits the delayed `session.create`. + if (this.prompts.length === 0) await this.admission; + this.prompts.push(options.modelText ?? prompt); + return { + sessionId: this.getSessionId() ?? 'session-1', + turnId: options.turnId ?? `turn-${this.prompts.length}`, + events: this.promptEvents(), + }; + } + + // Once admitted the turn completes as soon as it opens. + async *promptEvents(): AsyncIterable { + yield { + type: 'complete', + id: 'complete-1', + turnId: `turn-${this.prompts.length}`, + ts: 1, + stopReason: 'end_turn', + }; + } + + async *compactSession(): AsyncIterable {} + + /** Resolve the delayed `session.create`. */ + admit(): void { + this.admitted = true; + this.releaseAdmission?.(); + this.releaseAdmission = null; + } + + async steer(text: string): Promise { + return this.enqueue(text); + } + + async queueMessage(text: string): Promise { + return this.enqueue(text); + } + + private enqueue(text: string): QueueEnqueueOutcome { + if (!this.admitted) return { kind: 'fallback' }; + // Session exists and no Turn is running: the Host starts the next Turn. + this.prompts.push(text); + return { kind: 'queued' }; + } + + async stop(): Promise { + this.stopCalls += 1; + } + + 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 | null { + return this.admitted ? 'session-1' : null; + } +} + class SlowStopDriver implements MakaSessionDriver { stopCalls = 0; readonly prompts: string[] = []; diff --git a/packages/cli/src/pi-transcript.ts b/packages/cli/src/pi-transcript.ts index 89a957d024..4842ed9f5d 100644 --- a/packages/cli/src/pi-transcript.ts +++ b/packages/cli/src/pi-transcript.ts @@ -104,6 +104,14 @@ export interface MakaPiTranscriptState { */ steering: string[]; followup: string[]; + /** + * Messages whose enqueue hit the no-session `fallback` outcome during the + * first Session's admission window (the turn is running but `session.create` + * has not yet assigned a session id). CLI-owned, NOT a runtime mirror: the + * runner holds them durably and re-enqueues at the turn boundary, so the + * text is never dropped. Rendered in the pending bar alongside the mirror. + */ + pendingAdmission: Array<{ text: string; enqueue: 'steer' | 'queue' }>; /** Current non-durable provider retry progress for the activity strip. */ providerRetry?: ProviderRetryEvent; } @@ -208,6 +216,7 @@ export function createMakaPiTranscriptState(): MakaPiTranscriptState { usage: { costUsd: 0, cacheHitInput: 0, cacheMissInput: 0 }, steering: [], followup: [], + pendingAdmission: [], }; } @@ -334,6 +343,7 @@ export function replaceTranscriptWithStoredMessages( // Queues are per-active-run; a switched/reset session has none pending. state.steering = []; state.followup = []; + state.pendingAdmission = []; for (const msg of messages) { if (msg.type === 'token_usage') accumulateUsage(state.usage, msg); } @@ -1438,12 +1448,26 @@ function formatElapsedDuration(elapsedMs: number): string { * Renders nothing when both queues are empty. */ export function renderMakaPiPendingQueue(state: MakaPiTranscriptState, width: number): string[] { - if (state.steering.length === 0 && state.followup.length === 0) { + if ( + state.steering.length === 0 && + state.followup.length === 0 && + state.pendingAdmission.length === 0 + ) { return []; } const safeWidth = Math.max(1, width); - const steering = state.steering; - const followup = state.followup; + const steering = [ + ...state.steering, + ...state.pendingAdmission + .filter((entry) => entry.enqueue === 'steer') + .map((entry) => entry.text), + ]; + const followup = [ + ...state.followup, + ...state.pendingAdmission + .filter((entry) => entry.enqueue === 'queue') + .map((entry) => entry.text), + ]; const lines: string[] = []; for (const text of steering) { lines.push( diff --git a/packages/cli/src/pi-tui-runner.ts b/packages/cli/src/pi-tui-runner.ts index b17b8ba7ca..a20b56c776 100644 --- a/packages/cli/src/pi-tui-runner.ts +++ b/packages/cli/src/pi-tui-runner.ts @@ -838,7 +838,8 @@ export async function runMakaPiTui(input: MakaPiTuiInput): Promise { void (async () => { await settlePendingEnqueues(); const retracted = (await input.driver.retractQueued?.()) ?? ''; - refillEditorFromQueues(retracted); + const held = pendingAdmissionText(); + refillEditorFromQueues([held, retracted].filter(Boolean).join('\n\n')); requestRender(); await input.driver.stop(); })().catch((error) => { @@ -889,6 +890,30 @@ export async function runMakaPiTui(input: MakaPiTuiInput): Promise { }); }; + // First-session admission handoff. The production driver's `#enqueue` + // returns `fallback` while `session.create` has not yet assigned a session + // id, and `runAgentTurn` sets `turnRunning` before `preparePrompt` awaits + // `#ensureSession()` — so Enter / Alt+Enter inside that window produce a + // `fallback` even though a turn is running. Hold that text durably (no + // retry loop: the window is bounded by the first turn) and re-enqueue it at + // the turn boundary; anything still undelivered returns to the editor, so + // user input is never dropped. + const holdForAdmission = (text: string, enqueue: 'steer' | 'queue') => { + state.pendingAdmission.push({ text, enqueue }); + requestRender(); + }; + + const takePendingAdmission = (): Array<{ text: string; enqueue: 'steer' | 'queue' }> => { + const entries = state.pendingAdmission; + state.pendingAdmission = []; + return entries; + }; + + const pendingAdmissionText = (): string => + takePendingAdmission() + .map((entry) => entry.text) + .join('\n\n'); + // Enter during a turn steers it (inject at the next step boundary). const steerRunningTurn = (text: string) => { if (!text.trim()) { @@ -900,7 +925,12 @@ export async function runMakaPiTui(input: MakaPiTuiInput): Promise { if (!enqueue) return; const task = enqueue .call(input.driver, text) - .then(() => { + .then((outcome) => { + if (outcome.kind === 'fallback') { + if (turnRunning || busy) holdForAdmission(text, 'steer'); + else submitPrompt(text); + return; + } // The runtime's `queue_update` event refreshes the mirror. requestRender(); }) @@ -933,7 +963,12 @@ export async function runMakaPiTui(input: MakaPiTuiInput): Promise { if (!enqueue) return; const task = enqueue .call(input.driver, text) - .then(() => { + .then((outcome) => { + if (outcome.kind === 'fallback') { + if (turnRunning || busy) holdForAdmission(text, 'queue'); + else submitPrompt(text); + return; + } // The runtime's `queue_update` event refreshes the mirror. requestRender(); }) @@ -950,7 +985,8 @@ export async function runMakaPiTui(input: MakaPiTuiInput): Promise { void (async () => { await settlePendingEnqueues(); const retracted = (await input.driver.retractQueued?.()) ?? ''; - refillEditorFromQueues(retracted); + const held = pendingAdmissionText(); + refillEditorFromQueues([held, retracted].filter(Boolean).join('\n\n')); requestRender(); })().catch(reportError); }; @@ -1221,6 +1257,38 @@ export async function runMakaPiTui(input: MakaPiTuiInput): Promise { // Wait for enqueue calls already in flight before releasing this turn. await settlePendingEnqueues(); + // Deliver CLI-held admission texts (first-session window) now that the + // turn has settled: re-enqueue via the original steer/queue intent — + // the session id exists by this point, so a queued outcome hands the + // text to the runtime; anything still falling back (or a turn that + // aborted or errored, where auto-opening would defeat the interrupt) + // returns to the editor as an editable draft instead of being dropped. + const admissionEntries = takePendingAdmission(); + if (admissionEntries.length > 0) { + if (outcome.kind !== 'completed') { + refillEditorFromQueues( + admissionEntries.map((entry) => entry.text).join('\n\n'), + ); + } else { + const undelivered: string[] = []; + for (const entry of admissionEntries) { + const enqueue = + entry.enqueue === 'steer' ? input.driver.steer : input.driver.queueMessage; + try { + if ( + !enqueue || + (await enqueue.call(input.driver, entry.text)).kind === 'fallback' + ) { + undelivered.push(entry.text); + } + } catch { + undelivered.push(entry.text); + } + } + if (undelivered.length > 0) refillEditorFromQueues(undelivered.join('\n\n')); + requestRender(); + } + } if (outcome.kind === 'completed' && pendingAttachedTurn) { const attached = pendingAttachedTurn; pendingAttachedTurn = undefined; From b694670681afd8d03bf073f8ad1443e7fbe1597d Mon Sep 17 00:00:00 2001 From: rbalachandar Date: Mon, 24 Aug 2026 21:26:51 +0200 Subject: [PATCH 3/3] fixed formatting errors --- packages/cli/src/__tests__/pi-tui-runner.test.ts | 4 +--- packages/cli/src/pi-tui-runner.ts | 4 +--- 2 files changed, 2 insertions(+), 6 deletions(-) diff --git a/packages/cli/src/__tests__/pi-tui-runner.test.ts b/packages/cli/src/__tests__/pi-tui-runner.test.ts index 7ecf2dce31..c79d81fea6 100644 --- a/packages/cli/src/__tests__/pi-tui-runner.test.ts +++ b/packages/cli/src/__tests__/pi-tui-runner.test.ts @@ -1923,9 +1923,7 @@ describe('Maka Pi TUI runner', () => { terminal.input('\x1b\r'); await waitFor(() => { const screen = plainTerminalOutput(terminal.screenOutput()); - return ( - screen.includes('Steering: must survive') && screen.includes('Queued: and afterwards') - ); + return screen.includes('Steering: must survive') && screen.includes('Queued: and afterwards'); }); // The first prepare is still parked on session.create, and nothing is // delivered while the session id is missing. diff --git a/packages/cli/src/pi-tui-runner.ts b/packages/cli/src/pi-tui-runner.ts index a20b56c776..9eb9d1d63b 100644 --- a/packages/cli/src/pi-tui-runner.ts +++ b/packages/cli/src/pi-tui-runner.ts @@ -1266,9 +1266,7 @@ export async function runMakaPiTui(input: MakaPiTuiInput): Promise { const admissionEntries = takePendingAdmission(); if (admissionEntries.length > 0) { if (outcome.kind !== 'completed') { - refillEditorFromQueues( - admissionEntries.map((entry) => entry.text).join('\n\n'), - ); + refillEditorFromQueues(admissionEntries.map((entry) => entry.text).join('\n\n')); } else { const undelivered: string[] = []; for (const entry of admissionEntries) {