diff --git a/docs/providers.md b/docs/providers.md index 8e98fde809..fc3ea055f3 100644 --- a/docs/providers.md +++ b/docs/providers.md @@ -85,9 +85,9 @@ OpenCode owns user message IDs. Do not pass Paseo-generated IDs to OpenCode prom `AgentManager` owns the one canonical timeline row for a foreground prompt carrying a Paseo `clientMessageId`. It records that row when `startTurn` accepts, with the wire `messageId` set to the same value. Provider adapters still emit their native user-message echo with the same `clientMessageId` when available; the manager records its provider identity on the internal row without changing or redispatching the wire item. If an adapter emits the echo before `startTurn` resolves, the manager records the provider identity with the row at acceptance. Provider adapters continue to own externally initiated user rows that have no Paseo client identity. Do not perform global transcript text dedupe. -Active-turn steering is an optional `AgentSession.steerActiveTurn` operation. The manager owns admission against its exact foreground turn, canonical user-message creation, echo reconciliation, and falls back to the normal interrupt-and-replace path only when the adapter reports `unavailable`. An adapter error leaves the steer's fate ambiguous and must surface without an interrupt or retry. Codex calls `turn/steer` with the native expected turn and Paseo client user-message ID. Claude pushes an admitted steer into the exact active SDK query input; isolated control commands remain unavailable. OpenCode calls `session/prompt_async` with an OpenCode-generated message ID; the server queues the prompt while busy and the next LLM call in the same Paseo turn includes it. A missing session reports `unavailable` and uses the normal interrupt fallback. +Active-turn steering is an optional `AgentSession.steerActiveTurn` operation. The manager owns admission against its exact foreground turn, canonical user-message creation, echo reconciliation, and falls back to the normal interrupt-and-replace path only when the adapter reports `unavailable`. An adapter error leaves the steer's fate ambiguous and must surface without an interrupt or retry. Codex calls `turn/steer` with the native expected turn and Paseo client user-message ID. Claude pushes an admitted steer into the exact active SDK query input; isolated control commands remain unavailable. OpenCode calls `session/prompt_async` with an OpenCode-generated message ID; the server queues the prompt while busy and the next LLM call in the same Paseo turn includes it. Pi sends its native `steer` RPC, which queues the message for delivery after the in-flight assistant turn's tool calls. Slash-command inputs report `unavailable` because pi rejects extension commands on the steer path, and echo identity is correlated by message text because pi's steer RPC takes no message ID. A missing session reports `unavailable` and uses the normal interrupt fallback. -A steering adapter also owes its interrupt: stopping a turn must discard the steers the provider has not read yet, or one of them resumes the turn the user just stopped. Codex clears pending input when it aborts a turn; Claude does not, so its adapter cancels the SDK messages it queued before calling `query.interrupt()`. +A steering adapter also owes its interrupt: stopping a turn must discard the steers the provider has not read yet, or one of them resumes the turn the user just stopped. Codex clears pending input when it aborts a turn; Claude does not, so its adapter cancels the SDK messages it queued before calling `query.interrupt()`. Pi requires `clear_queue` before `abort`; older binaries without that RPC retain their native queue behavior until the pi compatibility floor reaches 0.84.4. `SteerActiveTurnOptions.clearPendingPermissions` makes permission release part of the provider contract. A provider that accepts such a steer queues it first, denies permissions blocking its delivery, and stops once the steer is read. Steers without the flag leave permissions open. A denied plan remains in the timeline because the pending card was the only other copy of its text. diff --git a/packages/server/src/server/agent/providers/pi/agent.test.ts b/packages/server/src/server/agent/providers/pi/agent.test.ts index a6e1f30316..da2e785ca5 100644 --- a/packages/server/src/server/agent/providers/pi/agent.test.ts +++ b/packages/server/src/server/agent/providers/pi/agent.test.ts @@ -942,6 +942,27 @@ describe("PiRpcAgentSession", () => { ).not.toContain("turn_failed"); }); + test("clears queued Pi messages before interrupting", async () => { + const { pi, session } = await createSession(); + const fakeSession = pi.latestSession(); + + await session.startTurn("work"); + await session.interrupt(); + + expect(fakeSession.controlRequests).toEqual(["clear_queue", "abort"]); + }); + + test("still interrupts when an older Pi binary lacks clear_queue", async () => { + const { pi, session } = await createSession(); + const fakeSession = pi.latestSession(); + fakeSession.clearQueueError = new Error("Unknown command: clear_queue"); + + await session.startTurn("work"); + await session.interrupt(); + + expect(fakeSession.controlRequests).toEqual(["clear_queue", "abort"]); + }); + test("adds Pi assistant context to generic provider finish errors", async () => { const { pi, session, events } = await createSession(); @@ -1641,6 +1662,213 @@ describe("PiRpcAgentSession", () => { }); }); +describe("PiRpcAgentSession steering", () => { + test("steers the active Pi turn and correlates the echoed user message", async () => { + const { pi, session, events } = await createSession(); + const fakeSession = pi.latestSession(); + + const { turnId } = await session.startTurn("fix the tests", { + clientMessageId: "client-prompt-1", + }); + const result = await session.steerActiveTurn("steer this turn", { + expectedTurnId: turnId, + clientMessageId: "client-steer-1", + }); + + expect(result).toEqual({ status: "accepted" }); + expect(fakeSession.steerCalls).toEqual([{ message: "steer this turn", imageCount: 0 }]); + + fakeSession.finishSubmittedUserMessage({ + id: "entry-steer-1", + parentId: null, + text: "steer this turn", + }); + + const userMessages = events.timelineItems().filter((item) => item.type === "user_message"); + expect(userMessages).toEqual([ + { + type: "user_message", + text: "steer this turn", + messageId: "entry-steer-1", + clientMessageId: "client-steer-1", + }, + ]); + }); + + test("does not reuse the foreground client ID for a steer without one", async () => { + const { pi, session, events } = await createSession(); + const fakeSession = pi.latestSession(); + const { turnId } = await session.startTurn("work", { clientMessageId: "client-prompt-1" }); + + await session.steerActiveTurn("steer without a client ID", { expectedTurnId: turnId }); + fakeSession.finishSubmittedUserMessage({ + id: "entry-steer-1", + parentId: null, + text: "steer without a client ID", + }); + + expect(events.timelineItems()).toContainEqual({ + type: "user_message", + text: "steer without a client ID", + messageId: "entry-steer-1", + }); + }); + + test("keeps multiple steers correlated in admission order", async () => { + const { pi, session, events } = await createSession(); + const fakeSession = pi.latestSession(); + const { turnId } = await session.startTurn("work"); + + await session.steerActiveTurn("steer one", { + expectedTurnId: turnId, + clientMessageId: "client-steer-1", + }); + await session.steerActiveTurn("steer two", { + expectedTurnId: turnId, + clientMessageId: "client-steer-2", + }); + + fakeSession.finishSubmittedUserMessage({ id: "entry-1", parentId: null, text: "steer one" }); + fakeSession.finishSubmittedUserMessage({ id: "entry-2", parentId: null, text: "steer two" }); + + const userMessages = events.timelineItems().filter((item) => item.type === "user_message"); + expect(userMessages).toEqual([ + { + type: "user_message", + text: "steer one", + messageId: "entry-1", + clientMessageId: "client-steer-1", + }, + { + type: "user_message", + text: "steer two", + messageId: "entry-2", + clientMessageId: "client-steer-2", + }, + ]); + }); + + test("reports unavailable when the expected turn is not the active one", async () => { + const { pi, session } = await createSession(); + const fakeSession = pi.latestSession(); + const { turnId } = await session.startTurn("run"); + + const stale = await session.steerActiveTurn("steer", { + expectedTurnId: "turn-that-ended", + }); + expect(stale).toEqual({ status: "unavailable" }); + + fakeSession.finishTurn(); + await flushTurnScheduling(); + const idle = await session.steerActiveTurn("steer", { expectedTurnId: turnId }); + expect(idle).toEqual({ status: "unavailable" }); + expect(fakeSession.steerCalls).toEqual([]); + }); + + test("keeps slash-command steers on the interrupt fallback", async () => { + const { pi, session } = await createSession(); + const fakeSession = pi.latestSession(); + const { turnId } = await session.startTurn("run"); + + const result = await session.steerActiveTurn("/model", { expectedTurnId: turnId }); + + expect(result).toEqual({ status: "unavailable" }); + expect(fakeSession.steerCalls).toEqual([]); + }); + + test("falls back when the Pi binary lacks the steer RPC", async () => { + const { pi, session } = await createSession(); + const fakeSession = pi.latestSession(); + const { turnId } = await session.startTurn("run"); + fakeSession.steerError = new Error("Unknown command: steer"); + + const result = await session.steerActiveTurn("steer", { expectedTurnId: turnId }); + + expect(result).toEqual({ status: "unavailable" }); + expect(fakeSession.steerCalls).toEqual([{ message: "steer", imageCount: 0 }]); + }); + + test("surfaces an ambiguous steer failure without interrupting", async () => { + const { pi, session } = await createSession(); + const fakeSession = pi.latestSession(); + const { turnId } = await session.startTurn("run"); + fakeSession.steerError = new Error("Pi RPC socket closed"); + + await expect(session.steerActiveTurn("steer", { expectedTurnId: turnId })).rejects.toThrow( + "Pi RPC socket closed", + ); + }); + + test("denies pending permissions that block an accepted steer", async () => { + const { pi, session, events } = await createSession(); + const fakeSession = pi.latestSession(); + const { turnId } = await session.startTurn("work"); + fakeSession.emit({ + type: "extension_ui_request", + id: "perm-1", + method: "confirm", + title: "Allow command?", + }); + await events.nextPermissionRequest(); + expect(session.getPendingPermissions()).toHaveLength(1); + + const result = await session.steerActiveTurn("answer instead", { + expectedTurnId: turnId, + clearPendingPermissions: true, + }); + + expect(result).toEqual({ status: "accepted" }); + expect(fakeSession.extensionUiResponses).toEqual([ + { id: "perm-1", response: { cancelled: true } }, + ]); + expect(session.getPendingPermissions()).toEqual([]); + }); + + test("leaves permissions open for a steer without the clearing flag", async () => { + const { pi, session, events } = await createSession(); + const fakeSession = pi.latestSession(); + const { turnId } = await session.startTurn("work"); + fakeSession.emit({ + type: "extension_ui_request", + id: "perm-1", + method: "confirm", + title: "Allow command?", + }); + await events.nextPermissionRequest(); + + const result = await session.steerActiveTurn("answer instead", { expectedTurnId: turnId }); + + expect(result).toEqual({ status: "accepted" }); + expect(fakeSession.extensionUiResponses).toEqual([]); + expect(session.getPendingPermissions()).toHaveLength(1); + }); + + test("drops pending steer correlation once the turn completes", async () => { + const { pi, session, events } = await createSession(); + const fakeSession = pi.latestSession(); + const first = await session.startTurn("work"); + await session.steerActiveTurn("stale steer", { + expectedTurnId: first.turnId, + clientMessageId: "client-steer-1", + }); + fakeSession.finishTurn(); + await flushTurnScheduling(); + + const second = await session.startTurn("next"); + await session.steerActiveTurn("fresh steer", { expectedTurnId: second.turnId }); + fakeSession.finishSubmittedUserMessage({ + id: "entry-late", + parentId: null, + text: "stale steer", + }); + + const userMessages = events.timelineItems().filter((item) => item.type === "user_message"); + expect(userMessages).toEqual([ + { type: "user_message", text: "stale steer", messageId: "entry-late" }, + ]); + }); +}); + describe("PiRpcAgentClient", () => { test("lists JSONL persisted sessions from configured provider params", async () => { const root = mkdtempSync(path.join(tmpdir(), "paseo-pi-sessions-")); diff --git a/packages/server/src/server/agent/providers/pi/agent.ts b/packages/server/src/server/agent/providers/pi/agent.ts index 24e8f02d06..dc5c84375e 100644 --- a/packages/server/src/server/agent/providers/pi/agent.ts +++ b/packages/server/src/server/agent/providers/pi/agent.ts @@ -30,6 +30,8 @@ import { type AgentSlashCommandKind, type AgentStreamEvent, type FetchCatalogOptions, + type SteerActiveTurnOptions, + type SteerResult, type ImportableProviderSession, type ImportProviderSessionContext, type ImportProviderSessionInput, @@ -218,6 +220,21 @@ interface StartTurnResult { turnId: string; } +interface PiPendingSteerSubmission { + text: string; + clientMessageId: string | null; +} + +// COMPAT(piSteerFallback): added in v0.5.0, remove after 2027-03-01 once the pi floor +// supports the steer RPC. Older binaries answer with "Unknown command: steer". +function isPiDefinitiveSteerRejection(error: unknown): boolean { + return toDiagnosticErrorMessage(error) === "Unknown command: steer"; +} + +function isPiMissingClearQueueRpc(error: unknown): boolean { + return toDiagnosticErrorMessage(error) === "Unknown command: clear_queue"; +} + interface PiRpcAgentSessionOptions { runtimeSession: PiRuntimeSession; config: AgentSessionConfig; @@ -1219,6 +1236,7 @@ export class PiRpcAgentSession implements AgentSession { private readonly pendingNoTurnOutputs: Array<{ turnId: string; message: string }> = []; private activePromptRequestId: string | null = null; private readonly pendingPromptResults = new Map(); + private readonly pendingSteerSubmissions: PiPendingSteerSubmission[] = []; private lastKnownThinkingOptionId: string | null; currentLeafOverrideId: string | null | undefined; private readonly capturedUserEntries: PiCapturedEntry[] = []; @@ -1311,6 +1329,7 @@ export class PiRpcAgentSession implements AgentSession { this.activeTurnStartedEmitted = false; this.pendingSettledMessages = null; this.activePromptRequestId = null; + this.pendingSteerSubmissions.length = 0; this.clearNoTurnBuffers(); this.activeNoTurnPromptText = payload.text; const shouldProbeForNoTurnPrompt = this.parseSlashCommandInput(payload.text) !== null; @@ -1344,6 +1363,7 @@ export class PiRpcAgentSession implements AgentSession { this.activeTurnStartedEmitted = false; this.pendingSettledMessages = null; this.activeAssistantMessageId = null; + this.pendingSteerSubmissions.length = 0; this.clearNoTurnBuffers(); if (isPiRequestAbortError(error)) { this.emit({ @@ -1366,6 +1386,63 @@ export class PiRpcAgentSession implements AgentSession { return { turnId }; } + async steerActiveTurn( + prompt: AgentPromptInput, + options: SteerActiveTurnOptions, + ): Promise { + if (this.closed || this.activeTurnId !== options.expectedTurnId) { + return { status: "unavailable" }; + } + const payload = convertPromptInput(prompt, { model: this.state.model }); + // Pi rejects steer RPCs that are extension commands, so slash inputs keep the + // interrupt-and-replace fallback where they can run directly. + if (this.parseSlashCommandInput(payload.text)) { + return { status: "unavailable" }; + } + + try { + await this.runtimeSession.steer(payload.text, payload.images); + } catch (error) { + if (isPiDefinitiveSteerRejection(error)) { + return { status: "unavailable" }; + } + throw error; + } + // The steer is already queued inside pi; if the turn moved on meanwhile its fate + // is ambiguous, so surface it instead of replacing the wrong turn. + if (this.closed || this.activeTurnId !== options.expectedTurnId) { + return { status: "unavailable" }; + } + this.pendingSteerSubmissions.push({ + text: payload.text, + clientMessageId: options.clientMessageId ?? null, + }); + if (options.clearPendingPermissions) { + await this.clearPendingPermissionsForSteer(); + } + return { status: "accepted" }; + } + + private async clearPendingPermissionsForSteer(): Promise { + const requestIds = Array.from(this.pendingExtensionUiRequests.keys()); + for (const requestId of requestIds) { + if (!this.pendingExtensionUiRequests.has(requestId)) continue; + await this.respondToPermission(requestId, { + behavior: "deny", + message: "The user answered with a message instead of approving. Their message follows.", + }); + } + } + + private takePendingSteerSubmission(text: string): PiPendingSteerSubmission | undefined { + const index = this.pendingSteerSubmissions.findIndex((submission) => submission.text === text); + if (index < 0) { + return undefined; + } + const [submission] = this.pendingSteerSubmissions.splice(index, 1); + return submission; + } + subscribe(callback: (event: AgentStreamEvent) => void): () => void { this.subscribers.add(callback); return () => { @@ -1459,6 +1536,15 @@ export class PiRpcAgentSession implements AgentSession { this.lastInterruptedTurnId = turnId; } try { + try { + await this.runtimeSession.clearQueue(); + } catch (error) { + // COMPAT(piClearQueueFallback): added in v0.5.0, remove after 2027-03-01 once + // the pi floor supports clear_queue (added in pi 0.84.4). + if (!isPiMissingClearQueueRpc(error)) { + throw error; + } + } await this.runtimeSession.abort(); } catch (error) { if (this.interruptingTurnId === turnId) { @@ -1474,6 +1560,7 @@ export class PiRpcAgentSession implements AgentSession { this.activeTurnStartedEmitted = false; this.pendingSettledMessages = null; this.activeAssistantMessageId = null; + this.pendingSteerSubmissions.length = 0; this.clearNoTurnBuffers(); this.emit({ type: "turn_failed", @@ -1492,6 +1579,7 @@ export class PiRpcAgentSession implements AgentSession { this.activeTurnStartedEmitted = false; this.pendingSettledMessages = null; this.activeAssistantMessageId = null; + this.pendingSteerSubmissions.length = 0; this.clearNoTurnBuffers(); this.emit({ type: "turn_canceled", @@ -1885,6 +1973,10 @@ export class PiRpcAgentSession implements AgentSession { if (!entry) { return true; } + const pendingSteer = this.takePendingSteerSubmission(entry.text); + const clientMessageId = pendingSteer + ? pendingSteer.clientMessageId + : this.activeClientMessageId; this.emit({ type: "timeline", provider: this.provider, @@ -1893,7 +1985,7 @@ export class PiRpcAgentSession implements AgentSession { type: "user_message", text: entry.text, messageId: entry.id, - ...(this.activeClientMessageId ? { clientMessageId: this.activeClientMessageId } : {}), + ...(clientMessageId ? { clientMessageId } : {}), }, }); return true; @@ -2068,6 +2160,7 @@ export class PiRpcAgentSession implements AgentSession { this.activeTurnStarted = false; this.activeTurnStartedEmitted = false; this.pendingSettledMessages = null; + this.pendingSteerSubmissions.length = 0; this.clearNoTurnBuffers(); this.emit({ type: "turn_failed", @@ -2368,6 +2461,7 @@ export class PiRpcAgentSession implements AgentSession { this.activeTurnStarted = false; this.activeTurnStartedEmitted = false; this.pendingSettledMessages = null; + this.pendingSteerSubmissions.length = 0; this.clearNoTurnBuffers(); const errorMessage = latestPiErrorMessage(messages); if (typeof errorMessage === "string" && errorMessage.length > 0) { diff --git a/packages/server/src/server/agent/providers/pi/cli-runtime.test.ts b/packages/server/src/server/agent/providers/pi/cli-runtime.test.ts index 9e2d300d95..94517bde0d 100644 --- a/packages/server/src/server/agent/providers/pi/cli-runtime.test.ts +++ b/packages/server/src/server/agent/providers/pi/cli-runtime.test.ts @@ -368,6 +368,47 @@ describe("PiCliRuntime", () => { expect(child.killedSignals).toContain("SIGTERM"); }); + test("sends the steer RPC frame", async () => { + const child = createPiChild(); + replyToCommands(child, () => ({})); + const session = await createRuntime(child).startSession({ cwd: "/workspace/project" }); + + const steerCommand = capturePendingCommand(child, "steer"); + await session.steer("focus on error handling", [ + { type: "image", data: "aW1hZ2U=", mimeType: "image/png" }, + ]); + + expect(await steerCommand).toMatchObject({ + type: "steer", + message: "focus on error handling", + images: [{ type: "image", data: "aW1hZ2U=", mimeType: "image/png" }], + }); + }); + + test("sends the steer RPC frame without images", async () => { + const child = createPiChild(); + replyToCommands(child, () => ({})); + const session = await createRuntime(child).startSession({ cwd: "/workspace/project" }); + + const steerCommand = capturePendingCommand(child, "steer"); + await session.steer("focus on error handling"); + + const command = await steerCommand; + expect(command).toMatchObject({ type: "steer", message: "focus on error handling" }); + expect(command.images).toBeUndefined(); + }); + + test("sends the clear_queue RPC frame", async () => { + const child = createPiChild(); + replyToCommands(child, () => ({})); + const session = await createRuntime(child).startSession({ cwd: "/workspace/project" }); + + const clearQueueCommand = capturePendingCommand(child, "clear_queue"); + await session.clearQueue(); + + expect(await clearQueueCommand).toMatchObject({ type: "clear_queue" }); + }); + test("falls back to get_state when get_session_stats is unsupported", async () => { const child = createPiChild(); let commandSequence: string[] = []; diff --git a/packages/server/src/server/agent/providers/pi/cli-runtime.ts b/packages/server/src/server/agent/providers/pi/cli-runtime.ts index c0e7f28f6f..783be30249 100644 --- a/packages/server/src/server/agent/providers/pi/cli-runtime.ts +++ b/packages/server/src/server/agent/providers/pi/cli-runtime.ts @@ -121,6 +121,21 @@ class PiCliRuntimeSession implements PiRuntimeSession { return { requestId }; } + async steer( + message: string, + images?: Array<{ type: "image"; data: string; mimeType: string }>, + ): Promise { + await this.request({ + type: "steer", + message, + ...(images?.length ? { images } : {}), + }); + } + + async clearQueue(): Promise { + await this.request({ type: "clear_queue" }); + } + async compact(customInstructions?: string): Promise { await this.waitForCompletion({ type: "compact", diff --git a/packages/server/src/server/agent/providers/pi/rpc-types.ts b/packages/server/src/server/agent/providers/pi/rpc-types.ts index ce3241a19c..f84e0ff69e 100644 --- a/packages/server/src/server/agent/providers/pi/rpc-types.ts +++ b/packages/server/src/server/agent/providers/pi/rpc-types.ts @@ -128,6 +128,8 @@ export interface PiRpcSlashCommand { export type PiRpcCommand = | { id?: string; type: "prompt"; message: string; images?: PiImageContent[] } + | { id?: string; type: "steer"; message: string; images?: PiImageContent[] } + | { id?: string; type: "clear_queue" } | { id?: string; type: "compact"; customInstructions?: string } | { id?: string; type: "set_auto_compaction"; enabled: boolean } | { id?: string; type: "abort" } diff --git a/packages/server/src/server/agent/providers/pi/runtime.ts b/packages/server/src/server/agent/providers/pi/runtime.ts index 7d29fc159f..3582a902bd 100644 --- a/packages/server/src/server/agent/providers/pi/runtime.ts +++ b/packages/server/src/server/agent/providers/pi/runtime.ts @@ -45,6 +45,11 @@ export interface PiRuntimeSession { message: string, images?: Array<{ type: "image"; data: string; mimeType: string }>, ): Promise; + steer( + message: string, + images?: Array<{ type: "image"; data: string; mimeType: string }>, + ): Promise; + clearQueue(): Promise; compact(customInstructions?: string): Promise; setAutoCompaction(enabled: boolean): Promise; abort(): Promise; diff --git a/packages/server/src/server/agent/providers/pi/test-utils/fake-pi.ts b/packages/server/src/server/agent/providers/pi/test-utils/fake-pi.ts index 0d23486e81..b1d369eb85 100644 --- a/packages/server/src/server/agent/providers/pi/test-utils/fake-pi.ts +++ b/packages/server/src/server/agent/providers/pi/test-utils/fake-pi.ts @@ -94,6 +94,10 @@ export class FakePi implements PiRuntime { export class FakePiSession implements PiRuntimeSession { readonly prompts: Array<{ message: string; imageCount: number }> = []; + readonly steerCalls: Array<{ message: string; imageCount: number }> = []; + steerError: Error | null = null; + readonly controlRequests: string[] = []; + clearQueueError: Error | null = null; readonly compactRequests: Array<{ customInstructions?: string }> = []; readonly setAutoCompactionRequests: boolean[] = []; readonly subagentSubscriptionRequests: FakePiSubagentSubscriptionLevel[] = []; @@ -200,6 +204,23 @@ export class FakePiSession implements PiRuntimeSession { await new Promise((resolve) => setImmediate(resolve)); } + async steer( + message: string, + images?: Array<{ type: "image"; data: string; mimeType: string }>, + ): Promise { + this.steerCalls.push({ message, imageCount: images?.length ?? 0 }); + if (this.steerError) { + throw this.steerError; + } + } + + async clearQueue(): Promise { + this.controlRequests.push("clear_queue"); + if (this.clearQueueError) { + throw this.clearQueueError; + } + } + async compact(customInstructions?: string): Promise { this.compactRequests.push(customInstructions === undefined ? {} : { customInstructions }); this.emit({ type: "compaction_start", reason: "manual" }); @@ -220,6 +241,7 @@ export class FakePiSession implements PiRuntimeSession { } async abort(): Promise { + this.controlRequests.push("abort"); this.abortRequested = true; }