diff --git a/packages/app/src/agent-stream/view.tsx b/packages/app/src/agent-stream/view.tsx index 4e77405cc6..c2f591f39d 100644 --- a/packages/app/src/agent-stream/view.tsx +++ b/packages/app/src/agent-stream/view.tsx @@ -423,6 +423,7 @@ const AgentStreamViewComponent = forwardRef ); }, diff --git a/packages/app/src/components/message.tsx b/packages/app/src/components/message.tsx index f4e73aa821..fbc9a09f05 100644 --- a/packages/app/src/components/message.tsx +++ b/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({ /> ) : null} + {deliveryHint === "steering" ? ( + + Steering conversation + + ) : null} ); diff --git a/packages/app/src/composer/actions.test.ts b/packages/app/src/composer/actions.test.ts index ab6266b09b..1b2d74af92 100644 --- a/packages/app/src/composer/actions.test.ts +++ b/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; + 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; + expect(userMessage.deliveryHint).toBeUndefined(); + }); + it("serializes browser_element workspace attachments as text attachments at the wire boundary", async () => { const client = createFakeSendClient(); const stream = createFakeStream(); diff --git a/packages/app/src/composer/actions.ts b/packages/app/src/composer/actions.ts index 82d61c63d4..96e98ea162 100644 --- a/packages/app/src/composer/actions.ts +++ b/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 | 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); diff --git a/packages/app/src/composer/index.tsx b/packages/app/src/composer/index.tsx index 2d10d7f702..1d0d257cdc 100644 --- a/packages/app/src/composer/index.tsx +++ b/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?.(); }; diff --git a/packages/app/src/contexts/session-context.tsx b/packages/app/src/contexts/session-context.tsx index 5adae71e74..01c9459446 100644 --- a/packages/app/src/contexts/session-context.tsx +++ b/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 diff --git a/packages/app/src/types/stream.test.ts b/packages/app/src/types/stream.test.ts index 019f083022..70142c42f0 100644 --- a/packages/app/src/types/stream.test.ts +++ b/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[] = [ { diff --git a/packages/app/src/types/stream.ts b/packages/app/src/types/stream.ts index 44bfa2e354..880751667d 100644 --- a/packages/app/src/types/stream.ts +++ b/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 } : {}), }; } diff --git a/packages/protocol/src/agent-types.ts b/packages/protocol/src/agent-types.ts index 7a1f9c04bc..394675b5fd 100644 --- a/packages/protocol/src/agent-types.ts +++ b/packages/protocol/src/agent-types.ts @@ -125,6 +125,7 @@ export interface AgentCapabilityFlags { supportsRewindConversation?: boolean; supportsRewindFiles?: boolean; supportsRewindBoth?: boolean; + supportsSteering?: boolean; } export interface AgentPersistenceHandle { diff --git a/packages/protocol/src/messages.providers-snapshot.test.ts b/packages/protocol/src/messages.providers-snapshot.test.ts index 1b066062d4..c14f7373b8 100644 --- a/packages/protocol/src/messages.providers-snapshot.test.ts +++ b/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", diff --git a/packages/protocol/src/messages.ts b/packages/protocol/src/messages.ts index 95ecf2a095..47cc1f9865 100644 --- a/packages/protocol/src/messages.ts +++ b/packages/protocol/src/messages.ts @@ -251,6 +251,8 @@ const AgentCapabilityFlagsSchema: z.ZodType = 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 = z.object({ diff --git a/packages/server/src/server/agent/agent-manager.test.ts b/packages/server/src/server/agent/agent-manager.test.ts index d4824e0f85..eb3687b16f 100644 --- a/packages/server/src/server/agent/agent-manager.test.ts +++ b/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(); + 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 { + 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 { + 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 {} + async getRuntimeInfo() { + return { provider: this.provider, sessionId: this.id, model: null, modeId: null }; + } + async getAvailableModes() { + return []; + } + async getCurrentMode() { + return null; + } + async setMode(): Promise {} + getPendingPermissions() { + return []; + } + async respondToPermission(): Promise {} + describePersistence() { + return { provider: this.provider, sessionId: this.id }; + } + async close(): Promise {} + } + + class SteerableClient extends TestAgentClient { + readonly capabilities = steeringCapabilities; + + override async createSession(_config: AgentSessionConfig): Promise { + 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(); + const allowSecondRunToEnd = deferred(); + 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 { + 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 {} + async getRuntimeInfo() { + return { provider: this.provider, sessionId: this.id, model: null, modeId: null }; + } + async getAvailableModes() { + return []; + } + async getCurrentMode() { + return null; + } + async setMode(): Promise {} + getPendingPermissions() { + return []; + } + async respondToPermission(): Promise {} + describePersistence() { + return { provider: this.provider, sessionId: this.id }; + } + async interrupt(): Promise { + interruptCount += 1; + allowFirstRunToCancel.resolve(); + } + async close(): Promise {} + } + + class UnsupportedSteeringClient extends TestAgentClient { + override async createSession(_config: AgentSessionConfig): Promise { + 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"); diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index 1e8fe4ba89..6db318b1d3 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/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 }; + 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 { + 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 { const agent = this.requireAgent(agentId); this.touchUpdatedAt(agent); diff --git a/packages/server/src/server/agent/agent-prompt.test.ts b/packages/server/src/server/agent/agent-prompt.test.ts index d8eb8eb2a6..597fce9e11 100644 --- a/packages/server/src/server/agent/agent-prompt.test.ts +++ b/packages/server/src/server/agent/agent-prompt.test.ts @@ -28,9 +28,6 @@ it("does not notify archived callers", async () => { 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, @@ -55,9 +52,11 @@ 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, @@ -91,6 +90,87 @@ it("does not notify archived callers", async () => { expect(agentStorageGetSpy).toHaveBeenCalledWith("caller-agent"); }); - 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"); + 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, "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") { + return childAgent; + } + if (agentId === "caller-agent") { + 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 () => null), + ); + + setupFinishNotification({ + agentManager, + agentStorage, + childAgentId: "child-agent", + callerAgentId: "caller-agent", + 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", + "\nAgent child-agent (Child Agent) finished.\n", + { replaceRunning: true }, + ); + }); }); diff --git a/packages/server/src/server/agent/agent-sdk-types.ts b/packages/server/src/server/agent/agent-sdk-types.ts index 7227de4eaf..472ef76d74 100644 --- a/packages/server/src/server/agent/agent-sdk-types.ts +++ b/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; startTurn(prompt: AgentPromptInput, options?: AgentRunOptions): Promise<{ turnId: string }>; + steerTurn?(prompt: AgentPromptInput, options?: AgentRunOptions): Promise; subscribe(callback: (event: AgentStreamEvent) => void): () => void; streamHistory(): AsyncGenerator; getRuntimeInfo(): Promise; diff --git a/packages/server/src/server/agent/provider-registry.ts b/packages/server/src/server/agent/provider-registry.ts index b474047fe7..e481c3d2e5 100644 --- a/packages/server/src/server/agent/provider-registry.ts +++ b/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()) { diff --git a/packages/server/src/server/agent/providers/claude/agent.test.ts b/packages/server/src/server/agent/providers/claude/agent.test.ts index be905bf9e6..b8648fa3a3 100644 --- a/packages/server/src/server/agent/providers/claude/agent.test.ts +++ b/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>>) { + function createQueryFactoryForTurns( + turns: Array>>, + onPrompt?: (prompt: unknown) => void, + ) { return vi.fn(({ prompt }: { prompt: AsyncIterable }) => { const queuedMessages: Array> = []; 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(); diff --git a/packages/server/src/server/agent/providers/claude/agent.ts b/packages/server/src/server/agent/providers/claude/agent.ts index d35aa1bd18..6298489eec 100644 --- a/packages/server/src/server/agent/providers/claude/agent.ts +++ b/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 { + 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 () => { diff --git a/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts b/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts index ffa0b66d2d..8f85cd08a7 100644 --- a/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts +++ b/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({ 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({ 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({ 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({ 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({ + 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({ + 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({ + 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) => { diff --git a/packages/server/src/server/agent/providers/codex-app-server-agent.ts b/packages/server/src/server/agent/providers/codex-app-server-agent.ts index 78609f13f1..37dcc8faf7 100644 --- a/packages/server/src/server/agent/providers/codex-app-server-agent.ts +++ b/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 { + const slashCommand = await this.resolveSlashCommandInvocation(prompt); + return slashCommand + ? await this.buildCommandPromptInput(slashCommand.commandName, slashCommand.args) + : prompt; + } + async run(prompt: AgentPromptInput, options?: AgentRunOptions): Promise { 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 { + 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(); }