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..0766771e60 100644 --- a/packages/cli/src/__tests__/pi-tui-runner.test.ts +++ b/packages/cli/src/__tests__/pi-tui-runner.test.ts @@ -1896,6 +1896,106 @@ 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('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(); @@ -2200,199 +2300,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 +2346,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 +6539,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'); @@ -6728,162 +6585,77 @@ 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. + * 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 FallbackSteeringDriver implements MakaSessionDriver { - readonly prompts: string[] = []; - readonly steered: string[] = []; - readonly queuedMessages: string[] = []; +class AdmissionWindowDriver implements MakaSessionDriver { 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; - } + admitted = false; + readonly prompts: string[] = []; + private releaseAdmission: (() => void) | null = null; + private readonly admission: Promise = new Promise((resolve) => { + this.releaseAdmission = resolve; + }); async listSessions(): Promise { return []; } - preparePrompt( + 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); - 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; + return { + sessionId: this.getSessionId() ?? 'session-1', + turnId: options.turnId ?? `turn-${this.prompts.length}`, + events: this.promptEvents(), + }; } - 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; - } + // Once admitted the turn completes as soon as it opens. + async *promptEvents(): AsyncIterable { yield { type: 'complete', - id: `complete-${this.prompts.length}`, - turnId, + id: 'complete-1', + turnId: `turn-${this.prompts.length}`, ts: 1, stopReason: 'end_turn', }; - this.completedTurns += 1; } - /** Next endTurn() finishes the turn as aborted instead of end_turn. */ - abortNextTurn = false; + async *compactSession(): AsyncIterable {} - 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' }; + /** Resolve the delayed `session.create`. */ + admit(): void { + this.admitted = true; + this.releaseAdmission?.(); + this.releaseAdmission = null; } - 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 steer(text: string): Promise { + return this.enqueue(text); } - async retractQueued(): Promise { - const joined = [...this.steering, ...this.followup].join('\n\n'); - this.steering = []; - this.followup = []; - this.emitQueueUpdate(); - return joined; + async queueMessage(text: string): Promise { + return this.enqueue(text); } - endTurn(): void { - this.turnEnded = true; - this.wakeTurn?.(); - this.wakeTurn = null; + 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; - this.steering = []; - this.followup = []; - this.endTurn(); } async respondToSandboxBoundary(_response: SandboxBoundaryResponse): Promise {} @@ -6901,40 +6673,84 @@ class FallbackSteeringDriver implements MakaSessionDriver { throw new Error('rewind not supported in this fake'); } startNewSession(): void {} - getSessionId(): string { - return 'session-1'; + getSessionId(): string | null { + return this.admitted ? 'session-1' : null; } } -class DeferredAdmissionDriver extends FallbackSteeringDriver { - steerCalls = 0; - readonly #admission = deferred(); +/** + * 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; + }); - override async steer(_text: string): Promise { - this.steerCalls += 1; - return this.#admission.promise; + async listSessions(): Promise { + return []; } - releaseAdmission(outcome: QueueEnqueueOutcome): void { - this.#admission.resolve(outcome); + 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(), + }; } -} -class DeferredRetryDriver extends FallbackSteeringDriver { - steerCalls = 0; - readonly delivered: string[] = []; - readonly #retry = deferred(); + async *promptEvents(): AsyncIterable { + yield { + type: 'complete', + id: 'complete-1', + turnId: `turn-${this.prompts.length}`, + ts: 1, + stopReason: 'end_turn', + }; + } - 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' }; + async *compactSession(): AsyncIterable {} + + /** Let the parked first turn proceed. */ + release(): void { + this.releaseAdmission?.(); + this.releaseAdmission = null; + } + + async stop(): Promise { + this.stopCalls += 1; } - releaseRetry(): void { - this.#retry.resolve(); + 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'; } } diff --git a/packages/cli/src/pi-transcript.ts b/packages/cli/src/pi-transcript.ts index 87433d8e0e..4842ed9f5d 100644 --- a/packages/cli/src/pi-transcript.ts +++ b/packages/cli/src/pi-transcript.ts @@ -105,13 +105,13 @@ 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. + * 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. */ - pendingFallback: Array<{ text: string; enqueue: 'steer' | 'queue' }>; + pendingAdmission: Array<{ text: string; enqueue: 'steer' | 'queue' }>; /** Current non-durable provider retry progress for the activity strip. */ providerRetry?: ProviderRetryEvent; } @@ -216,7 +216,7 @@ export function createMakaPiTranscriptState(): MakaPiTranscriptState { usage: { costUsd: 0, cacheHitInput: 0, cacheMissInput: 0 }, steering: [], followup: [], - pendingFallback: [], + pendingAdmission: [], }; } @@ -343,7 +343,7 @@ export function replaceTranscriptWithStoredMessages( // Queues are per-active-run; a switched/reset session has none pending. state.steering = []; state.followup = []; - state.pendingFallback = []; + state.pendingAdmission = []; for (const msg of messages) { if (msg.type === 'token_usage') accumulateUsage(state.usage, msg); } @@ -1451,20 +1451,20 @@ export function renderMakaPiPendingQueue(state: MakaPiTranscriptState, width: nu if ( state.steering.length === 0 && state.followup.length === 0 && - state.pendingFallback.length === 0 + state.pendingAdmission.length === 0 ) { return []; } const safeWidth = Math.max(1, width); const steering = [ ...state.steering, - ...state.pendingFallback + ...state.pendingAdmission .filter((entry) => entry.enqueue === 'steer') .map((entry) => entry.text), ]; const followup = [ ...state.followup, - ...state.pendingFallback + ...state.pendingAdmission .filter((entry) => entry.enqueue === 'queue') .map((entry) => entry.text), ]; diff --git a/packages/cli/src/pi-tui-runner.ts b/packages/cli/src/pi-tui-runner.ts index 79c67bd6ed..e2b298a073 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,8 @@ 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')); + const held = pendingAdmissionText(); + refillEditorFromQueues([held, retracted].filter(Boolean).join('\n\n')); requestRender(); await input.driver.stop(); })().catch((error) => { @@ -867,8 +866,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 +890,31 @@ 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. + // 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 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 = []; + const takePendingAdmission = (): Array<{ text: string; enqueue: 'steer' | 'queue' }> => { + const entries = state.pendingAdmission; + state.pendingAdmission = []; 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'); + const pendingAdmissionText = (): string => + takePendingAdmission() + .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(); @@ -997,18 +923,22 @@ export async function runMakaPiTui(input: MakaPiTuiInput): Promise { editor.addToHistory(text); const enqueue = input.driver.steer; if (!enqueue) { - deferFallback(text, 'steer'); + // 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) => { if (outcome.kind === 'fallback') { - if (turnRunning || busy) deferFallback(text, 'steer'); + if (turnRunning || busy) holdForAdmission(text, 'steer'); else submitPrompt(text); return; } - // Queued: the runtime's `queue_update` event refreshes the mirror. + // The runtime's `queue_update` event refreshes the mirror. requestRender(); }) .catch((error) => { @@ -1038,18 +968,19 @@ export async function runMakaPiTui(input: MakaPiTuiInput): Promise { editor.addToHistory(text); const enqueue = input.driver.queueMessage; if (!enqueue) { - deferFallback(text, 'queue'); + // Same optional-method handoff as Enter above: hold instead of drop. + holdForAdmission(text, 'queue'); return; } const task = enqueue .call(input.driver, text) .then((outcome) => { if (outcome.kind === 'fallback') { - if (turnRunning || busy) deferFallback(text, 'queue'); + if (turnRunning || busy) holdForAdmission(text, 'queue'); else submitPrompt(text); return; } - // Queued: the runtime's `queue_update` event refreshes the mirror. + // The runtime's `queue_update` event refreshes the mirror. requestRender(); }) .catch((error) => { @@ -1059,14 +990,14 @@ 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')); + const held = pendingAdmissionText(); + refillEditorFromQueues([held, retracted].filter(Boolean).join('\n\n')); requestRender(); })().catch(reportError); }; @@ -1322,9 +1253,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 +1266,46 @@ 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') { + // 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); } - } 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); } + if (undelivered.length > 0) refillEditorFromQueues(undelivered.join('\n\n')); + requestRender(); } + } + if (outcome.kind === 'completed' && pendingAttachedTurn) { + const attached = pendingAttachedTurn; + pendingAttachedTurn = undefined; 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; 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,