From 403a32ad9f38933e5a8f38fe77f9aa787e9d39c8 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Tue, 18 Aug 2026 15:59:13 +0200 Subject: [PATCH] fix(server): steer autonomous agent turns A running autonomous turn was excluded from steer admission because only foreground ownership was considered. That silently degraded steering into interruption and could race a provider-initiated restart, dropping the prompt.\n\nUse the provider's canonical active-turn identity for admission so autonomous work receives steering on its existing stream without replacement. --- .../src/server/agent/agent-manager.test.ts | 42 ++++++++++ .../server/src/server/agent/agent-manager.ts | 31 ++++---- ...agent.interrupt-restart-regression.test.ts | 77 +++++++++++++++++++ .../server/agent/providers/claude/agent.ts | 11 ++- 4 files changed, 145 insertions(+), 16 deletions(-) diff --git a/packages/server/src/server/agent/agent-manager.test.ts b/packages/server/src/server/agent/agent-manager.test.ts index 2f4bf7000..7331688f6 100644 --- a/packages/server/src/server/agent/agent-manager.test.ts +++ b/packages/server/src/server/agent/agent-manager.test.ts @@ -1783,6 +1783,48 @@ test("does not replace a newer foreground turn after unavailable steer fallback } }); +test("steers a tracked autonomous turn without creating a replacement run", async () => { + const session = new SteeringTestSession({ provider: "claude", cwd: process.cwd() }); + const workdir = mkdtempSync(join(tmpdir(), "agent-manager-autonomous-steer-")); + const client = new (class extends TestAgentClient { + override async createSession() { + return session; + } + })(); + const manager = new AgentManager({ clients: { claude: client }, logger }); + let agentId: string | null = null; + + try { + const agent = await manager.createAgent({ provider: "claude", cwd: workdir }, undefined, { + workspaceId: undefined, + }); + agentId = agent.id; + session.pushEvent({ type: "turn_started", provider: "claude", turnId: "active-turn-0" }); + await vi.waitFor(() => expect(manager.getAgent(agent.id)?.activeTurnId).toBe("active-turn-0")); + + const result = await startAgentRun(manager, agent.id, "autonomous follow-up", logger, { + activeTurnBehavior: "steer", + runOptions: { clientMessageId: "autonomous-follow-up-client" }, + }); + + expect(result).toEqual({ disposition: "steered" }); + expect(session.steerCount).toBe(1); + expect(session.interruptCount).toBe(0); + expect(manager.getAgent(agent.id)?.activeTurnId).toBe("active-turn-0"); + expect(manager.getAgent(agent.id)?.activeForegroundTurnId).toBeNull(); + expect(manager.getTimeline(agent.id)).toContainEqual( + expect.objectContaining({ + type: "user_message", + text: "autonomous follow-up", + clientMessageId: "autonomous-follow-up-client", + }), + ); + } finally { + if (agentId) await manager.closeAgent(agentId).catch(() => undefined); + rmSync(workdir, { recursive: true, force: true }); + } +}); + test("isolated rewind falls back from steering to the normal replacement path", async () => { const session = new SteeringTestSession({ provider: "claude", cwd: process.cwd() }); session.steerResult = "unavailable"; diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index eb71c2205..9a1838a78 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -2398,8 +2398,8 @@ export class AgentManager { options?: AgentRunOptions, ): Promise { const agent = this.requireSessionAgent(agentId); - const expectedTurnId = agent.activeForegroundTurnId; - if (!expectedTurnId || !this.runs.hasRun(agentId) || !agent.session.steerActiveTurn) { + const expectedTurnId = agent.activeForegroundTurnId ?? agent.activeTurnId; + if (!expectedTurnId || !agent.session.steerActiveTurn) { return { status: "unavailable" }; } const result = await this.runSteerAdmission(agent, expectedTurnId, async () => { @@ -2413,8 +2413,8 @@ export class AgentManager { return admission; }); // An unavailable answer is only safe to fall back from while this admission - // still owns the foreground. Never let an A admission replace a later B. - if (result.status === "unavailable" && agent.activeForegroundTurnId !== expectedTurnId) { + // still owns the active turn. Never let an A admission replace a later B. + if (result.status === "unavailable" && agent.activeTurnId !== expectedTurnId) { throw new Error("Active turn changed before steering could be delivered"); } return result; @@ -2426,8 +2426,8 @@ export class AgentManager { options?: AgentRunOptions, ): Promise { const agent = this.requireSessionAgent(agentId); - const expectedTurnId = agent.activeForegroundTurnId; - if (!expectedTurnId || !this.runs.hasRun(agentId)) { + const expectedTurnId = agent.activeForegroundTurnId ?? agent.activeTurnId; + if (!expectedTurnId) { return { status: "inactive" }; } @@ -2447,19 +2447,22 @@ export class AgentManager { return { status: "steered" }; } + // Providers without autonomous steering keep their existing dispatch behavior. The shared + // admission may recognize the turn, but only an accepted steer can own it without replacement. + if (agent.activeForegroundTurnId === null && agent.activeTurnId === expectedTurnId) { + return { status: "inactive" }; + } + await this.beforeSteerUnavailableFallback?.({ agentId, expectedTurnId }); - this.assertSteerAdmissionOwnsForeground(agent, expectedTurnId); + this.assertSteerAdmissionOwnsTurn(agent, expectedTurnId); return { status: "replaced", iterator: await this.replaceAdmittedForegroundTurn(agent, expectedTurnId, prompt, options), }; } - private assertSteerAdmissionOwnsForeground( - agent: ActiveManagedAgent, - expectedTurnId: string, - ): void { - if (agent.activeForegroundTurnId !== expectedTurnId || !this.runs.hasRun(agent.id)) { + private assertSteerAdmissionOwnsTurn(agent: ActiveManagedAgent, expectedTurnId: string): void { + if (agent.activeTurnId !== expectedTurnId) { throw new Error("Active turn changed before steering could be delivered"); } } @@ -2472,7 +2475,7 @@ export class AgentManager { return this.runForegroundMutation(agent.id, async () => { await this.drainSessionEvents(agent.id); this.agentStreamCoalescer.flushFor(agent.id); - this.assertSteerAdmissionOwnsForeground(agent, expectedTurnId); + this.assertSteerAdmissionOwnsTurn(agent, expectedTurnId); const barrier: SteerEventBarrier = { events: [] }; this.steerEventBarriers.set(agent.id, barrier); try { @@ -2512,7 +2515,7 @@ export class AgentManager { prompt: AgentPromptInput, options?: AgentRunOptions, ): Promise> { - this.assertSteerAdmissionOwnsForeground(agent, expectedTurnId); + this.assertSteerAdmissionOwnsTurn(agent, expectedTurnId); agent.pendingReplacement = true; agent.lifecycle = "running"; this.touchUpdatedAt(agent); diff --git a/packages/server/src/server/agent/providers/claude/agent.interrupt-restart-regression.test.ts b/packages/server/src/server/agent/providers/claude/agent.interrupt-restart-regression.test.ts index c4f002d16..7e785bdda 100644 --- a/packages/server/src/server/agent/providers/claude/agent.interrupt-restart-regression.test.ts +++ b/packages/server/src/server/agent/providers/claude/agent.interrupt-restart-regression.test.ts @@ -620,6 +620,83 @@ test("creates an autonomous live turn when assistant output arrives without a fo await session.close(); }); +test("steers an autonomous turn through its existing query without restarting it", async () => { + const logger = createTestLogger(); + let queryRef: ScriptedQuery | null = null; + + queryFactory.mockImplementation(({ prompt }: { prompt: AsyncIterable }) => { + queryRef = createScriptedQuery({ + prompt, + sessionId: "autonomous-steer-session", + async handlePrompt({ promptRecord, query }) { + if (promptRecord.text === "seed prompt") { + query.emit({ + type: "assistant", + message: { content: "SEED_RESPONSE" }, + session_id: "autonomous-steer-session", + }); + query.emit(buildSuccessResult("autonomous-steer-session")); + return; + } + if (promptRecord.text === "steer prompt") { + query.emit({ + type: "assistant", + message: { content: "STEERED_RESPONSE" }, + session_id: "autonomous-steer-session", + }); + query.emit(buildSuccessResult("autonomous-steer-session")); + } + }, + }); + return queryRef; + }); + + const session = await new ClaudeAgentClient({ + logger, + queryFactory, + resolveBinary: async () => "/test/claude/bin", + }).createSession({ provider: "claude", cwd: process.cwd() }); + + await collectUntilTerminal(streamSession(session, "seed prompt")); + const autonomousEvents = subscribeToEvents(session); + queryRef?.emit({ + type: "assistant", + message: { content: "AUTONOMOUS_RESPONSE" }, + session_id: "autonomous-steer-session", + }); + const autonomousStart = await autonomousEvents.next(); + const autonomousTimeline = await autonomousEvents.next(); + const autonomousTurnId = autonomousStart.value?.turnId; + expect(autonomousTurnId).toBeTruthy(); + + const steer = await session.steerActiveTurn!("steer prompt", { + expectedTurnId: autonomousTurnId!, + clientMessageId: "steer-client", + }); + const steeredTimeline = await autonomousEvents.next(); + const completion = await autonomousEvents.next(); + + expect(steer).toEqual({ status: "accepted" }); + expect(queryFactory).toHaveBeenCalledTimes(1); + expect(queryRef?.interrupt).not.toHaveBeenCalled(); + expect(queryRef?.prompts.map((prompt) => prompt.text)).toEqual(["seed prompt", "steer prompt"]); + expect(autonomousTimeline.value).toMatchObject({ turnId: autonomousTurnId }); + expect(steeredTimeline.value).toMatchObject({ + type: "timeline", + turnId: autonomousTurnId, + item: { type: "assistant_message", text: "STEERED_RESPONSE" }, + }); + expect(completion.value).toMatchObject({ type: "turn_completed", turnId: autonomousTurnId }); + expect( + [autonomousStart.value, autonomousTimeline.value, steeredTimeline.value, completion.value].map( + (event) => event?.turnId, + ), + ).toEqual([autonomousTurnId, autonomousTurnId, autonomousTurnId, autonomousTurnId]); + + autonomousEvents.close(); + await session.close(); +}); + test("auto-completes an open autonomous turn when a foreground prompt starts", async () => { const logger = createTestLogger(); let queryRef: ScriptedQuery | null = null; diff --git a/packages/server/src/server/agent/providers/claude/agent.ts b/packages/server/src/server/agent/providers/claude/agent.ts index 83d655fc2..0f4cab9e8 100644 --- a/packages/server/src/server/agent/providers/claude/agent.ts +++ b/packages/server/src/server/agent/providers/claude/agent.ts @@ -2266,7 +2266,8 @@ class ClaudeAgentSession implements AgentSession { if (this.resolveSlashCommandInvocation(prompt)) { return { status: "unavailable" }; } - if (this.compacting || this.activeForegroundTurnId !== options.expectedTurnId) { + const activeTurnId = this.activeForegroundTurnId ?? this.autonomousTurn?.id; + if (this.compacting || activeTurnId !== options.expectedTurnId) { return { status: "unavailable" }; } @@ -2280,7 +2281,7 @@ class ClaudeAgentSession implements AgentSession { const message = this.toSdkUserMessage(prompt); message.priority = "next"; if ( - this.activeForegroundTurnId !== options.expectedTurnId || + (this.activeForegroundTurnId ?? this.autonomousTurn?.id) !== options.expectedTurnId || this.activeForegroundQuery !== query || this.activeForegroundInput !== input || this.query !== query || @@ -3479,6 +3480,8 @@ class ClaudeAgentSession implements AgentSession { this.syncTurnState("foreground turn terminal"); } else if (this.autonomousTurn) { this.autonomousTurn = null; + this.activeForegroundQuery = null; + this.activeForegroundInput = null; this.activeTurnHasAssistantText = false; this.syncTurnState("autonomous turn terminal"); } @@ -3492,6 +3495,8 @@ class ClaudeAgentSession implements AgentSession { this.autonomousTurn = { id: this.createTurnId("autonomous"), }; + this.activeForegroundQuery = this.query; + this.activeForegroundInput = this.input; this.activeTurnHasAssistantText = false; this.contextUsage.beginTurn(); this.notifySubscribers({ type: "turn_started", provider: "claude" }); @@ -3504,6 +3509,8 @@ class ClaudeAgentSession implements AgentSession { } this.notifySubscribers({ type: "turn_completed", provider: "claude" }); this.autonomousTurn = null; + this.activeForegroundQuery = null; + this.activeForegroundInput = null; this.activeTurnHasAssistantText = false; this.syncTurnState("autonomous turn completed"); }