diff --git a/packages/server/src/server/agent/provider-registry.test.ts b/packages/server/src/server/agent/provider-registry.test.ts index 656349e10b..df69050d5f 100644 --- a/packages/server/src/server/agent/provider-registry.test.ts +++ b/packages/server/src/server/agent/provider-registry.test.ts @@ -7,6 +7,7 @@ import type { AgentFeature, AgentModelDefinition, AgentMode, + AgentSession, AgentSessionConfig, ProviderCatalog, } from "./agent-sdk-types.js"; @@ -532,6 +533,7 @@ import { AGENT_PROVIDER_DEFINITIONS, buildProviderRegistry, createAllClients, + wrapSessionProvider, } from "./provider-registry.js"; import { FakeOmp } from "./providers/omp/test-utils/fake-omp.js"; @@ -1831,3 +1833,57 @@ describe("fetchCatalog", () => { expect(catalog.modes.map((mode) => mode.id)).toEqual(["ask"]); }); }); + +describe("wrapSessionProvider", () => { + function makeFakeSession(overrides?: Partial): AgentSession { + return { + provider: "acp", + id: "inner-session-id", + capabilities: {}, + run: vi.fn(), + startTurn: vi.fn(), + subscribe: vi.fn(() => () => {}), + streamHistory: (async function* () {})(), + getRuntimeInfo: vi.fn(), + getAvailableModes: vi.fn(), + getCurrentMode: vi.fn(), + setMode: vi.fn(), + getPendingPermissions: vi.fn(() => []), + respondToPermission: vi.fn(), + describePersistence: vi.fn(() => null), + interrupt: vi.fn(), + close: vi.fn(), + ...overrides, + } satisfies Partial as AgentSession; + } + + // Any provider built on GenericACPAgentClient (cursor, kimi, kiro, + // traecli, or a user-configured `extends: "acp"` entry) reports its own + // `provider` as the base id ("acp"), which never matches the requested + // label, so createResolvedProviderClient always routes it through this + // wrapper. A forward missing here silently drops that capability for + // every such provider, with no type error, since AgentSession declares + // it optional. + test("forwards steerActiveTurn to the inner session when present", async () => { + const steerActiveTurn = vi.fn(async () => ({ status: "accepted" as const })); + const inner = makeFakeSession({ steerActiveTurn }); + + const wrapped = wrapSessionProvider("cursor", inner); + + expect(wrapped.steerActiveTurn).toBeDefined(); + const result = await wrapped.steerActiveTurn?.({ role: "user", content: [] } as never, { + expectedTurnId: "turn-1", + }); + expect(result).toEqual({ status: "accepted" }); + expect(steerActiveTurn).toHaveBeenCalledWith( + { role: "user", content: [] }, + { expectedTurnId: "turn-1" }, + ); + }); + + test("leaves steerActiveTurn undefined when the inner session doesn't implement it", () => { + const inner = makeFakeSession(); + const wrapped = wrapSessionProvider("cursor", inner); + expect(wrapped.steerActiveTurn).toBeUndefined(); + }); +}); diff --git a/packages/server/src/server/agent/provider-registry.ts b/packages/server/src/server/agent/provider-registry.ts index 58e45d37c7..3706d210f6 100644 --- a/packages/server/src/server/agent/provider-registry.ts +++ b/packages/server/src/server/agent/provider-registry.ts @@ -461,6 +461,7 @@ export function wrapSessionProvider(provider: AgentProvider, inner: AgentSession close: () => inner.close(), listCommands: inner.listCommands?.bind(inner), setModel: inner.setModel?.bind(inner), + steerActiveTurn: inner.steerActiveTurn?.bind(inner), setThinkingOption: inner.setThinkingOption?.bind(inner), setFeature: inner.setFeature?.bind(inner), revertConversation: inner.revertConversation?.bind(inner), diff --git a/packages/server/src/server/agent/providers/acp-agent.test.ts b/packages/server/src/server/agent/providers/acp-agent.test.ts index 20a0d1e075..0b108efe49 100644 --- a/packages/server/src/server/agent/providers/acp-agent.test.ts +++ b/packages/server/src/server/agent/providers/acp-agent.test.ts @@ -93,6 +93,15 @@ interface ACPSessionInternals { acpMcpServers(): unknown[]; } +interface ACPSteerInternals { + sessionId: string | null; + connection: { + extMethod: (method: string, params: unknown) => Promise<{ outcome: string }>; + }; + steeringSupported: boolean; + activeForegroundTurnId: string | null; +} + interface ACPModelSelectionInternals { sessionId: string | null; connection: { @@ -148,6 +157,117 @@ function createSession(terminateProcess?: ProcessTerminator): ACPAgentSession { ); } +describe("ACPAgentSession.steerActiveTurn", () => { + function primeForSteer( + session: ACPAgentSession, + overrides: Partial = {}, + ): { internals: ACPSteerInternals; extMethod: ReturnType } { + const extMethod = vi.fn(async () => ({ outcome: "injected" })); + const internals = asInternals(session); + internals.sessionId = "session-1"; + internals.connection = { extMethod }; + internals.steeringSupported = true; + internals.activeForegroundTurnId = "turn-1"; + Object.assign(internals, overrides); + return { internals, extMethod }; + } + + test("returns unavailable when the agent never advertised steering support", async () => { + const session = createSession(); + primeForSteer(session, { steeringSupported: false }); + + const result = await session.steerActiveTurn("steer this", { expectedTurnId: "turn-1" }); + + expect(result).toEqual({ status: "unavailable" }); + }); + + test("returns unavailable when the caller's expected turn doesn't match the live one", async () => { + // Guards against steering into the wrong turn: the caller's belief about + // what's running raced with the agent's own turn boundary. + const session = createSession(); + const { extMethod } = primeForSteer(session, { activeForegroundTurnId: "turn-2" }); + + const result = await session.steerActiveTurn("steer this", { expectedTurnId: "turn-1" }); + + expect(result).toEqual({ status: "unavailable" }); + expect(extMethod).not.toHaveBeenCalled(); + }); + + test("returns unavailable when there is no live connection", async () => { + const session = createSession(); + primeForSteer(session, { connection: undefined as never }); + + const result = await session.steerActiveTurn("steer this", { expectedTurnId: "turn-1" }); + + expect(result).toEqual({ status: "unavailable" }); + }); + + test("forwards to _session/steering with the session id and prompt content", async () => { + const session = createSession(); + const { extMethod } = primeForSteer(session); + + await session.steerActiveTurn("steer this", { expectedTurnId: "turn-1" }); + + expect(extMethod).toHaveBeenCalledWith("_session/steering", { + sessionId: "session-1", + prompt: [{ type: "text", text: "steer this" }], + _meta: { steering: { idleBehavior: "promptRequired" } }, + }); + }); + + test.each([ + ["injected", "accepted"], + ["startedNewTurn", "unavailable"], + ["promptRequired", "unavailable"], + ["failed", "unavailable"], + ] as const)("maps agent outcome %s to status %s", async (outcome, status) => { + const session = createSession(); + const { extMethod } = primeForSteer(session); + extMethod.mockResolvedValue({ outcome }); + + const result = await session.steerActiveTurn("steer this", { expectedTurnId: "turn-1" }); + + expect(result).toEqual({ status }); + }); + + test("denies pending permissions after the steer is accepted", async () => { + const session = createSession(); + primeForSteer(session); + const events: AgentStreamEvent[] = []; + session.subscribe((event) => events.push(event)); + const permission = session.requestPermission({ + sessionId: "session-1", + toolCall: { + toolCallId: "tool-1", + title: "Run command", + kind: "execute", + status: "pending", + }, + options: [ + { optionId: "allow-once", name: "Allow", kind: "allow_once" }, + { optionId: "reject-once", name: "Reject", kind: "reject_once" }, + ], + } satisfies RequestPermissionRequest); + await Promise.resolve(); + + await session.steerActiveTurn("steer this", { + expectedTurnId: "turn-1", + clearPendingPermissions: true, + }); + + await expect(permission).resolves.toEqual({ + outcome: { outcome: "selected", optionId: "reject-once" }, + }); + expect(session.getPendingPermissions()).toEqual([]); + expect(events).toContainEqual( + expect.objectContaining({ + type: "permission_resolved", + resolution: expect.objectContaining({ behavior: "deny" }), + }), + ); + }); +}); + // Typed substitute for the real tree-kill terminator. Records which child // processes it was asked to terminate, so tests assert on observable state // instead of spying on the production function. In "deferred" mode the diff --git a/packages/server/src/server/agent/providers/acp-agent.ts b/packages/server/src/server/agent/providers/acp-agent.ts index fcf97723a7..de90f9dd90 100644 --- a/packages/server/src/server/agent/providers/acp-agent.ts +++ b/packages/server/src/server/agent/providers/acp-agent.ts @@ -94,6 +94,8 @@ import { type ProviderCatalog, type ResolveAgentCreateConfigInput, type ResolveAgentCreateConfigResult, + type SteerActiveTurnOptions, + type SteerResult, type ToolCallDetail, type ToolCallTimelineItem, } from "../agent-sdk-types.js"; @@ -1456,6 +1458,7 @@ export class ACPAgentSession implements AgentSession, ACPClient { private readonly extensionCommandsParser?: ACPExtensionCommandsParser; private currentTurnUsage: AgentUsage | undefined; private activeForegroundTurnId: string | null = null; + private steeringSupported = false; private fallbackAssistantMessageId: string | null = null; private closed = false; private historyPending = false; @@ -1506,6 +1509,8 @@ export class ACPAgentSession implements AgentSession, ACPClient { this.child = spawned.child; this.connection = spawned.connection; this.agentCapabilities = spawned.initialize.agentCapabilities ?? null; + this.steeringSupported = + readRecord(readRecord(spawned.initialize._meta)?.steering)?.supported === true; const response = await this.runACPRequest(() => this.connection!.newSession({ @@ -1540,6 +1545,8 @@ export class ACPAgentSession implements AgentSession, ACPClient { this.child = spawned.child; this.connection = spawned.connection; this.agentCapabilities = spawned.initialize.agentCapabilities ?? null; + this.steeringSupported = + readRecord(readRecord(spawned.initialize._meta)?.steering)?.supported === true; this.sessionId = handle.sessionId; this.bootstrapThreadEventPending = true; @@ -1654,6 +1661,45 @@ export class ACPAgentSession implements AgentSession, ACPClient { return { turnId }; } + async steerActiveTurn( + prompt: AgentPromptInput, + options: SteerActiveTurnOptions, + ): Promise { + if ( + !this.connection || + !this.sessionId || + !this.steeringSupported || + this.activeForegroundTurnId !== options.expectedTurnId + ) { + return { status: "unavailable" }; + } + const response = await this.runACPRequest(() => + this.connection!.extMethod("_session/steering", { + sessionId: this.sessionId, + prompt: toACPContentBlocks(prompt), + _meta: { steering: { idleBehavior: "promptRequired" } }, + }), + ); + if (response.outcome !== "injected") { + return { status: "unavailable" }; + } + if (options.clearPendingPermissions) { + await this.clearPendingPermissionsForSteer(); + } + return { status: "accepted" }; + } + + private async clearPendingPermissionsForSteer(): Promise { + const requestIds = Array.from(this.pendingPermissions.keys()); + for (const requestId of requestIds) { + if (!this.pendingPermissions.has(requestId)) continue; + await this.respondToPermission(requestId, { + behavior: "deny", + message: "The user answered with a message instead of approving. Their message follows.", + }); + } + } + subscribe(callback: (event: AgentStreamEvent) => void): () => void { this.subscribers.add(callback); if (this.sessionId) {