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.
This commit is contained in:
Mohamed Boudra
2026-08-18 15:59:13 +02:00
parent bd3f3c7714
commit 403a32ad9f
4 changed files with 145 additions and 16 deletions
@@ -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 () => { test("isolated rewind falls back from steering to the normal replacement path", async () => {
const session = new SteeringTestSession({ provider: "claude", cwd: process.cwd() }); const session = new SteeringTestSession({ provider: "claude", cwd: process.cwd() });
session.steerResult = "unavailable"; session.steerResult = "unavailable";
@@ -2398,8 +2398,8 @@ export class AgentManager {
options?: AgentRunOptions, options?: AgentRunOptions,
): Promise<SteerResult> { ): Promise<SteerResult> {
const agent = this.requireSessionAgent(agentId); const agent = this.requireSessionAgent(agentId);
const expectedTurnId = agent.activeForegroundTurnId; const expectedTurnId = agent.activeForegroundTurnId ?? agent.activeTurnId;
if (!expectedTurnId || !this.runs.hasRun(agentId) || !agent.session.steerActiveTurn) { if (!expectedTurnId || !agent.session.steerActiveTurn) {
return { status: "unavailable" }; return { status: "unavailable" };
} }
const result = await this.runSteerAdmission(agent, expectedTurnId, async () => { const result = await this.runSteerAdmission(agent, expectedTurnId, async () => {
@@ -2413,8 +2413,8 @@ export class AgentManager {
return admission; return admission;
}); });
// An unavailable answer is only safe to fall back from while this 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. // still owns the active turn. Never let an A admission replace a later B.
if (result.status === "unavailable" && agent.activeForegroundTurnId !== expectedTurnId) { if (result.status === "unavailable" && agent.activeTurnId !== expectedTurnId) {
throw new Error("Active turn changed before steering could be delivered"); throw new Error("Active turn changed before steering could be delivered");
} }
return result; return result;
@@ -2426,8 +2426,8 @@ export class AgentManager {
options?: AgentRunOptions, options?: AgentRunOptions,
): Promise<ActiveTurnSteerDispatchResult> { ): Promise<ActiveTurnSteerDispatchResult> {
const agent = this.requireSessionAgent(agentId); const agent = this.requireSessionAgent(agentId);
const expectedTurnId = agent.activeForegroundTurnId; const expectedTurnId = agent.activeForegroundTurnId ?? agent.activeTurnId;
if (!expectedTurnId || !this.runs.hasRun(agentId)) { if (!expectedTurnId) {
return { status: "inactive" }; return { status: "inactive" };
} }
@@ -2447,19 +2447,22 @@ export class AgentManager {
return { status: "steered" }; 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 }); await this.beforeSteerUnavailableFallback?.({ agentId, expectedTurnId });
this.assertSteerAdmissionOwnsForeground(agent, expectedTurnId); this.assertSteerAdmissionOwnsTurn(agent, expectedTurnId);
return { return {
status: "replaced", status: "replaced",
iterator: await this.replaceAdmittedForegroundTurn(agent, expectedTurnId, prompt, options), iterator: await this.replaceAdmittedForegroundTurn(agent, expectedTurnId, prompt, options),
}; };
} }
private assertSteerAdmissionOwnsForeground( private assertSteerAdmissionOwnsTurn(agent: ActiveManagedAgent, expectedTurnId: string): void {
agent: ActiveManagedAgent, if (agent.activeTurnId !== expectedTurnId) {
expectedTurnId: string,
): void {
if (agent.activeForegroundTurnId !== expectedTurnId || !this.runs.hasRun(agent.id)) {
throw new Error("Active turn changed before steering could be delivered"); throw new Error("Active turn changed before steering could be delivered");
} }
} }
@@ -2472,7 +2475,7 @@ export class AgentManager {
return this.runForegroundMutation(agent.id, async () => { return this.runForegroundMutation(agent.id, async () => {
await this.drainSessionEvents(agent.id); await this.drainSessionEvents(agent.id);
this.agentStreamCoalescer.flushFor(agent.id); this.agentStreamCoalescer.flushFor(agent.id);
this.assertSteerAdmissionOwnsForeground(agent, expectedTurnId); this.assertSteerAdmissionOwnsTurn(agent, expectedTurnId);
const barrier: SteerEventBarrier = { events: [] }; const barrier: SteerEventBarrier = { events: [] };
this.steerEventBarriers.set(agent.id, barrier); this.steerEventBarriers.set(agent.id, barrier);
try { try {
@@ -2512,7 +2515,7 @@ export class AgentManager {
prompt: AgentPromptInput, prompt: AgentPromptInput,
options?: AgentRunOptions, options?: AgentRunOptions,
): Promise<AsyncGenerator<AgentStreamEvent>> { ): Promise<AsyncGenerator<AgentStreamEvent>> {
this.assertSteerAdmissionOwnsForeground(agent, expectedTurnId); this.assertSteerAdmissionOwnsTurn(agent, expectedTurnId);
agent.pendingReplacement = true; agent.pendingReplacement = true;
agent.lifecycle = "running"; agent.lifecycle = "running";
this.touchUpdatedAt(agent); this.touchUpdatedAt(agent);
@@ -620,6 +620,83 @@ test("creates an autonomous live turn when assistant output arrives without a fo
await session.close(); 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<unknown> }) => {
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 () => { test("auto-completes an open autonomous turn when a foreground prompt starts", async () => {
const logger = createTestLogger(); const logger = createTestLogger();
let queryRef: ScriptedQuery | null = null; let queryRef: ScriptedQuery | null = null;
@@ -2266,7 +2266,8 @@ class ClaudeAgentSession implements AgentSession {
if (this.resolveSlashCommandInvocation(prompt)) { if (this.resolveSlashCommandInvocation(prompt)) {
return { status: "unavailable" }; 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" }; return { status: "unavailable" };
} }
@@ -2280,7 +2281,7 @@ class ClaudeAgentSession implements AgentSession {
const message = this.toSdkUserMessage(prompt); const message = this.toSdkUserMessage(prompt);
message.priority = "next"; message.priority = "next";
if ( if (
this.activeForegroundTurnId !== options.expectedTurnId || (this.activeForegroundTurnId ?? this.autonomousTurn?.id) !== options.expectedTurnId ||
this.activeForegroundQuery !== query || this.activeForegroundQuery !== query ||
this.activeForegroundInput !== input || this.activeForegroundInput !== input ||
this.query !== query || this.query !== query ||
@@ -3479,6 +3480,8 @@ class ClaudeAgentSession implements AgentSession {
this.syncTurnState("foreground turn terminal"); this.syncTurnState("foreground turn terminal");
} else if (this.autonomousTurn) { } else if (this.autonomousTurn) {
this.autonomousTurn = null; this.autonomousTurn = null;
this.activeForegroundQuery = null;
this.activeForegroundInput = null;
this.activeTurnHasAssistantText = false; this.activeTurnHasAssistantText = false;
this.syncTurnState("autonomous turn terminal"); this.syncTurnState("autonomous turn terminal");
} }
@@ -3492,6 +3495,8 @@ class ClaudeAgentSession implements AgentSession {
this.autonomousTurn = { this.autonomousTurn = {
id: this.createTurnId("autonomous"), id: this.createTurnId("autonomous"),
}; };
this.activeForegroundQuery = this.query;
this.activeForegroundInput = this.input;
this.activeTurnHasAssistantText = false; this.activeTurnHasAssistantText = false;
this.contextUsage.beginTurn(); this.contextUsage.beginTurn();
this.notifySubscribers({ type: "turn_started", provider: "claude" }); this.notifySubscribers({ type: "turn_started", provider: "claude" });
@@ -3504,6 +3509,8 @@ class ClaudeAgentSession implements AgentSession {
} }
this.notifySubscribers({ type: "turn_completed", provider: "claude" }); this.notifySubscribers({ type: "turn_completed", provider: "claude" });
this.autonomousTurn = null; this.autonomousTurn = null;
this.activeForegroundQuery = null;
this.activeForegroundInput = null;
this.activeTurnHasAssistantText = false; this.activeTurnHasAssistantText = false;
this.syncTurnState("autonomous turn completed"); this.syncTurnState("autonomous turn completed");
} }