From ad2d5a69cbd79e450688d7c40fa46b22da77c329 Mon Sep 17 00:00:00 2001 From: rbalachandar Date: Sun, 23 Aug 2026 21:01:21 +0200 Subject: [PATCH 1/5] 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/5] 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/5] 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) { From bb28bcd6eae2baa5ad9ca07f73dc01dbb89b8d5a Mon Sep 17 00:00:00 2001 From: rbalachandar Date: Tue, 25 Aug 2026 20:06:28 +0200 Subject: [PATCH 4/5] fix(cli): hold mid-turn input when optional queue methods are absent MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit steer/queueMessage are optional on the public MakaSessionDriver type, but the simplified Enter / Alt+Enter handlers returned early when a driver omitted them — after the editor was cleared, silently dropping the text for custom drivers conforming to the public contract. Route the missing-method case into the same durable admission handoff the first-session window uses: the pending bar shows the held texts and the turn boundary returns undeliverable entries to the editor as an editable draft. Adds a regression test with a driver that omits the optional methods entirely; it fails on the previous head. --- .../cli/src/__tests__/pi-tui-runner.test.ts | 131 ++++++++++++++++++ packages/cli/src/pi-tui-runner.ts | 15 +- 2 files changed, 144 insertions(+), 2 deletions(-) diff --git a/packages/cli/src/__tests__/pi-tui-runner.test.ts b/packages/cli/src/__tests__/pi-tui-runner.test.ts index c79d81fea6..0766771e60 100644 --- a/packages/cli/src/__tests__/pi-tui-runner.test.ts +++ b/packages/cli/src/__tests__/pi-tui-runner.test.ts @@ -1941,6 +1941,61 @@ describe('Maka Pi TUI runner', () => { await run; }); + test('a driver without the optional queue methods holds mid-turn input instead of dropping it', async () => { + const terminal = new FakeTerminal(); + // `steer`/`queueMessage` are optional on `MakaSessionDriver`: a custom + // driver may omit them entirely, so mid-turn input must land in the + // durable handoff rather than vanish. + const driver = new NoQueueMethodsDriver(); + 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 with no driver methods to call: both are held and + // shown in the pending bar, not silently dropped. + terminal.input('held by enter'); + terminal.input('\r'); + terminal.input('held by alt-enter'); + terminal.input('\x1b\r'); + await waitFor(() => { + const screen = plainTerminalOutput(terminal.screenOutput()); + return ( + screen.includes('Steering: held by enter') && screen.includes('Queued: held by alt-enter') + ); + }); + // The first prepare is still parked on release(), so nothing has been + // recorded yet; the held texts are visible only in the pending bar. + assert.deepEqual(driver.prompts, []); + + // The first turn completes; the boundary cannot deliver through missing + // methods, so both held texts come back as one editable draft. + driver.release(); + await waitFor(() => { + const screen = plainTerminalOutput(terminal.screenOutput()); + return ( + terminal.progressStates.at(-1) === false && + screen.includes('held by enter') && + screen.includes('held by alt-enter') && + !screen.includes('Steering: held by enter') + ); + }); + + terminal.input('\x03'); + 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(); @@ -6623,6 +6678,82 @@ class AdmissionWindowDriver implements MakaSessionDriver { } } +/** + * A custom driver exercising the optional shape of `MakaSessionDriver`: no + * `steer`, `queueMessage`, or `retractQueued` methods at all. The first turn's + * prepare parks until `release()`, so mid-turn input happens while a Turn is + * running and the runner must hold it in the durable handoff instead of + * dropping it; the boundary returns undeliverable entries to the editor. + */ +class NoQueueMethodsDriver implements MakaSessionDriver { + stopCalls = 0; + 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 parks until release(), keeping it running for the + // mid-turn submissions under test. + 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(), + }; + } + + async *promptEvents(): AsyncIterable { + yield { + type: 'complete', + id: 'complete-1', + turnId: `turn-${this.prompts.length}`, + ts: 1, + stopReason: 'end_turn', + }; + } + + async *compactSession(): AsyncIterable {} + + /** Let the parked first turn proceed. */ + release(): void { + this.releaseAdmission?.(); + this.releaseAdmission = null; + } + + 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 'session-1'; + } +} + class SlowStopDriver implements MakaSessionDriver { stopCalls = 0; readonly prompts: string[] = []; diff --git a/packages/cli/src/pi-tui-runner.ts b/packages/cli/src/pi-tui-runner.ts index 9eb9d1d63b..e2b298a073 100644 --- a/packages/cli/src/pi-tui-runner.ts +++ b/packages/cli/src/pi-tui-runner.ts @@ -922,7 +922,14 @@ export async function runMakaPiTui(input: MakaPiTuiInput): Promise { } editor.addToHistory(text); const enqueue = input.driver.steer; - if (!enqueue) return; + if (!enqueue) { + // Optional on the public driver type: a custom driver without it must + // not lose the submitted text (review finding on silent drops). The + // durable admission handoff renders it in the pending bar and the turn + // boundary returns undeliverable entries to the editor. + holdForAdmission(text, 'steer'); + return; + } const task = enqueue .call(input.driver, text) .then((outcome) => { @@ -960,7 +967,11 @@ export async function runMakaPiTui(input: MakaPiTuiInput): Promise { } editor.addToHistory(text); const enqueue = input.driver.queueMessage; - if (!enqueue) return; + if (!enqueue) { + // Same optional-method handoff as Enter above: hold instead of drop. + holdForAdmission(text, 'queue'); + return; + } const task = enqueue .call(input.driver, text) .then((outcome) => { From 512fb27c4648942f55aded33a4d242790a59f8d9 Mon Sep 17 00:00:00 2001 From: rbalachandar Date: Tue, 25 Aug 2026 20:16:42 +0200 Subject: [PATCH 5/5] refactor(runtime): drop the embedded coarse text queue API Section 3 of #3556. The runtime-kernel steer / queueMessage / drainFollowup / retractQueue coarse text queue and its four pass-through methods on SessionManager have no production call sites: desktop steers via Host turn.message.submit, the CLI drives the Host, and assertEmbeddedMessageQueue already throws in hosted mode. The only consumers were their own tests. The decisive check from the issue passes: neither ARCHITECTURE.md nor docs/ describes an embedded (Host-less) SessionManager composition as a supported product shape, so the block goes together with its tests. Hosted-mode delivery keeps the SessionSteeringState lease/pull path via bindMessage, which is out of scope and stays. --- .../src/__tests__/session-manager.test.ts | 786 +----------------- packages/runtime/src/runtime-kernel.ts | 83 -- packages/runtime/src/session-manager.ts | 20 - 3 files changed, 1 insertion(+), 888 deletions(-) diff --git a/packages/runtime/src/__tests__/session-manager.test.ts b/packages/runtime/src/__tests__/session-manager.test.ts index ab2a4dc8d1..9de8c80641 100644 --- a/packages/runtime/src/__tests__/session-manager.test.ts +++ b/packages/runtime/src/__tests__/session-manager.test.ts @@ -48,7 +48,7 @@ import type { UserMessageInput, } from '@maka/core/runtime-inputs'; import type { ExecutionBoundary } from '@maka/core/sandbox-boundary'; -import type { QueueEnqueueOutcome, SessionEvent, ShellRunSnapshotResult } from '@maka/core/events'; +import type { SessionEvent, ShellRunSnapshotResult } from '@maka/core/events'; import type { AgentGraphIntentClaim, AgentGraphIntentClaimStore, @@ -120,7 +120,6 @@ import { WEB_RESEARCH_AGENT_ID, } from '../agent-catalog.js'; import { - RuntimeMessageAuthorityInvariantError, type RuntimeHostedRootAuthority, type RuntimeMessageRunIdentity, } from '../message-authority.js'; @@ -14768,27 +14767,6 @@ describe('SessionManager steering and followup queues', () => { return { manager, store }; } - // Run a turn and invoke `duringFirstDelta` synchronously the first time the - // turn streams text — the point at which a real user would type while the - // agent works. Returns every streamed event. - async function runTurnWith( - manager: SessionManager, - sessionId: string, - turnId: string, - duringFirstDelta: () => void, - ): Promise { - const events: SessionEvent[] = []; - let fired = false; - for await (const event of manager.sendMessage(sessionId, { turnId, text: 'hello' })) { - events.push(event); - if (!fired && event.type === 'text_delta') { - fired = true; - duringFirstDelta(); - } - } - return events; - } - test('hosted root runs consume the Host owner and release it exactly once', async () => { const store = new MemorySessionStore(); const runStore = new MemoryAgentRunStore(); @@ -14852,20 +14830,6 @@ describe('SessionManager steering and followup queues', () => { ), ).toBe(true); expect(events.some((event) => event.type === 'queue_update')).toBe(false); - for (const operation of [ - () => manager.steer(session.id, 'runtime mirror'), - () => manager.queueMessage(session.id, 'runtime mirror'), - () => manager.drainFollowup(session.id), - () => manager.retractQueue(session.id), - ]) { - let error: unknown; - try { - operation(); - } catch (caught) { - error = caught; - } - expect(error instanceof RuntimeMessageAuthorityInvariantError).toBe(true); - } }); test('hosted Interaction binds the durable Run identity and closes before release', async () => { @@ -15039,463 +15003,6 @@ describe('SessionManager steering and followup queues', () => { expect(releases).toBe(1); }); - test('a failed turn begin never leaks a steering owner', async () => { - const store = new MemorySessionStore(); - const runStore = new MemoryAgentRunStore(); - const backends = new BackendRegistry(); - let failBuilds = 1; - backends.register('ai-sdk', (ctx) => { - if (failBuilds > 0) { - failBuilds -= 1; - throw new Error('backend build failed'); - } - return new FakeBackend(ctx); - }); - const manager = new SessionManager({ - store, - runStore, - runtimeEventStore: runStore, - backends, - newId: nextId(), - now: nextNow(1_000), - }); - const session = await manager.createSession(makeInput({ permissionMode: 'bypass' })); - - let failed: unknown; - try { - for await (const _event of manager.sendMessage(session.id, { - turnId: 'turn-fail', - text: 'hello', - })) { - // drain - } - } catch (error) { - failed = error; - } - expect((failed as Error).message).toBe('backend build failed'); - - // The failed begin must not have left a live owner: steering falls back - // instead of queueing a message no run will ever consume. - expect(manager.steer(session.id, 'orphaned')).toEqual({ kind: 'fallback' }); - expect(manager.queueMessage(session.id, 'orphaned too')).toEqual({ kind: 'fallback' }); - - // A later successful turn establishes ownership normally. - let outcome: QueueEnqueueOutcome | undefined; - const events = await runTurnWith(manager, session.id, 'turn-2', () => { - outcome = manager.steer(session.id, 'now consumed'); - }); - expect(outcome?.kind).toBe('queued'); - expect( - events.some( - (event) => event.type === 'steering_message' && event.content.text === 'now consumed', - ), - ).toBe(true); - }); - - test('an overlapping turn cannot drain steering queued for the current owner', async () => { - const store = new MemorySessionStore(); - const runStore = new MemoryAgentRunStore(); - const backends = new BackendRegistry(); - let backend: GatedSteeringBackend | undefined; - backends.register('ai-sdk', (ctx) => { - backend = new GatedSteeringBackend(ctx); - return backend; - }); - const manager = new SessionManager({ - store, - runStore, - runtimeEventStore: runStore, - backends, - newId: nextId(), - now: nextNow(1_000), - }); - const session = await manager.createSession(makeInput({ permissionMode: 'bypass' })); - - const first = drainAll(manager.sendMessage(session.id, { turnId: 'turn-a', text: 'first' })); - await waitUntil(() => backend?.gates.has('turn-a') === true); - const second = drainAll(manager.sendMessage(session.id, { turnId: 'turn-b', text: 'second' })); - await waitUntil(() => backend?.gates.has('turn-b') === true); - - // turn-b established ownership last, so the steer targets it. - expect(manager.steer(session.id, 'for the owner').kind).toBe('queued'); - - // The stale turn's pull hook fails the identity check and drains nothing. - backend?.release('turn-a'); - const firstEvents = await first; - expect(backend?.pulls.get('turn-a')).toEqual([[]]); - expect(firstEvents.some((event) => event.type === 'steering_message')).toBe(false); - - // The owner drains exactly the queued message. - backend?.release('turn-b'); - const secondEvents = await second; - expect(backend?.pulls.get('turn-b')).toEqual([['for the owner']]); - expect( - secondEvents.some( - (event) => event.type === 'steering_message' && event.content.text === 'for the owner', - ), - ).toBe(true); - }); - - test('a pulled lease is past the retract point: retract excludes it and it delivers exactly once', async () => { - // Round-5 F1/D1: pull() is the single atomic commit point. Once leased, - // the message belongs to this turn's delivery — a retract during the - // (slow) durable append returns only still-queued text, never the - // in-flight lease; otherwise the retracted text would ALSO be executed by - // the provider once the append lands (refill + execute = two copies). - const gate = makeGate(); - const parked = makeGate(); - class GatedRuntimeEventStore extends MemoryAgentRunStore { - override async appendRuntimeEvent( - sessionId: string, - runId: string, - event: RuntimeEvent, - ): Promise { - if ( - event.content?.kind === 'text' && - (event.content as { steering?: boolean }).steering === true - ) { - parked.release(); - await gate.promise; - } - return super.appendRuntimeEvent(sessionId, runId, event); - } - } - const runStore = new GatedRuntimeEventStore(); - const model = steeringToolThenDoneModel(); - const { manager, session } = await steeringDeliverySession( - runStore, - model, - (manager, sessionId) => { - expect(manager.steer(sessionId, 'urgent steer').kind).toBe('queued'); - }, - ); - - const turnEvents: SessionEvent[] = []; - const turn = (async () => { - for await (const event of manager.sendMessage(session.id, { turnId: 'turn-1', text: 'go' })) { - turnEvents.push(event); - } - })(); - await parked.promise; - // The steering append has not committed: the next provider request must - // not have started while the message is not durable. - await new Promise((resolve) => setTimeout(resolve, 25)); - expect(model.doStreamCalls.length).toBe(1); - // Pulled means committed to this turn: retract returns nothing. - expect(manager.retractQueue(session.id)).toBe(''); - gate.release(); - await turn; - // The message delivered exactly once: in the next provider request… - expect(model.doStreamCalls.length).toBe(2); - expect(JSON.stringify(model.doStreamCalls[1]?.prompt).includes('urgent steer')).toBe(true); - // …echoed once in the stream/ledger… - expect(turnEvents.filter((event) => event.type === 'steering_message').length).toBe(1); - // …and owned by no queue afterwards. - expect(manager.drainFollowup(session.id)).toBe(null); - expect(manager.retractQueue(session.id)).toBe(''); - }); - - test('an abort never converts a durably appended steering message into a redelivery', async () => { - // Round-5 F1/D3: abort does not settle a pushed lease — settlement is - // decided only by the persistence fact. Here the append is parked when - // the stop arrives; once it commits, the message belongs to the ledger - // (history replay presents it to the next turn) and must NOT also be - // nacked into the followup queue, which would put the same directive in - // the account twice. - const gate = makeGate(); - const parked = makeGate(); - class GatedRuntimeEventStore extends MemoryAgentRunStore { - override async appendRuntimeEvent( - sessionId: string, - runId: string, - event: RuntimeEvent, - ): Promise { - if ( - event.content?.kind === 'text' && - (event.content as { steering?: boolean }).steering === true - ) { - parked.release(); - await gate.promise; - } - return super.appendRuntimeEvent(sessionId, runId, event); - } - } - const runStore = new GatedRuntimeEventStore(); - const model = steeringToolThenDoneModel(); - const { manager, session } = await steeringDeliverySession( - runStore, - model, - (manager, sessionId) => { - expect(manager.steer(sessionId, 'urgent steer').kind).toBe('queued'); - }, - ); - - const turn = (async () => { - try { - for await (const _event of manager.sendMessage(session.id, { - turnId: 'turn-1', - text: 'go', - })) { - // drain - } - } catch { - // the abort may end the stream abruptly - } - })(); - await parked.promise; - void manager.stopSession(session.id, { source: 'stop_button' }); - // Let the abort reach the backend's durability wait while the append is - // still parked — the exact window where an abort-settles-the-lease bug - // nacks a message that then also commits to the ledger. - await new Promise((resolve) => setTimeout(resolve, 25)); - gate.release(); - // Teardown converges: the parked append commits, the lease settles, and - // the aborted send terminates without hanging. - await turn; - - // The dying request was never sent… - expect(model.doStreamCalls.length).toBe(1); - // …the ledger owns the message (exactly one durable steering event)… - const runs = await runStore.listSessionRuns(session.id); - const steeringEvents: RuntimeEvent[] = []; - for (const run of runs) { - const events = await runStore.readRuntimeEvents(session.id, run.runId); - steeringEvents.push( - ...events.filter( - (event) => - event.content?.kind === 'text' && - (event.content as { steering?: boolean }).steering === true, - ), - ); - } - expect(steeringEvents.length).toBe(1); - // …and no queue redelivers it. - expect(manager.drainFollowup(session.id)).toBe(null); - expect(manager.retractQueue(session.id)).toBe(''); - }); - - test('a nack that lands after the owner released folds into the followup queue, not an ownerless steering queue', async () => { - // Round-5 F3: turn A's append fails only after turn B took over and - // released. A's nack can no longer target A (it will never pull again) — - // the text's only safe home is the followup queue, exactly where a - // release-time fold would have put it. - const gate = makeGate(); - const parked = makeGate(); - class ParkThenFailStore extends MemoryAgentRunStore { - override async appendRuntimeEvent( - sessionId: string, - runId: string, - event: RuntimeEvent, - ): Promise { - if ( - event.content?.kind === 'text' && - (event.content as { steering?: boolean }).steering === true - ) { - parked.release(); - await gate.promise; - throw new Error('steering append failed'); - } - return super.appendRuntimeEvent(sessionId, runId, event); - } - } - const runStore = new ParkThenFailStore(); - const model = steeringToolThenDoneModel(); - const { manager, session } = await steeringDeliverySession( - runStore, - model, - (manager, sessionId) => { - manager.steer(sessionId, 'urgent steer'); - }, - ); - - const turnA = (async () => { - try { - for await (const _event of manager.sendMessage(session.id, { - turnId: 'turn-1', - text: 'go', - })) { - // drain - } - } catch { - // the failed append ends the stream abruptly - } - })(); - await parked.promise; - // Turn B takes ownership and releases it while A is parked. - for await (const _event of manager.sendMessage(session.id, { - turnId: 'turn-2', - text: 'second', - })) { - // drain - } - gate.release(); - await turnA; - - // The failed message is redeliverable exactly once, via followup. - expect(manager.drainFollowup(session.id)).toBe('urgent steer'); - expect(manager.retractQueue(session.id)).toBe(''); - }); - - test('steer falls back when no RuntimeEventStore is configured', async () => { - // Round-5 F4: without a runtime event ledger, the steering durability ack - // has nothing to anchor to — the fail-closed persist contract cannot be - // honored. The fallback path opens a fresh turn whose user message is - // persisted by the SessionStore, keeping the same durability guarantee. - const store = new MemorySessionStore(); - const backends = new BackendRegistry(); - let backend: GatedSteeringBackend | undefined; - backends.register('ai-sdk', (ctx) => { - backend = new GatedSteeringBackend(ctx); - return backend; - }); - const manager = new SessionManager({ - store, - backends, - newId: nextId(), - now: nextNow(1_000), - }); - const session = await manager.createSession(makeInput({ permissionMode: 'bypass' })); - - const turn = drainAll(manager.sendMessage(session.id, { turnId: 'turn-1', text: 'go' })); - await waitUntil(() => backend?.gates.has('turn-1') === true); - // A live turn exists, but steering cannot be made durable: fall back. - expect(manager.steer(session.id, 'no ledger')).toEqual({ kind: 'fallback' }); - // Followups are unaffected — they open a normal turn anyway. - expect(manager.queueMessage(session.id, 'later').kind).toBe('queued'); - backend?.gates.get('turn-1')?.release(); - backend?.pullDone.get('turn-1')?.release(); - await turn; - }); - - test('a failed steering append nacks the lease back to the queue and the request never carries it', async () => { - // Fail-CLOSED persistence: the steering append throws, the ack judgment - // propagates the failure (no fail-open swallow), the lease is nacked back - // to the queue (folded into followup at release), and neither the ledger - // nor the projection carries the undelivered message. - class FailingSteeringStore extends MemoryAgentRunStore { - override async appendRuntimeEvent( - sessionId: string, - runId: string, - event: RuntimeEvent, - ): Promise { - if ( - event.content?.kind === 'text' && - (event.content as { steering?: boolean }).steering === true - ) { - throw new Error('steering append failed'); - } - return super.appendRuntimeEvent(sessionId, runId, event); - } - } - const runStore = new FailingSteeringStore(); - const model = steeringToolThenDoneModel(); - const { manager, session } = await steeringDeliverySession( - runStore, - model, - (manager, sessionId) => { - expect(manager.steer(sessionId, 'urgent steer').kind).toBe('queued'); - }, - ); - - let failed: unknown; - try { - for await (const _event of manager.sendMessage(session.id, { - turnId: 'turn-1', - text: 'go', - })) { - // drain - } - } catch (error) { - failed = error; - } - expect(failed instanceof Error).toBe(true); - // The dying request never carried the steering: no second provider call. - expect(model.doStreamCalls.length).toBe(1); - // Nacked back to the queue and folded into followup at release — the - // text is redeliverable, not lost. - expect(manager.drainFollowup(session.id)).toBe('urgent steer'); - // Ledger and projection agree: the message was never persisted. - const messages = await manager.getMessages(session.id); - expect( - messages.some((message) => message.type === 'user' && message.text === 'urgent steer'), - ).toBe(false); - }); - - test('an overlapping turn cannot turn a delivered lease into a followup redelivery', async () => { - // Round-4 V1: turn A leases the steer and parks in the (gated) durable - // append; turn B starts meanwhile and takes the owner slot. A's append - // then commits and A's provider request carries the message — so A's ack - // MUST still settle the lease (it is keyed by issuer, not by the current - // owner), and B's teardown must not fold A's in-flight lease into the - // followup queue, which would redeliver an already-executed directive. - const gate = makeGate(); - const parked = makeGate(); - class GatedRuntimeEventStore extends MemoryAgentRunStore { - override async appendRuntimeEvent( - sessionId: string, - runId: string, - event: RuntimeEvent, - ): Promise { - if ( - event.content?.kind === 'text' && - (event.content as { steering?: boolean }).steering === true - ) { - parked.release(); - await gate.promise; - } - return super.appendRuntimeEvent(sessionId, runId, event); - } - } - const runStore = new GatedRuntimeEventStore(); - const model = steeringToolThenDoneModel(); - const { manager, session } = await steeringDeliverySession( - runStore, - model, - (manager, sessionId) => { - manager.steer(sessionId, 'urgent steer'); - }, - ); - - const turnAEvents: SessionEvent[] = []; - const turnA = (async () => { - try { - for await (const event of manager.sendMessage(session.id, { - turnId: 'turn-1', - text: 'go', - })) { - turnAEvents.push(event); - } - } catch { - // A gated teardown may end the stream abruptly. - } - })(); - await parked.promise; - - // Turn B runs to completion while A is parked mid-lease. - for await (const _event of manager.sendMessage(session.id, { - turnId: 'turn-2', - text: 'second', - })) { - // drain - } - expect(model.doStreamCalls.length).toBe(2); - - gate.release(); - await turnA; - - // A's post-steer request went out carrying the directive exactly once… - expect(model.doStreamCalls.length).toBe(3); - expect(JSON.stringify(model.doStreamCalls[2]?.prompt).includes('urgent steer')).toBe(true); - // …B's request never did… - expect(JSON.stringify(model.doStreamCalls[1]?.prompt).includes('urgent steer')).toBe(false); - // …the ledger echoes it exactly once… - expect(turnAEvents.filter((event) => event.type === 'steering_message').length).toBe(1); - // …and NOTHING redelivers it: the delivered lease was acked by its - // issuer, so no queue still holds the text. - expect(manager.drainFollowup(session.id)).toBe(null); - expect(manager.retractQueue(session.id)).toBe(''); - }); - test('a backend-forged queue_update never reaches the ledger or observers', async () => { // Round-6 R3: the kernel is the only legal producer of queue_update (it // pushes them directly into the turn stream). A backend that yields one @@ -15568,106 +15075,6 @@ describe('SessionManager steering and followup queues', () => { ), ).toBe(false); }); - - test('an append error after the write landed settles by the ledger read-back, not a duplicate nack', async () => { - // Round-6 R5: appendRuntimeEvent can fail AFTER the bytes landed (e.g. a - // close error). Treating every append error as not-durable would nack a - // message the ledger already owns — history replay plus the followup - // redelivery equals a double. The ambiguous failure is settled by reading - // the ledger back: present ⇒ durable ⇒ ack path. - class WriteThenThrowStore extends MemoryAgentRunStore { - override async appendRuntimeEvent( - sessionId: string, - runId: string, - event: RuntimeEvent, - ): Promise { - if ( - event.content?.kind === 'text' && - (event.content as { steering?: boolean }).steering === true - ) { - await super.appendRuntimeEvent(sessionId, runId, event); - throw new Error('close failed after the write landed'); - } - return super.appendRuntimeEvent(sessionId, runId, event); - } - } - const runStore = new WriteThenThrowStore(); - const model = steeringToolThenDoneModel(); - const { manager, session } = await steeringDeliverySession( - runStore, - model, - (manager, sessionId) => { - manager.steer(sessionId, 'urgent steer'); - }, - ); - - const turnEvents: SessionEvent[] = []; - for await (const event of manager.sendMessage(session.id, { turnId: 'turn-1', text: 'go' })) { - turnEvents.push(event); - } - - // Delivered exactly once: the next request carries it… - expect(model.doStreamCalls.length).toBe(2); - expect(JSON.stringify(model.doStreamCalls[1]?.prompt).includes('urgent steer')).toBe(true); - // …the ledger owns exactly one copy… - const runs = await runStore.listSessionRuns(session.id); - const steeringEvents: RuntimeEvent[] = []; - for (const run of runs) { - const events = await runStore.readRuntimeEvents(session.id, run.runId); - steeringEvents.push( - ...events.filter( - (event) => - event.content?.kind === 'text' && - (event.content as { steering?: boolean }).steering === true, - ), - ); - } - expect(steeringEvents.length).toBe(1); - // …and no queue redelivers it. - expect(manager.drainFollowup(session.id)).toBe(null); - expect(manager.retractQueue(session.id)).toBe(''); - }); - - test('stranded steering emits a final queue snapshot when it folds into the followup queue', async () => { - const store = new MemorySessionStore(); - const runStore = new MemoryAgentRunStore(); - const backends = new BackendRegistry(); - let backend: GatedSteeringBackend | undefined; - backends.register('ai-sdk', (ctx) => { - backend = new GatedSteeringBackend(ctx); - return backend; - }); - const manager = new SessionManager({ - store, - runStore, - runtimeEventStore: runStore, - backends, - newId: nextId(), - now: nextNow(1_000), - }); - const session = await manager.createSession(makeInput({ permissionMode: 'bypass' })); - - const turn = drainAll(manager.sendMessage(session.id, { turnId: 'turn-1', text: 'go' })); - await waitUntil(() => backend?.gates.has('turn-1') === true); - backend?.gates.get('turn-1')?.release(); - // The turn's only step boundary has already pulled (empty)… - await waitUntil(() => backend?.pulls.has('turn-1') === true); - // …so this steer is stranded: no step is left to consume it. - expect(manager.steer(session.id, 'late').kind).toBe('queued'); - backend?.pullDone.get('turn-1')?.release(); - const events = await turn; - - // The stranded → followup migration is a queue change; the LAST snapshot - // in the stream reflects it, not the stale pre-fold state. - const updates = events.filter( - (event): event is Extract => - event.type === 'queue_update', - ); - expect(updates.at(-1)?.steering).toEqual([]); - expect(updates.at(-1)?.followup).toEqual(['late']); - // And the followup queue is the authoritative owner of the text. - expect(manager.drainFollowup(session.id)).toBe('late'); - }); }); async function drainAll(iterable: AsyncIterable): Promise { @@ -15676,181 +15083,6 @@ async function drainAll(iterable: AsyncIterable): Promise boolean | Promise): Promise { - for (let i = 0; i < 500 && !(await predicate()); i += 1) { - await new Promise((resolve) => setTimeout(resolve, 2)); - } - expect(await predicate()).toBe(true); -} - -/** Mock model: first request calls the Probe tool, second finishes with text. */ -function steeringToolThenDoneModel(): MockLanguageModelV4 { - const usage = { - inputTokens: { total: 100, noCache: 100, cacheRead: 0, cacheWrite: 0 }, - outputTokens: { total: 10, text: 10, reasoning: 0 }, - }; - const model: MockLanguageModelV4 = new MockLanguageModelV4({ - doStream: async () => { - const call = model.doStreamCalls.length; - const chunks: LanguageModelV4StreamPart[] = - call === 1 - ? [ - { type: 'stream-start', warnings: [] }, - { - type: 'tool-call', - toolCallId: 'tool-1', - toolName: 'Probe', - input: JSON.stringify({ q: 'x' }), - }, - { type: 'finish', finishReason: { unified: 'tool-calls', raw: 'tool_calls' }, usage }, - ] - : [ - { type: 'stream-start', warnings: [] }, - { type: 'text-start', id: 'text-1' }, - { type: 'text-delta', id: 'text-1', delta: 'done' }, - { type: 'text-end', id: 'text-1' }, - { type: 'finish', finishReason: { unified: 'stop', raw: 'stop' }, usage }, - ]; - return { - stream: simulateReadableStream({ chunks, initialDelayInMs: null, chunkDelayInMs: null }), - }; - }, - }); - return model; -} - -/** - * A SessionManager wired to a REAL AiSdkBackend over a mock model, so the - * full steering delivery chain (kernel lease -> backend durability wait -> - * AgentRun fail-closed persist) is exercised. `duringTool` runs inside the - * first step's tool execution — the moment a real user steers. - */ -async function steeringDeliverySession( - runStore: MemoryAgentRunStore, - model: MockLanguageModelV4, - duringTool: (manager: SessionManager, sessionId: string) => Promise | void, -) { - const store = new MemorySessionStore(); - const backends = new BackendRegistry(); - let manager!: SessionManager; - let sessionId = ''; - backends.register('ai-sdk', (ctx) => - createTestAiSdkBackend({ - sessionId: ctx.sessionId, - header: ctx.header, - appendMessage: async () => {}, - connection: { - slug: 'mock-main', - providerType: 'anthropic', - defaultModel: 'mock-model-id', - }, - apiKey: 'sk-test', - modelId: 'mock-model-id', - modelFactory: () => model, - tools: [ - { - name: 'Probe', - description: 'Probe description', - parameters: z.object({ q: z.string() }), - impl: async () => { - await duringTool(manager, sessionId); - return { ok: true }; - }, - }, - ], - loadTurnRuntimeEvents: ctx.loadTurnRuntimeEvents, - allowMidTurnHistoryCompaction: ctx.allowMidTurnHistoryCompaction, - newId: nextId(), - now: nextNow(1), - }), - ); - const managerDeps = { - store, - runStore, - runtimeEventStore: runStore, - backends, - newId: nextId(), - now: nextNow(1_000), - }; - manager = new SessionManager(managerDeps); - const session = await manager.createSession(makeInput({ permissionMode: 'bypass' })); - sessionId = session.id; - return { manager, session, store }; -} - -/** - * Parks each send behind a per-turn gate, pulls steering exactly once after - * release, then parks again behind a post-pull gate before finishing — a - * deterministic harness for the owner-identity rule and for enqueues that - * land after the final step boundary (stranded steering). - */ -class GatedSteeringBackend implements AgentBackend { - readonly kind = 'ai-sdk' as const; - readonly sessionId: string; - readonly gates = new Map(); - readonly pullDone = new Map(); - readonly pulls = new Map(); - - constructor(ctx: BackendFactoryContext) { - this.sessionId = ctx.sessionId; - } - - async *send(input: BackendSendInput): AsyncIterable { - const gate = makeGate(); - const afterPull = makeGate(); - this.gates.set(input.turnId, gate); - this.pullDone.set(input.turnId, afterPull); - await gate.promise; - const leases = input.pullSteering?.() ?? []; - const record = this.pulls.get(input.turnId) ?? []; - record.push(leases.map((lease) => lease.content.text)); - this.pulls.set(input.turnId, record); - let seq = 0; - for (const lease of leases) { - seq += 1; - yield { - type: 'steering_message', - id: `${input.turnId}-steer-${seq}`, - turnId: input.turnId, - ts: seq, - messageId: lease.messageId, - content: lease.content, - }; - } - // Delivery for this fake is the echo itself; ack the leases. - input.ackSteering?.(leases.map((lease) => lease.id)); - await afterPull.promise; - yield { - type: 'text_complete', - id: `${input.turnId}-final`, - turnId: input.turnId, - ts: 10, - messageId: `${input.turnId}-m`, - text: 'ok', - }; - yield { - type: 'complete', - id: `${input.turnId}-complete`, - turnId: input.turnId, - ts: 11, - stopReason: 'end_turn', - }; - } - - /** Release both of a turn's gates (start + post-pull). */ - release(turnId: string): void { - this.gates.get(turnId)?.release(); - this.pullDone.get(turnId)?.release(); - } - - async stop(): Promise { - for (const turnId of this.gates.keys()) this.release(turnId); - } - - async respondToSandboxBoundary(_decision: SandboxBoundaryResponse): Promise {} - - async dispose(): Promise {} -} class DelegatingRuntimeKernel implements RuntimeKernelLike { readonly starts: Array<{ @@ -15941,22 +15173,6 @@ class DelegatingRuntimeKernel implements RuntimeKernelLike { this.permissionResponses.push(sessionId); } - steer(): QueueEnqueueOutcome { - return { kind: 'fallback' }; - } - - queueMessage(): QueueEnqueueOutcome { - return { kind: 'fallback' }; - } - - drainFollowup(): string | null { - return null; - } - - retractQueue(): string { - return ''; - } - hasActiveRuns(): boolean { return this.activeRuns; } diff --git a/packages/runtime/src/runtime-kernel.ts b/packages/runtime/src/runtime-kernel.ts index fa6043b72c..e4dccc7a88 100644 --- a/packages/runtime/src/runtime-kernel.ts +++ b/packages/runtime/src/runtime-kernel.ts @@ -28,7 +28,6 @@ import { isSessionInlineRun } from '@maka/core/agent-run'; import type { ActiveInteractionRequestEvent, CompleteEvent, - QueueEnqueueOutcome, QueueUpdateEvent, SessionEvent, TokenUsageEvent, @@ -168,14 +167,6 @@ export interface RuntimeKernelLike { respondToSandboxBoundary(sessionId: string, response: SandboxBoundaryResponse): Promise; listActiveInteractions?(sessionId: string): ActiveInteractionRequestEvent[]; respondToUserQuestion?(sessionId: string, response: UserQuestionResponse): Promise; - /** Queue a user message for mid-turn injection at the next step boundary. */ - steer(sessionId: string, text: string): QueueEnqueueOutcome; - /** Queue a user message to open the turn after the current one finishes. */ - queueMessage(sessionId: string, text: string): QueueEnqueueOutcome; - /** Drain the followup queue into one `\n\n`-joined prompt, or null if empty. */ - drainFollowup(sessionId: string): string | null; - /** Take back every queued message (both queues) as one `\n\n`-joined string. */ - retractQueue(sessionId: string): string; hasActiveRuns(sessionId: string): boolean; /** * The turns of the runs in flight for this session. The same fact @@ -2397,61 +2388,6 @@ export class RuntimeKernel implements RuntimeKernelLike { // Steering / followup queues (authoritative source of truth) // -------------------------------------------------------------------------- - steer(sessionId: string, text: string): QueueEnqueueOutcome { - this.assertEmbeddedMessageQueue('steer'); - // Steering's delivery contract is anchored to the runtime event ledger - // (fail-closed persist + durable-consume ack). Without a RuntimeEventStore - // that anchor does not exist — same condition as requireTerminalWrite — - // so fall back to a fresh turn, whose user message the SessionStore - // persists with the ordinary turn-open guarantee. - if (!this.deps.runtimeEventStore) return { kind: 'fallback' }; - // Double responsibility (codex): with no live steering owner to inject - // into — the turn just ended, begin() failed, or only child/compact runs - // are active (they never consume this queue) — tell the caller to open a - // fresh turn instead so the message is never dropped. - const state = this.liveSteeringState(sessionId); - if (!state) return { kind: 'fallback' }; - const messageId = this.deps.newId(); - state.steering.push({ id: messageId, messageId, content: { text } }); - this.emitQueueUpdate(sessionId, state); - return { kind: 'queued' }; - } - - queueMessage(sessionId: string, text: string): QueueEnqueueOutcome { - this.assertEmbeddedMessageQueue('queueMessage'); - const state = this.liveSteeringState(sessionId); - if (!state) return { kind: 'fallback' }; - state.followup.push(text); - this.emitQueueUpdate(sessionId, state); - return { kind: 'queued' }; - } - - drainFollowup(sessionId: string): string | null { - this.assertEmbeddedMessageQueue('drainFollowup'); - const state = this.steeringBySession.get(sessionId); - if (!state || state.followup.length === 0) return null; - const drained = state.followup.splice(0); - this.emitQueueUpdate(sessionId, state); - return drained.join('\n\n'); - } - - retractQueue(sessionId: string): string { - this.assertEmbeddedMessageQueue('retractQueue'); - const state = this.steeringBySession.get(sessionId); - if (!state) return ''; - // Retract reclaims QUEUED messages only. pull() is the single atomic - // commit point of delivery: an in-flight lease is already committed to - // the running turn — its durable append may land at any moment, so - // handing its text back to the user here would refill AND execute the - // same directive. An in-flight lease settles only by the persistence - // fact (ack when the ledger owns it, nack back to a queue otherwise). - const all = [...state.steering.map((message) => message.content.text), ...state.followup]; - state.steering = []; - state.followup = []; - this.emitQueueUpdate(sessionId, state); - return all.join('\n\n'); - } - private ensureSteering(sessionId: string): SessionSteeringState { const existing = this.steeringBySession.get(sessionId); if (existing) return existing; @@ -2460,25 +2396,6 @@ export class RuntimeKernel implements RuntimeKernelLike { return created; } - private assertEmbeddedMessageQueue(operation: string): void { - if (this.deps.messageAuthority) { - throw new RuntimeMessageAuthorityInvariantError( - `Hosted Runtime cannot ${operation}; the Runtime Host owns message admission and queues`, - ); - } - } - - /** - * The session's steering state only while a steering-capable top-level run - * owns it (sink registered after begin() succeeded and not yet released). - * Child agent and compact runs never establish ownership, so their activity - * alone yields undefined — enqueue must fall back rather than strand text. - */ - private liveSteeringState(sessionId: string): SessionSteeringState | undefined { - const state = this.steeringBySession.get(sessionId); - return state?.sink ? state : undefined; - } - private emitQueueUpdate(sessionId: string, state: SessionSteeringState): void { state.sink?.({ type: 'queue_update', diff --git a/packages/runtime/src/session-manager.ts b/packages/runtime/src/session-manager.ts index 582bafe686..2c172350fa 100644 --- a/packages/runtime/src/session-manager.ts +++ b/packages/runtime/src/session-manager.ts @@ -41,7 +41,6 @@ import type { AbortEvent, PermissionDecisionAckEvent, PermissionRequestEvent, - QueueEnqueueOutcome, ShellRunUpdate, MessageContent, } from '@maka/core/events'; @@ -4812,25 +4811,6 @@ export class SessionManager { : this.runtimeKernel.stopSession(identity.sessionId, input); } - /** Queue a user message for mid-turn injection at the next step boundary. */ - steer(sessionId: string, text: string): QueueEnqueueOutcome { - return this.runtimeKernel.steer(sessionId, text); - } - - /** Queue a user message to open the turn after the current one finishes. */ - queueMessage(sessionId: string, text: string): QueueEnqueueOutcome { - return this.runtimeKernel.queueMessage(sessionId, text); - } - - /** Drain the followup queue into one `\n\n`-joined prompt, or null if empty. */ - drainFollowup(sessionId: string): string | null { - return this.runtimeKernel.drainFollowup(sessionId); - } - - /** Take back every queued message (both queues) as one `\n\n`-joined string. */ - retractQueue(sessionId: string): string { - return this.runtimeKernel.retractQueue(sessionId); - } async *regenerateTurn( sessionId: string,