Update tests for steering capability and send-path routing

Test drift surfaced by rebasing onto main:

- client SDK agent fixture now expects supportsSteering: false (the steering feature adds this capability flag, schema-defaulted).

- permission-response fake AgentManager gains startAgentRun, since the shared startAgentRun helper now routes through AgentManager.startAgentRun (the steer-vs-replace-vs-stream dispatch) instead of calling streamAgent/replaceAgentRun directly.

packages/app/src/agent-stream/view.tsx

@@ -423,6 +423,7 @@ const AgentStreamViewComponent = forwardRef<AgentStreamViewHandle, AgentStreamVi
             client={client}
             isFirstInGroup={layoutItem.isFirstInUserGroup}
             isLastInGroup={layoutItem.isLastInUserGroup}
+            deliveryHint={item.deliveryHint}
           />
         );
       },

packages/app/src/components/message.tsx

@@ -63,7 +63,11 @@ import Animated, {
 import Svg, { Defs, LinearGradient as SvgLinearGradient, Rect, Stop } from "react-native-svg";
 import { createMarkdownStyles } from "@/styles/markdown-styles";
 import { Fonts } from "@/constants/theme";
-import type { TodoEntry, UserMessageImageAttachment } from "@/types/stream";
+import type {
+  TodoEntry,
+  UserMessageDeliveryHint,
+  UserMessageImageAttachment,
+} from "@/types/stream";
 import type { AgentAttachment } from "@getpaseo/protocol/messages";
 import type { ToolCallDetail } from "@getpaseo/protocol/agent-types";
 import { buildToolCallPresentation } from "@/tool-calls/presentation";
@@ -124,6 +128,7 @@ interface UserMessageProps {
   isFirstInGroup?: boolean;
   isLastInGroup?: boolean;
   disableOuterSpacing?: boolean;
+  deliveryHint?: UserMessageDeliveryHint;
 }
 
 const MessageOuterSpacingContext = createContext(false);
@@ -434,6 +439,12 @@ const userMessageStylesheet = StyleSheet.create((theme) => ({
     color: theme.colors.foregroundMuted,
     fontSize: STREAM_METADATA_FONT_SIZE,
   },
+  deliveryHintLabel: {
+    alignSelf: "flex-end",
+    marginTop: theme.spacing[1],
+    color: theme.colors.foregroundMuted,
+    fontSize: theme.fontSize.xs,
+  },
 }));
 
 function UserMessageAttachmentThumbnail({ image }: { image: UserMessageImageAttachment }) {
@@ -475,6 +486,7 @@ export const UserMessage = memo(function UserMessage({
   isFirstInGroup = true,
   isLastInGroup = true,
   disableOuterSpacing,
+  deliveryHint,
 }: UserMessageProps) {
   const isCompact = useIsCompactFormFactor();
   const [isHovered, setIsHovered] = useState(false);
@@ -589,6 +601,14 @@ export const UserMessage = memo(function UserMessage({
             />
           </View>
         ) : null}
+        {deliveryHint === "steering" ? (
+          <Text
+            accessibilityLabel="Steering conversation"
+            style={userMessageStylesheet.deliveryHintLabel}
+          >
+            Steering conversation
+          </Text>
+        ) : null}
       </View>
     </View>
   );

packages/app/src/composer/actions.test.ts

@@ -417,6 +417,44 @@ describe("dispatchComposerAgentMessage", () => {
     expect(client.calls[0]?.options.images).toEqual([]);
   });
 
+  it("attaches a steering deliveryHint to the optimistic user_message when provided", async () => {
+    const client = createFakeSendClient();
+    const stream = createFakeStream();
+
+    await dispatchComposerAgentMessage({
+      client,
+      agentId: "agent",
+      text: "steer this",
+      attachments: [],
+      encodeImages: passthroughEncodeImages,
+      stream,
+      deliveryHint: "steering",
+    });
+
+    const tail = stream.tail.get("agent");
+    expect(tail).toHaveLength(1);
+    const userMessage = tail?.[0] as Extract<StreamItem, { kind: "user_message" }>;
+    expect(userMessage.deliveryHint).toBe("steering");
+  });
+
+  it("omits deliveryHint when none is supplied (no marker for non-steering sends)", async () => {
+    const client = createFakeSendClient();
+    const stream = createFakeStream();
+
+    await dispatchComposerAgentMessage({
+      client,
+      agentId: "agent",
+      text: "no steering",
+      attachments: [],
+      encodeImages: passthroughEncodeImages,
+      stream,
+    });
+
+    const tail = stream.tail.get("agent");
+    const userMessage = tail?.[0] as Extract<StreamItem, { kind: "user_message" }>;
+    expect(userMessage.deliveryHint).toBeUndefined();
+  });
+
   it("serializes browser_element workspace attachments as text attachments at the wire boundary", async () => {
     const client = createFakeSendClient();
     const stream = createFakeStream();

packages/app/src/composer/actions.ts

@@ -14,6 +14,7 @@ import {
   buildOptimisticUserMessage,
   generateMessageId,
   type StreamItem,
+  type UserMessageDeliveryHint,
   type UserMessageItem,
 } from "@/types/stream";
 import type { PickedImageAttachmentInput } from "@/hooks/image-attachment-picker";
@@ -130,6 +131,7 @@ export interface DispatchComposerAgentMessageInput {
     images: AttachmentMetadata[],
   ) => Promise<Array<{ data: string; mimeType: string }> | undefined>;
   stream: AgentStreamWriter;
+  deliveryHint?: UserMessageDeliveryHint;
 }
 
 export async function dispatchComposerAgentMessage(
@@ -143,6 +145,7 @@ export async function dispatchComposerAgentMessage(
     timestamp: new Date(),
     images: wirePayload.images,
     attachments: wirePayload.attachments,
+    deliveryHint: input.deliveryHint,
   });
   appendUserMessageToStream(input.agentId, userMessage, input.stream);
   const imagesData = await input.encodeImages(wirePayload.images);

packages/app/src/composer/index.tsx

@@ -1070,6 +1070,9 @@ export function Composer({
         setHead: (updater) => setAgentStreamHead(serverId, updater),
         setTail: (updater) => setAgentStreamTail(serverId, updater),
       };
+      const targetAgent = useSessionStore.getState().sessions[serverId]?.agents?.get(targetAgentId);
+      const isSteering =
+        targetAgent?.status === "running" && targetAgent.capabilities.supportsSteering === true;
       await dispatchComposerAgentMessage({
         client,
         agentId: targetAgentId,
@@ -1077,6 +1080,7 @@ export function Composer({
         attachments: sendAttachments,
         encodeImages,
         stream,
+        ...(isSteering ? { deliveryHint: "steering" as const } : {}),
       });
       onAttentionPromptSend?.();
     };

packages/app/src/contexts/session-context.tsx

@@ -1700,6 +1700,9 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
       attachments?: AgentAttachment[],
     ) => {
       const messageId = generateMessageId();
+      const targetAgent = useSessionStore.getState().sessions[serverId]?.agents?.get(agentId);
+      const isSteering =
+        targetAgent?.status === "running" && targetAgent.capabilities.supportsSteering === true;
       const userMessage: StreamItem = {
         kind: "user_message",
         id: messageId,
@@ -1708,6 +1711,7 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
         optimistic: true,
         ...(images && images.length > 0 ? { images } : {}),
         ...(attachments && attachments.length > 0 ? { attachments } : {}),
+        ...(isSteering ? { deliveryHint: "steering" as const } : {}),
       };
 
       // Append to head if streaming (keeps the user message with the current

packages/app/src/types/stream.test.ts

@@ -765,6 +765,52 @@ describe("stream reducer canonical tool calls", () => {
     assert.strictEqual(message.timestamp.getTime(), optimisticTimestamp.getTime());
   });
 
+  it("preserves optimistic steering deliveryHint when authoritative user message arrives", () => {
+    const messageId = "msg-user-steering";
+    const initialState: StreamItem[] = [
+      {
+        kind: "user_message",
+        id: messageId,
+        text: "Steer to a different approach",
+        timestamp: new Date("2025-01-01T11:10:00Z"),
+        deliveryHint: "steering",
+      },
+    ];
+    const event: AgentStreamEventPayload = {
+      type: "timeline",
+      provider: "claude",
+      item: {
+        type: "user_message",
+        text: "Steer to a different approach",
+        messageId,
+      },
+    };
+
+    const state = reduceStreamUpdate(initialState, event, new Date("2025-01-01T11:10:01Z"));
+    const message = state.find((item) => item.kind === "user_message");
+
+    assert.ok(message);
+    assert.strictEqual(message.deliveryHint, "steering");
+  });
+
+  it("does not invent a deliveryHint for canonical user messages without prior optimistic state", () => {
+    const event: AgentStreamEventPayload = {
+      type: "timeline",
+      provider: "claude",
+      item: {
+        type: "user_message",
+        text: "Hello there",
+        messageId: "canonical-only",
+      },
+    };
+
+    const state = reduceStreamUpdate([], event, new Date("2025-01-01T11:11:00Z"));
+    const message = state.find((item) => item.kind === "user_message");
+
+    assert.ok(message);
+    assert.strictEqual(message.deliveryHint, undefined);
+  });
+
   it("keeps canonical assistant/user/assistant order during replay", () => {
     const state: StreamItem[] = [
       {

packages/app/src/types/stream.ts

@@ -54,6 +54,8 @@ export type StreamItem =
 
 export type UserMessageImageAttachment = AttachmentMetadata;
 
+export type UserMessageDeliveryHint = "steering";
+
 export interface UserMessageItem {
   kind: "user_message";
   id: string;
@@ -62,6 +64,13 @@ export interface UserMessageItem {
   optimistic?: true;
   images?: UserMessageImageAttachment[];
   attachments?: AgentAttachment[];
+  /**
+   * Local-only marker set when this user message was sent optimistically
+   * while the agent was running and the provider supports steering. Never
+   * sent or received over the wire — preserved across canonical replays so
+   * the UI can keep showing the marker.
+   */
+  deliveryHint?: UserMessageDeliveryHint;
 }
 
 export interface OptimisticUserMessageInput {
@@ -70,6 +79,7 @@ export interface OptimisticUserMessageInput {
   timestamp: Date;
   images?: UserMessageImageAttachment[];
   attachments?: AgentAttachment[];
+  deliveryHint?: UserMessageDeliveryHint;
 }
 
 export type OptimisticUserMessagePlacement = "tail" | "active-head";
@@ -213,6 +223,7 @@ function buildUserMessageItem(input: {
       ...(input.optimistic.attachments && input.optimistic.attachments.length > 0
         ? { attachments: input.optimistic.attachments }
         : {}),
+      ...(input.optimistic.deliveryHint ? { deliveryHint: input.optimistic.deliveryHint } : {}),
     };
   }
 
@@ -235,6 +246,7 @@ export function buildOptimisticUserMessage(input: OptimisticUserMessageInput): U
     ...(input.attachments && input.attachments.length > 0
       ? { attachments: input.attachments }
       : {}),
+    ...(input.deliveryHint ? { deliveryHint: input.deliveryHint } : {}),
   };
 }
 

packages/client/src/index.test.ts

@@ -152,6 +152,7 @@ function createAgent(input: Partial<PaseoAgent> = {}): PaseoAgent {
       supportsRewindBoth: false,
       supportsRewindConversation: false,
       supportsRewindFiles: false,
+      supportsSteering: false,
       supportsToolInvocations: true,
     },
     currentModeId: null,

packages/protocol/src/agent-types.ts

@@ -125,6 +125,7 @@ export interface AgentCapabilityFlags {
   supportsRewindConversation?: boolean;
   supportsRewindFiles?: boolean;
   supportsRewindBoth?: boolean;
+  supportsSteering?: boolean;
 }
 
 export interface AgentPersistenceHandle {

packages/protocol/src/messages.providers-snapshot.test.ts

@@ -1,5 +1,6 @@
 import { describe, expect, test } from "vitest";
 import {
+  AgentSnapshotPayloadSchema,
   GetProvidersSnapshotResponseMessageSchema,
   ProviderSnapshotEntrySchema,
   ProvidersSnapshotUpdateMessageSchema,
@@ -16,6 +17,34 @@ describe("provider snapshot message schemas", () => {
     expect(parsed.enabled).toBe(true);
   });
 
+  test("defaults missing agent capability steering support to false", () => {
+    const parsed = AgentSnapshotPayloadSchema.parse({
+      id: "agent-1",
+      provider: "codex",
+      cwd: "/tmp/repo",
+      model: null,
+      createdAt: "2026-04-24T00:00:00.000Z",
+      updatedAt: "2026-04-24T00:00:00.000Z",
+      lastUserMessageAt: null,
+      status: "idle",
+      capabilities: {
+        supportsStreaming: true,
+        supportsSessionPersistence: true,
+        supportsDynamicModes: false,
+        supportsMcpServers: true,
+        supportsReasoningStream: true,
+        supportsToolInvocations: true,
+      },
+      currentModeId: null,
+      availableModes: [],
+      pendingPermissions: [],
+      persistence: null,
+      title: null,
+    });
+
+    expect(parsed.capabilities.supportsSteering).toBe(false);
+  });
+
   test("preserves disabled provider snapshot entries", () => {
     const parsed = ProviderSnapshotEntrySchema.parse({
       provider: "claude",

packages/protocol/src/messages.ts

@@ -251,6 +251,8 @@ const AgentCapabilityFlagsSchema: z.ZodType<AgentCapabilityFlags> = z.object({
   supportsRewindFiles: z.boolean().optional().default(false),
   // COMPAT(rewind): added in v0.1.X, drop when floor >= v0.1.X.
   supportsRewindBoth: z.boolean().optional().default(false),
+  // COMPAT(steering): added in v0.1.X, drop when floor >= v0.1.X.
+  supportsSteering: z.boolean().optional().default(false),
 });
 
 const AgentUsageSchema: z.ZodType<AgentUsage> = z.object({

packages/server/src/server/agent/agent-manager.test.ts

@@ -18,6 +18,7 @@ import type {
   AgentLaunchContext,
   AgentProvider,
   AgentPersistenceHandle,
+  AgentPromptInput,
   AgentRunResult,
   AgentSession,
   AgentSessionConfig,
@@ -3010,6 +3011,314 @@ test("replaceAgentRun does not emit idle or resolve waiters between interrupted
   unsubscribe();
 });
 
+test("startAgentRun applies the replace-vs-stream policy in AgentManager", async () => {
+  const workdir = mkdtempSync(join(tmpdir(), "agent-manager-start-policy-"));
+  const storagePath = join(workdir, "agents");
+  const storage = new AgentStorage(storagePath, logger);
+  const manager = new AgentManager({
+    clients: {
+      codex: new TestAgentClient(),
+    },
+    registry: storage,
+    logger,
+    idFactory: () => "00000000-0000-4000-8000-000000000126",
+  });
+
+  const snapshot = await manager.createAgent({
+    provider: "codex",
+    cwd: workdir,
+  });
+
+  const replacementEvents = (async function* replacementEvents() {})();
+  const streamEvents = (async function* streamEvents() {})();
+  const hasInFlightRunSpy = vi.spyOn(manager, "hasInFlightRun").mockReturnValue(true);
+  const replaceAgentRunSpy = vi
+    .spyOn(manager, "replaceAgentRun")
+    .mockReturnValue(replacementEvents);
+  const streamAgentSpy = vi.spyOn(manager, "streamAgent").mockReturnValue(streamEvents);
+
+  const replacement = manager.startAgentRun(snapshot.id, "replace me", {
+    replaceRunning: true,
+  });
+
+  expect(replacement).toEqual({ outOfBand: false, events: replacementEvents });
+  expect(hasInFlightRunSpy).toHaveBeenCalledWith(snapshot.id);
+  expect(replaceAgentRunSpy).toHaveBeenCalledWith(snapshot.id, "replace me", undefined);
+  expect(streamAgentSpy).not.toHaveBeenCalled();
+
+  replaceAgentRunSpy.mockClear();
+  streamAgentSpy.mockClear();
+
+  const streamed = manager.startAgentRun(snapshot.id, "stream me", {
+    replaceRunning: false,
+  });
+
+  expect(streamed).toEqual({ outOfBand: false, events: streamEvents });
+  expect(streamAgentSpy).toHaveBeenCalledWith(snapshot.id, "stream me", undefined);
+  expect(replaceAgentRunSpy).not.toHaveBeenCalled();
+});
+
+test("startAgentRun steers a running foreground turn when the provider supports steering", async () => {
+  const workdir = mkdtempSync(join(tmpdir(), "agent-manager-steer-run-"));
+  const storagePath = join(workdir, "agents");
+  const storage = new AgentStorage(storagePath, logger);
+  const allowFirstRunToEnd = deferred<void>();
+  const steeredPrompts: AgentPromptInput[] = [];
+  let capturedSession: SteerableSession | null = null;
+
+  const steeringCapabilities = {
+    ...TEST_CAPABILITIES,
+    supportsSteering: true,
+  } as const;
+
+  class SteerableSession implements AgentSession {
+    readonly provider = "codex" as const;
+    readonly capabilities = steeringCapabilities;
+    readonly id = randomUUID();
+    readonly interrupt = vi.fn(async () => undefined);
+    private subscribers = new Set<(event: AgentStreamEvent) => void>();
+    private turnIdCounter = 0;
+
+    async run(): Promise<AgentRunResult> {
+      return {
+        sessionId: this.id,
+        finalText: "",
+        timeline: [],
+      };
+    }
+
+    async startTurn(): Promise<{ turnId: string }> {
+      const turnId = `turn-${++this.turnIdCounter}`;
+      void (async () => {
+        this.pushEvent({ type: "turn_started", provider: this.provider, turnId });
+        await allowFirstRunToEnd.promise;
+        this.pushEvent({ type: "turn_completed", provider: this.provider, turnId });
+      })();
+      return { turnId };
+    }
+
+    async steerTurn(prompt: AgentPromptInput): Promise<void> {
+      steeredPrompts.push(prompt);
+    }
+
+    subscribe(callback: (event: AgentStreamEvent) => void): () => void {
+      this.subscribers.add(callback);
+      return () => {
+        this.subscribers.delete(callback);
+      };
+    }
+
+    private pushEvent(event: AgentStreamEvent): void {
+      for (const callback of this.subscribers) {
+        callback(event);
+      }
+    }
+
+    async *streamHistory(): AsyncGenerator<AgentStreamEvent> {}
+    async getRuntimeInfo() {
+      return { provider: this.provider, sessionId: this.id, model: null, modeId: null };
+    }
+    async getAvailableModes() {
+      return [];
+    }
+    async getCurrentMode() {
+      return null;
+    }
+    async setMode(): Promise<void> {}
+    getPendingPermissions() {
+      return [];
+    }
+    async respondToPermission(): Promise<void> {}
+    describePersistence() {
+      return { provider: this.provider, sessionId: this.id };
+    }
+    async close(): Promise<void> {}
+  }
+
+  class SteerableClient extends TestAgentClient {
+    readonly capabilities = steeringCapabilities;
+
+    override async createSession(_config: AgentSessionConfig): Promise<AgentSession> {
+      capturedSession = new SteerableSession();
+      return capturedSession;
+    }
+  }
+
+  const manager = new AgentManager({
+    clients: {
+      codex: new SteerableClient(),
+    },
+    registry: storage,
+    logger,
+    idFactory: () => "00000000-0000-4000-8000-000000000127",
+  });
+
+  const snapshot = await manager.createAgent({
+    provider: "codex",
+    cwd: workdir,
+  });
+
+  const firstRun = manager.streamAgent(snapshot.id, "first run");
+  const firstRunDrain = (async () => {
+    for await (const _event of firstRun) {
+      // Drain the original foreground run.
+    }
+  })();
+
+  await manager.waitForAgentRunStart(snapshot.id);
+
+  const steered = manager.startAgentRun(snapshot.id, "steer this", {
+    replaceRunning: true,
+  });
+  expect(steered.outOfBand).toBe(false);
+  if (!steered.outOfBand) {
+    for await (const _event of steered.events) {
+      // Steering produces no foreground stream events.
+    }
+  }
+
+  expect(steeredPrompts).toEqual(["steer this"]);
+  expect(capturedSession?.interrupt).not.toHaveBeenCalled();
+  expect(manager.getAgent(snapshot.id)?.lifecycle).toBe("running");
+
+  allowFirstRunToEnd.resolve();
+  await firstRunDrain;
+});
+
+test("startAgentRun replaces a running foreground turn when steering is unsupported", async () => {
+  const workdir = mkdtempSync(join(tmpdir(), "agent-manager-unsupported-steer-run-"));
+  const storagePath = join(workdir, "agents");
+  const storage = new AgentStorage(storagePath, logger);
+  const allowFirstRunToCancel = deferred<void>();
+  const allowSecondRunToEnd = deferred<void>();
+  let startTurnCount = 0;
+  let interruptCount = 0;
+
+  class UnsupportedSteeringSession implements AgentSession {
+    readonly provider = "codex" as const;
+    readonly capabilities = TEST_CAPABILITIES;
+    readonly id = randomUUID();
+    private subscribers = new Set<(event: AgentStreamEvent) => void>();
+
+    async run(): Promise<AgentRunResult> {
+      return {
+        sessionId: this.id,
+        finalText: "",
+        timeline: [],
+      };
+    }
+
+    async startTurn(): Promise<{ turnId: string }> {
+      startTurnCount += 1;
+      const turnId = `turn-${startTurnCount}`;
+      const turnNumber = startTurnCount;
+      void (async () => {
+        this.pushEvent({ type: "turn_started", provider: this.provider, turnId });
+        if (turnNumber === 1) {
+          await allowFirstRunToCancel.promise;
+          this.pushEvent({
+            type: "turn_canceled",
+            provider: this.provider,
+            reason: "interrupted",
+            turnId,
+          });
+          return;
+        }
+        await allowSecondRunToEnd.promise;
+        this.pushEvent({ type: "turn_completed", provider: this.provider, turnId });
+      })();
+      return { turnId };
+    }
+
+    subscribe(callback: (event: AgentStreamEvent) => void): () => void {
+      this.subscribers.add(callback);
+      return () => {
+        this.subscribers.delete(callback);
+      };
+    }
+
+    private pushEvent(event: AgentStreamEvent): void {
+      for (const callback of this.subscribers) {
+        callback(event);
+      }
+    }
+
+    async *streamHistory(): AsyncGenerator<AgentStreamEvent> {}
+    async getRuntimeInfo() {
+      return { provider: this.provider, sessionId: this.id, model: null, modeId: null };
+    }
+    async getAvailableModes() {
+      return [];
+    }
+    async getCurrentMode() {
+      return null;
+    }
+    async setMode(): Promise<void> {}
+    getPendingPermissions() {
+      return [];
+    }
+    async respondToPermission(): Promise<void> {}
+    describePersistence() {
+      return { provider: this.provider, sessionId: this.id };
+    }
+    async interrupt(): Promise<void> {
+      interruptCount += 1;
+      allowFirstRunToCancel.resolve();
+    }
+    async close(): Promise<void> {}
+  }
+
+  class UnsupportedSteeringClient extends TestAgentClient {
+    override async createSession(_config: AgentSessionConfig): Promise<AgentSession> {
+      return new UnsupportedSteeringSession();
+    }
+  }
+
+  const manager = new AgentManager({
+    clients: {
+      codex: new UnsupportedSteeringClient(),
+    },
+    registry: storage,
+    logger,
+    idFactory: () => "00000000-0000-4000-8000-000000000128",
+  });
+
+  const snapshot = await manager.createAgent({
+    provider: "codex",
+    cwd: workdir,
+  });
+
+  const firstRun = manager.streamAgent(snapshot.id, "first run");
+  const firstRunDrain = (async () => {
+    for await (const _event of firstRun) {
+      // Drain the original foreground run.
+    }
+  })();
+
+  await manager.waitForAgentRunStart(snapshot.id);
+
+  const replacement = manager.startAgentRun(snapshot.id, "replace this", {
+    replaceRunning: true,
+  });
+  expect(replacement.outOfBand).toBe(false);
+  const replacementDrain =
+    replacement.outOfBand === false
+      ? (async () => {
+          for await (const _event of replacement.events) {
+            // Drain the replacement foreground run.
+          }
+        })()
+      : Promise.resolve();
+
+  await manager.waitForAgentRunStart(snapshot.id);
+  expect(interruptCount).toBe(1);
+  expect(startTurnCount).toBe(2);
+
+  allowSecondRunToEnd.resolve();
+  await firstRunDrain;
+  await replacementDrain;
+});
+
 test("replaceAgentRun stays running when a stale old terminal arrives before the replacement turn is current", async () => {
   const workdir = mkdtempSync(join(tmpdir(), "agent-manager-replace-stale-terminal-"));
   const storagePath = join(workdir, "agents");

packages/server/src/server/agent/agent-manager.ts

@@ -205,6 +205,15 @@ export interface WaitForAgentStartOptions {
   signal?: AbortSignal;
 }
 
+export interface StartAgentRunOptions {
+  replaceRunning?: boolean;
+  runOptions?: AgentRunOptions;
+}
+
+export type StartAgentRunResult =
+  | { outOfBand: true }
+  | { outOfBand: false; events: AsyncGenerator<AgentStreamEvent> };
+
 type AttentionState =
   | { requiresAttention: false }
   | {
@@ -1482,6 +1491,55 @@ export class AgentManager {
     return true;
   }
 
+  startAgentRun(
+    agentId: string,
+    prompt: AgentPromptInput,
+    options?: StartAgentRunOptions,
+  ): StartAgentRunResult {
+    // Out-of-band commands (e.g. /goal pause) must run WITHOUT canceling an
+    // in-flight turn. Keeping this policy here makes every send surface share
+    // the same replace-vs-stream decision.
+    if (this.tryRunOutOfBand(agentId, prompt)) {
+      return { outOfBand: true };
+    }
+
+    const shouldReplace = Boolean(options?.replaceRunning && this.hasInFlightRun(agentId));
+    const runOptions = options?.runOptions;
+    if (shouldReplace && this.canSteerActiveForegroundRun(agentId)) {
+      const events = this.steerAgentRun(agentId, prompt, runOptions);
+      return { outOfBand: false, events };
+    }
+
+    const events = shouldReplace
+      ? this.replaceAgentRun(agentId, prompt, runOptions)
+      : this.streamAgent(agentId, prompt, runOptions);
+    return { outOfBand: false, events };
+  }
+
+  private canSteerActiveForegroundRun(agentId: string): boolean {
+    const agent = this.agents.get(agentId);
+    return Boolean(
+      agent?.activeForegroundTurnId &&
+      agent.capabilities.supportsSteering === true &&
+      agent.session.steerTurn,
+    );
+  }
+
+  private async *steerAgentRun(
+    agentId: string,
+    prompt: AgentPromptInput,
+    options?: AgentRunOptions,
+  ): AsyncGenerator<AgentStreamEvent> {
+    const agent = this.requireSessionAgent(agentId);
+    if (!agent.session.steerTurn) {
+      throw new Error(`Agent ${agentId} does not support steering`);
+    }
+    await agent.session.steerTurn(prompt, options);
+    this.touchUpdatedAt(agent);
+    this.emitState(agent);
+    yield* [];
+  }
+
   async appendTimelineItem(agentId: string, item: AgentTimelineItem): Promise<void> {
     const agent = this.requireAgent(agentId);
     this.touchUpdatedAt(agent);

packages/server/src/server/agent/agent-prompt.test.ts

@@ -10,6 +10,9 @@ import {
 } from "./agent-prompt.js";
 import type { AgentManagerEvent, ManagedAgent } from "./agent-manager.js";
 
+const CHILD_AGENT_ID = "11111111-1111-4111-8111-111111111111";
+const CALLER_AGENT_ID = "22222222-2222-4222-8222-222222222222";
+
 test("isSystemInjectedEnvelope matches the envelope formatSystemNotificationPrompt produces", () => {
   expect(isSystemInjectedEnvelope(formatSystemNotificationPrompt("child finished"))).toBe(true);
   expect(isSystemInjectedEnvelope("hello world")).toBe(false);
@@ -19,27 +22,24 @@ it("does not notify archived callers", async () => {
   let subscriber: ((event: AgentManagerEvent) => void) | null = null;
 
   const childAgent: ManagedAgent = Object.create(null);
-  Reflect.set(childAgent, "id", "child-agent");
+  Reflect.set(childAgent, "id", CHILD_AGENT_ID);
   Reflect.set(childAgent, "lifecycle", "idle");
   Reflect.set(childAgent, "config", { title: "Child Agent" });
 
   const callerAgent: ManagedAgent = Object.create(null);
-  Reflect.set(callerAgent, "id", "caller-agent");
+  Reflect.set(callerAgent, "id", CALLER_AGENT_ID);
   Reflect.set(callerAgent, "lifecycle", "idle");
   Reflect.set(callerAgent, "config", { title: "Caller Agent" });
 
-  const streamAgentSpy = vi.fn(() => (async function* noop() {})());
-  const replaceAgentRunSpy = vi.fn(() => (async function* noop() {})());
-
   const agentManager: AgentManager = Object.create(AgentManager.prototype);
   Reflect.set(
     agentManager,
     "getAgent",
     vi.fn((agentId: string) => {
-      if (agentId === "child-agent") {
+      if (agentId === CHILD_AGENT_ID) {
         return childAgent;
       }
-      if (agentId === "caller-agent") {
+      if (agentId === CALLER_AGENT_ID) {
         return callerAgent;
       }
       return null;
@@ -55,12 +55,14 @@ it("does not notify archived callers", async () => {
       };
     }),
   );
-  Reflect.set(agentManager, "hasInFlightRun", vi.fn().mockReturnValue(false));
-  Reflect.set(agentManager, "streamAgent", streamAgentSpy);
-  Reflect.set(agentManager, "replaceAgentRun", replaceAgentRunSpy);
+  const startAgentRunSpy = vi.fn(() => ({
+    outOfBand: false,
+    events: (async function* noop() {})(),
+  }));
+  Reflect.set(agentManager, "startAgentRun", startAgentRunSpy);
 
   const agentStorageGetSpy = vi.fn(async (agentId: string) =>
-    agentId === "caller-agent" ? { archivedAt: "2024-01-01" } : null,
+    agentId === CALLER_AGENT_ID ? { archivedAt: "2024-01-01" } : null,
   );
   const agentStorage: AgentStorage = Object.create(AgentStorage.prototype);
   Reflect.set(agentStorage, "get", agentStorageGetSpy);
@@ -68,8 +70,8 @@ it("does not notify archived callers", async () => {
   setupFinishNotification({
     agentManager,
     agentStorage,
-    childAgentId: "child-agent",
-    callerAgentId: "caller-agent",
+    childAgentId: CHILD_AGENT_ID,
+    callerAgentId: CALLER_AGENT_ID,
     logger: createTestLogger(),
   });
 
@@ -88,9 +90,92 @@ it("does not notify archived callers", async () => {
   });
 
   await vi.waitFor(() => {
-    expect(agentStorageGetSpy).toHaveBeenCalledWith("caller-agent");
+    expect(agentStorageGetSpy).toHaveBeenCalledWith(CALLER_AGENT_ID);
   });
 
-  expect(streamAgentSpy).not.toHaveBeenCalled();
-  expect(replaceAgentRunSpy).not.toHaveBeenCalled();
+  expect(startAgentRunSpy).not.toHaveBeenCalled();
+});
+
+it("uses AgentManager startAgentRun for finish notifications", async () => {
+  let subscriber: ((event: AgentManagerEvent) => void) | null = null;
+
+  const childAgent: ManagedAgent = Object.create(null);
+  Reflect.set(childAgent, "id", CHILD_AGENT_ID);
+  Reflect.set(childAgent, "lifecycle", "idle");
+  Reflect.set(childAgent, "config", { title: "Child Agent" });
+
+  const callerAgent: ManagedAgent = Object.create(null);
+  Reflect.set(callerAgent, "id", CALLER_AGENT_ID);
+  Reflect.set(callerAgent, "lifecycle", "idle");
+  Reflect.set(callerAgent, "config", { title: "Caller Agent" });
+
+  const startAgentRunSpy = vi.fn(() => ({
+    outOfBand: false,
+    events: (async function* noop() {})(),
+  }));
+
+  const agentManager: AgentManager = Object.create(AgentManager.prototype);
+  Reflect.set(
+    agentManager,
+    "getAgent",
+    vi.fn((agentId: string) => {
+      if (agentId === CHILD_AGENT_ID) {
+        return childAgent;
+      }
+      if (agentId === CALLER_AGENT_ID) {
+        return callerAgent;
+      }
+      return null;
+    }),
+  );
+  Reflect.set(
+    agentManager,
+    "subscribe",
+    vi.fn((callback: (event: AgentManagerEvent) => void) => {
+      subscriber = callback;
+      return () => {
+        subscriber = null;
+      };
+    }),
+  );
+  Reflect.set(agentManager, "startAgentRun", startAgentRunSpy);
+
+  const agentStorage: AgentStorage = Object.create(AgentStorage.prototype);
+  Reflect.set(
+    agentStorage,
+    "get",
+    vi.fn(async (agentId: string) =>
+      agentId === CHILD_AGENT_ID ? { title: "Child Agent" } : null,
+    ),
+  );
+
+  setupFinishNotification({
+    agentManager,
+    agentStorage,
+    childAgentId: CHILD_AGENT_ID,
+    callerAgentId: CALLER_AGENT_ID,
+    logger: createTestLogger(),
+  });
+
+  expect(subscriber).not.toBeNull();
+
+  childAgent.lifecycle = "running";
+  subscriber?.({
+    type: "agent_state",
+    agent: childAgent,
+  });
+
+  childAgent.lifecycle = "idle";
+  subscriber?.({
+    type: "agent_state",
+    agent: childAgent,
+  });
+
+  await vi.waitFor(() => {
+    expect(startAgentRunSpy).toHaveBeenCalledWith(
+      CALLER_AGENT_ID,
+      `<paseo-system>\nAgent ${CHILD_AGENT_ID} (Child Agent) finished.\n</paseo-system>`,
+      { replaceRunning: true },
+    );
+  });
 });

packages/server/src/server/agent/agent-prompt.ts

@@ -1,19 +1,11 @@
 import type { Logger } from "pino";
 
 import type { AgentPromptInput, AgentRunOptions } from "./agent-sdk-types.js";
-import type { AgentManager, ManagedAgent } from "./agent-manager.js";
+import type { AgentManager, ManagedAgent, StartAgentRunOptions } from "./agent-manager.js";
 import type { AgentStorage } from "./agent-storage.js";
 import { ensureAgentLoaded } from "./agent-loading.js";
 
-export type AgentRunController = Pick<
-  AgentManager,
-  "getAgent" | "tryRunOutOfBand" | "hasInFlightRun" | "replaceAgentRun" | "streamAgent"
->;
-
-export interface StartAgentRunOptions {
-  replaceRunning?: boolean;
-  runOptions?: AgentRunOptions;
-}
+export type AgentRunController = Pick<AgentManager, "getAgent" | "startAgentRun">;
 
 export function startAgentRun(
   agentManager: AgentRunController,
@@ -35,23 +27,17 @@ export function startAgentRun(
     },
     "agent.session.start_stream.request",
   );
-  // Out-of-band commands (e.g. /goal pause) must run WITHOUT canceling an
-  // in-flight turn — replaceAgentRun would interrupt the running turn. The
-  // intercept lives at this layer so it covers every prompt entrypoint.
-  if (agentManager.tryRunOutOfBand(agentId, prompt)) {
+
+  const result = agentManager.startAgentRun(agentId, prompt, options);
+  if (result.outOfBand) {
     return { outOfBand: true };
   }
-  const shouldReplace = Boolean(options?.replaceRunning && agentManager.hasInFlightRun(agentId));
-  const runOptions = options?.runOptions;
-  const iterator = shouldReplace
-    ? agentManager.replaceAgentRun(agentId, prompt, runOptions)
-    : agentManager.streamAgent(agentId, prompt, runOptions);
+  const iterator = result.events;
   logger.trace(
     {
       agentId,
       provider: snapshot?.provider,
       providerSessionId: snapshot?.persistence?.sessionId ?? undefined,
-      shouldReplace,
     },
     "agent.session.start_stream.iterator_returned",
   );

packages/server/src/server/agent/agent-sdk-types.ts

@@ -153,6 +153,7 @@ export interface AgentCapabilityFlags {
   supportsRewindConversation?: boolean;
   supportsRewindFiles?: boolean;
   supportsRewindBoth?: boolean;
+  supportsSteering?: boolean;
 }
 
 export interface AgentPersistenceHandle {
@@ -558,6 +559,7 @@ export interface AgentSession {
   readonly features?: AgentFeature[];
   run(prompt: AgentPromptInput, options?: AgentRunOptions): Promise<AgentRunResult>;
   startTurn(prompt: AgentPromptInput, options?: AgentRunOptions): Promise<{ turnId: string }>;
+  steerTurn?(prompt: AgentPromptInput, options?: AgentRunOptions): Promise<void>;
   subscribe(callback: (event: AgentStreamEvent) => void): () => void;
   streamHistory(): AsyncGenerator<AgentStreamEvent>;
   getRuntimeInfo(): Promise<AgentRuntimeInfo>;

packages/server/src/server/agent/permission-response.test.ts

@@ -8,6 +8,7 @@ import type {
   AgentPermissionResponse,
 } from "./agent-sdk-types.js";
 import type { AgentStreamEvent } from "../messages.js";
+import type { StartAgentRunOptions, StartAgentRunResult } from "./agent-manager.js";
 import { respondToAgentPermission } from "./permission-response.js";
 
 class FakePermissionAgentManager {
@@ -61,6 +62,21 @@ class FakePermissionAgentManager {
     this.replacementRuns.push({ agentId, prompt, options });
     return emptyAgentStream();
   }
+
+  startAgentRun(
+    agentId: string,
+    prompt: AgentPromptInput,
+    options?: StartAgentRunOptions,
+  ): StartAgentRunResult {
+    if (this.tryRunOutOfBand()) {
+      return { outOfBand: true };
+    }
+    const shouldReplace = Boolean(options?.replaceRunning && this.hasInFlightRun());
+    const events = shouldReplace
+      ? this.replaceAgentRun(agentId, prompt, options?.runOptions)
+      : this.streamAgent(agentId, prompt, options?.runOptions);
+    return { outOfBand: false, events };
+  }
 }
 
 async function* emptyAgentStream(): AsyncGenerator<AgentStreamEvent> {}

packages/server/src/server/agent/provider-registry.ts

@@ -351,6 +351,7 @@ export function wrapSessionProvider(provider: AgentProvider, inner: AgentSession
     },
     run: (prompt, options) => inner.run(prompt, options),
     startTurn: (prompt, options) => inner.startTurn(prompt, options),
+    steerTurn: inner.steerTurn?.bind(inner),
     subscribe: (callback) => inner.subscribe((event) => callback(mapStreamEvent(provider, event))),
     async *streamHistory() {
       for await (const event of inner.streamHistory()) {

packages/server/src/server/agent/providers/claude/agent.test.ts

@@ -2,7 +2,7 @@ import * as fs from "node:fs/promises";
 import * as os from "node:os";
 import * as path from "node:path";
 import { afterEach, describe, expect, test, vi } from "vitest";
-import type { SDKMessage } from "@anthropic-ai/claude-agent-sdk";
+import type { SDKMessage, SDKUserMessage } from "@anthropic-ai/claude-agent-sdk";
 
 import { createTestLogger } from "../../../../test-utils/test-logger.js";
 import * as executableUtils from "../../../../utils/executable.js";
@@ -749,7 +749,10 @@ describe("ClaudeAgentSession context window usage", () => {
     return session as unknown as TestClaudeSession;
   }
 
-  function createQueryFactoryForTurns(turns: Array<Array<Record<string, unknown>>>) {
+  function createQueryFactoryForTurns(
+    turns: Array<Array<Record<string, unknown>>>,
+    onPrompt?: (prompt: unknown) => void,
+  ) {
     return vi.fn(({ prompt }: { prompt: AsyncIterable<unknown> }) => {
       const queuedMessages: Array<Record<string, unknown>> = [];
       const waiters: Array<() => void> = [];
@@ -768,6 +771,7 @@ describe("ClaudeAgentSession context window usage", () => {
 
       void (async () => {
         for await (const _prompt of prompt) {
+          onPrompt?.(_prompt);
           const turnMessages = turns[turnIndex] ?? [];
           turnIndex += 1;
           for (const message of turnMessages) {
@@ -1003,6 +1007,42 @@ describe("ClaudeAgentSession context window usage", () => {
     }
   });
 
+  test("steerTurn pushes a priority next user message into the existing input stream", async () => {
+    const capturedPrompts: SDKUserMessage[] = [];
+    const queryFactory = createQueryFactoryForTurns([], (prompt) => {
+      capturedPrompts.push(prompt as SDKUserMessage);
+    });
+    const client = new ClaudeAgentClient({ logger, queryFactory });
+    const session = await client.createSession({
+      provider: "claude",
+      cwd: process.cwd(),
+    });
+
+    try {
+      await session.startTurn("first prompt");
+      await vi.waitFor(() => {
+        expect(capturedPrompts).toHaveLength(1);
+      });
+
+      await expect(session.steerTurn?.("steered prompt")).resolves.toBeUndefined();
+
+      await vi.waitFor(() => {
+        expect(capturedPrompts).toHaveLength(2);
+      });
+      expect(capturedPrompts[1]).toMatchObject({
+        type: "user",
+        priority: "next",
+        message: {
+          role: "user",
+          content: [{ type: "text", text: "steered prompt" }],
+        },
+      });
+      expect(queryFactory).toHaveBeenCalledTimes(1);
+    } finally {
+      await session.close();
+    }
+  });
+
   test("convertUsage includes contextWindowMaxTokens and derives used tokens from result usage as initial fallback", async () => {
     const session = await createSessionForTest();
 

packages/server/src/server/agent/providers/claude/agent.ts

@@ -216,6 +216,7 @@ const CLAUDE_CAPABILITIES: AgentCapabilityFlags = {
   supportsRewindConversation: true,
   supportsRewindFiles: true,
   supportsRewindBoth: true,
+  supportsSteering: true,
 };
 
 const DEFAULT_MODES: AgentMode[] = [
@@ -1789,6 +1790,20 @@ class ClaudeAgentSession implements AgentSession {
     return { turnId };
   }
 
+  async steerTurn(prompt: AgentPromptInput, _options?: AgentRunOptions): Promise<void> {
+    if (this.closed) {
+      throw new Error("Claude session is closed");
+    }
+    const sdkMessage = this.toSdkUserMessage(prompt);
+    sdkMessage.priority = "next";
+
+    await this.ensureQuery();
+    if (!this.input) {
+      throw new Error("Claude session input stream not initialized");
+    }
+    this.input.push(sdkMessage);
+  }
+
   subscribe(callback: (event: AgentStreamEvent) => void): () => void {
     this.subscribers.add(callback);
     return () => {

packages/server/src/server/agent/providers/codex-app-server-agent.test.ts

@@ -63,6 +63,7 @@ type CodexTestSession = AgentSession & {
   connected: boolean;
   currentThreadId: string | null;
   activeForegroundTurnId: string | null;
+  activeAppServerTurnId: string | null;
   client: CodexClientLike | null;
 };
 
@@ -99,6 +100,7 @@ function createSession(
   session.connected = true;
   session.currentThreadId = "test-thread";
   session.activeForegroundTurnId = "test-turn";
+  session.activeAppServerTurnId = "test-turn";
   return session;
 }
 
@@ -395,6 +397,12 @@ describe("Codex app-server provider", () => {
     );
   });
 
+  test("advertises steering support for app-server sessions", () => {
+    const session = createSession();
+
+    expect(session.capabilities.supportsSteering).toBe(true);
+  });
+
   test("passes ephemeral: true to thread/start when constructed as ephemeral", async () => {
     const requests: Array<{ method: string; params: unknown }> = [];
     const fakeClient: CodexClientLike = {
@@ -880,6 +888,161 @@ describe("Codex app-server provider", () => {
     );
   });
 
+  test("steers against the app-server turn id returned by turn/start", async () => {
+    const session = createSession();
+    const request = vi.fn(async (method: string) => {
+      if (method === "thread/loaded/list") {
+        return { data: ["test-thread"] };
+      }
+      if (method === "turn/start") {
+        return { turn: { id: "app-server-turn-from-start" } };
+      }
+      if (method === "turn/steer") {
+        return { turnId: "app-server-turn-from-start" };
+      }
+      throw new Error(`Unexpected request: ${method}`);
+    });
+    session.activeForegroundTurnId = null;
+    session.client = createStub<CodexClientLike>({ request });
+
+    const started = await session.startTurn("Start the turn");
+    expect(started.turnId).not.toBe("app-server-turn-from-start");
+
+    await session.steerTurn!("Steer the active turn");
+
+    expect(request).toHaveBeenCalledWith(
+      "turn/steer",
+      expect.objectContaining({
+        expectedTurnId: "app-server-turn-from-start",
+      }),
+    );
+  });
+
+  test("waits for the app-server turn id before steering Codex", async () => {
+    const session = createSession();
+    const request = vi.fn(async (method: string) => {
+      if (method === "thread/loaded/list") {
+        return { data: ["test-thread"] };
+      }
+      if (method === "turn/start") {
+        return {};
+      }
+      if (method === "turn/steer") {
+        return {};
+      }
+      throw new Error(`Unexpected request: ${method}`);
+    });
+    session.activeForegroundTurnId = null;
+    session.client = createStub<CodexClientLike>({ request });
+
+    await session.startTurn("Start the turn");
+    await expect(session.steerTurn!("Too early")).rejects.toThrow(
+      "Cannot steer Codex turn without an active app-server turn",
+    );
+
+    asInternals(session).handleNotification("turn/started", {
+      turn: { id: "app-server-turn-from-notification" },
+    });
+
+    await session.steerTurn!("Steer after notification");
+
+    expect(request).toHaveBeenCalledWith(
+      "turn/steer",
+      expect.objectContaining({
+        expectedTurnId: "app-server-turn-from-notification",
+      }),
+    );
+  });
+
+  test("sends turn/steer with thread id, built input, and active foreground turn id", async () => {
+    const session = createSession();
+    const request = vi.fn(async (method: string) => {
+      if (method === "turn/steer") {
+        return {};
+      }
+      throw new Error(`Unexpected request: ${method}`);
+    });
+    session.client = createStub<CodexClientLike>({ request });
+    session.currentThreadId = "thread-for-steer";
+    session.activeForegroundTurnId = "paseo-foreground-turn-for-steer";
+    session.activeAppServerTurnId = "app-server-turn-for-steer";
+
+    await session.steerTurn!("Use this guidance next.");
+
+    expect(request).toHaveBeenCalledTimes(1);
+    expect(request).toHaveBeenCalledWith("turn/steer", {
+      threadId: "thread-for-steer",
+      expectedTurnId: "app-server-turn-for-steer",
+      input: [
+        {
+          type: "text",
+          text: "Use this guidance next.",
+          text_elements: [],
+        },
+      ],
+    });
+  });
+
+  test("does not start a new turn when steering Codex", async () => {
+    const session = createSession();
+    const request = vi.fn(async (method: string) => {
+      if (method === "turn/steer") {
+        return {};
+      }
+      throw new Error(`Unexpected request: ${method}`);
+    });
+    session.client = createStub<CodexClientLike>({ request });
+
+    await session.steerTurn!("Keep going with this constraint.");
+
+    expect(request.mock.calls.some(([method]) => method === "turn/start")).toBe(false);
+    expect(session.activeForegroundTurnId).toBe("test-turn");
+  });
+
+  test("fails Codex steering without an initialized client", async () => {
+    const session = createSession();
+    session.client = null;
+    session.connected = true;
+
+    await expect(session.steerTurn!("hello")).rejects.toThrow("Codex client not initialized");
+  });
+
+  test("fails Codex steering without a current thread", async () => {
+    const session = createSession();
+    session.currentThreadId = null;
+    session.client = createStub<CodexClientLike>({
+      request: vi.fn(async () => ({})),
+    });
+
+    await expect(session.steerTurn!("hello")).rejects.toThrow(
+      "Cannot steer Codex turn without an active thread",
+    );
+  });
+
+  test("fails Codex steering without an active foreground turn", async () => {
+    const session = createSession();
+    session.activeForegroundTurnId = null;
+    session.client = createStub<CodexClientLike>({
+      request: vi.fn(async () => ({})),
+    });
+
+    await expect(session.steerTurn!("hello")).rejects.toThrow(
+      "Cannot steer Codex turn without an active foreground turn",
+    );
+  });
+
+  test("fails Codex steering without an active app-server turn id", async () => {
+    const session = createSession();
+    session.activeAppServerTurnId = null;
+    session.client = createStub<CodexClientLike>({
+      request: vi.fn(async () => ({})),
+    });
+
+    await expect(session.steerTurn!("hello")).rejects.toThrow(
+      "Cannot steer Codex turn without an active app-server turn",
+    );
+  });
+
   test("resolves Codex skill slash commands into app-server skill input", async () => {
     const session = createSession();
     const request = vi.fn(async (method: string) => {

packages/server/src/server/agent/providers/codex-app-server-agent.ts

@@ -171,6 +171,7 @@ const CODEX_APP_SERVER_CAPABILITIES: AgentCapabilityFlags = {
   supportsRewindConversation: true,
   supportsRewindFiles: false,
   supportsRewindBoth: false,
+  supportsSteering: true,
 };
 
 const CODEX_MODES: AgentMode[] = [
@@ -2888,6 +2889,7 @@ export class CodexAppServerAgentSession implements AgentSession {
   private readonly subscribers = new Set<(event: AgentStreamEvent) => void>();
   private nextTurnOrdinal = 0;
   private activeForegroundTurnId: string | null = null;
+  private activeAppServerTurnId: string | null = null;
   private cachedRuntimeInfo: AgentRuntimeInfo | null = null;
   private serviceTier: "fast" | null = null;
   private planModeEnabled = false;
@@ -3481,6 +3483,13 @@ export class CodexAppServerAgentSession implements AgentSession {
     );
   }
 
+  private async buildEffectivePromptInput(prompt: AgentPromptInput): Promise<CodexPromptInput> {
+    const slashCommand = await this.resolveSlashCommandInvocation(prompt);
+    return slashCommand
+      ? await this.buildCommandPromptInput(slashCommand.commandName, slashCommand.args)
+      : prompt;
+  }
+
   async run(prompt: AgentPromptInput, options?: AgentRunOptions): Promise<AgentRunResult> {
     return runProviderTurn({
       prompt,
@@ -3513,10 +3522,7 @@ export class CodexAppServerAgentSession implements AgentSession {
       throw new Error("Codex client not initialized");
     }
 
-    const slashCommand = await this.resolveSlashCommandInvocation(prompt);
-    const effectivePrompt = slashCommand
-      ? await this.buildCommandPromptInput(slashCommand.commandName, slashCommand.args)
-      : prompt;
+    const effectivePrompt = await this.buildEffectivePromptInput(prompt);
 
     if (this.currentThreadId) {
       await this.ensureThreadLoaded();
@@ -3528,6 +3534,7 @@ export class CodexAppServerAgentSession implements AgentSession {
 
     const turnId = this.createTurnId();
     this.activeForegroundTurnId = turnId;
+    this.activeAppServerTurnId = null;
 
     try {
       this.logTurnStartSummary({
@@ -3539,9 +3546,15 @@ export class CodexAppServerAgentSession implements AgentSession {
         hasDeveloperInstructions: turnStart.hasDeveloperInstructions,
         hasCodexConfig: turnStart.hasCodexConfig,
       });
-      await this.client.request("turn/start", turnStart.params, TURN_START_TIMEOUT_MS);
+      const response = await this.client.request(
+        "turn/start",
+        turnStart.params,
+        TURN_START_TIMEOUT_MS,
+      );
+      this.rememberStartedAppServerTurn(response);
     } catch (error) {
       this.activeForegroundTurnId = null;
+      this.activeAppServerTurnId = null;
       throw error;
     }
 
@@ -3583,6 +3596,36 @@ export class CodexAppServerAgentSession implements AgentSession {
     };
   }
 
+  private rememberStartedAppServerTurn(response: unknown): void {
+    const record = toObjectRecord(response);
+    const turn = toObjectRecord(record?.turn);
+    this.activeAppServerTurnId = typeof turn?.id === "string" ? turn.id : null;
+  }
+
+  async steerTurn(prompt: AgentPromptInput, _options?: AgentRunOptions): Promise<void> {
+    await this.connect();
+    if (!this.client) {
+      throw new Error("Codex client not initialized");
+    }
+    if (!this.currentThreadId) {
+      throw new Error("Cannot steer Codex turn without an active thread");
+    }
+    if (!this.activeForegroundTurnId) {
+      throw new Error("Cannot steer Codex turn without an active foreground turn");
+    }
+    if (!this.activeAppServerTurnId) {
+      throw new Error("Cannot steer Codex turn without an active app-server turn");
+    }
+
+    const effectivePrompt = await this.buildEffectivePromptInput(prompt);
+    const input = await this.buildUserInput(effectivePrompt);
+    await this.client.request("turn/steer", {
+      threadId: this.currentThreadId,
+      input,
+      expectedTurnId: this.activeAppServerTurnId,
+    });
+  }
+
   subscribe(callback: (event: AgentStreamEvent) => void): () => void {
     this.subscribers.add(callback);
     return () => {
@@ -3920,6 +3963,7 @@ export class CodexAppServerAgentSession implements AgentSession {
     this.resolvedPermissionRequests.clear();
     this.subscribers.clear();
     this.activeForegroundTurnId = null;
+    this.activeAppServerTurnId = null;
     if (this.client) {
       await this.client.dispose();
     }
@@ -4541,6 +4585,7 @@ export class CodexAppServerAgentSession implements AgentSession {
       return;
     }
     this.currentTurnId = parsed.turnId;
+    this.activeAppServerTurnId = parsed.turnId;
     this.resetTurnTrackingState();
     this.emitEvent({ type: "turn_started", provider: CODEX_PROVIDER });
   }
@@ -4578,6 +4623,7 @@ export class CodexAppServerAgentSession implements AgentSession {
       });
     }
     this.activeForegroundTurnId = null;
+    this.activeAppServerTurnId = null;
     this.resetTurnTrackingState();
   }