feat(pi): add active-turn steering via native steer RPC (#3752)

* feat(pi): add active-turn steering via native steer RPC * fix(pi): clear queued steering before interrupt --------- Co-authored-by: paseo-bot[bot] <266920839+paseo-bot[bot]@users.noreply.github.com>
2026-11-10 10:25:03 +00:00 · 2026-08-31 21:10:27 +02:00
parent 22463bb7e5
commit ecec33265e
8 changed files with 410 additions and 3 deletions
@@ -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-"));
@@ -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<string, boolean>();
+  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<SteerResult> {
+    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<void> {
+    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) {
@@ -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[] = [];
@@ -121,6 +121,21 @@ class PiCliRuntimeSession implements PiRuntimeSession {
    return { requestId };
  }

+  async steer(
+    message: string,
+    images?: Array<{ type: "image"; data: string; mimeType: string }>,
+  ): Promise<void> {
+    await this.request({
+      type: "steer",
+      message,
+      ...(images?.length ? { images } : {}),
+    });
+  }
+
+  async clearQueue(): Promise<void> {
+    await this.request({ type: "clear_queue" });
+  }
+
  async compact(customInstructions?: string): Promise<void> {
    await this.waitForCompletion({
      type: "compact",
@@ -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" }
@@ -45,6 +45,11 @@ export interface PiRuntimeSession {
    message: string,
    images?: Array<{ type: "image"; data: string; mimeType: string }>,
  ): Promise<PiPromptAck>;
+  steer(
+    message: string,
+    images?: Array<{ type: "image"; data: string; mimeType: string }>,
+  ): Promise<void>;
+  clearQueue(): Promise<void>;
  compact(customInstructions?: string): Promise<void>;
  setAutoCompaction(enabled: boolean): Promise<void>;
  abort(): Promise<void>;
@@ -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<void>((resolve) => setImmediate(resolve));
  }

+  async steer(
+    message: string,
+    images?: Array<{ type: "image"; data: string; mimeType: string }>,
+  ): Promise<void> {
+    this.steerCalls.push({ message, imageCount: images?.length ?? 0 });
+    if (this.steerError) {
+      throw this.steerError;
+    }
+  }
+
+  async clearQueue(): Promise<void> {
+    this.controlRequests.push("clear_queue");
+    if (this.clearQueueError) {
+      throw this.clearQueueError;
+    }
+  }
+
  async compact(customInstructions?: string): Promise<void> {
    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<void> {
+    this.controlRequests.push("abort");
    this.abortRequested = true;
  }