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 546ff8c80..828b1b4de 100644 --- a/packages/server/src/server/agent/providers/acp-agent.test.ts +++ b/packages/server/src/server/agent/providers/acp-agent.test.ts @@ -2208,6 +2208,93 @@ describe("ACPAgentSession", () => { expect(assistantMessages[2].messageId).not.toBe(assistantMessages[0].messageId); }); + test("starts an autonomous turn for spontaneous session updates outside a foreground turn", async () => { + vi.useFakeTimers({ shouldAdvanceTime: true }); + const session = createSession(); + asInternals(session).sessionId = "session-1"; + + const events: AgentStreamEvent[] = []; + session.subscribe((event) => { + events.push(event); + }); + + await session.sessionUpdate({ + sessionId: "session-1", + update: { + sessionUpdate: "agent_message_chunk", + messageId: "autonomous-msg", + content: { type: "text", text: "Autonomous update" }, + } as SessionUpdate, + }); + + // Should emit turn_started before the timeline item. + const turnStartedIndex = events.findIndex((e) => e.type === "turn_started"); + expect(turnStartedIndex).toBeGreaterThanOrEqual(0); + const timelineIndex = events.findIndex( + (e) => + e.type === "timeline" && + e.item.type === "assistant_message" && + e.item.text === "Autonomous update", + ); + expect(timelineIndex).toBeGreaterThan(turnStartedIndex); + + const turnStarted = events[turnStartedIndex]; + expect(turnStarted.type).toBe("turn_started"); + const autonomousTurnId = (turnStarted as { turnId?: string }).turnId; + expect(autonomousTurnId).toEqual(expect.any(String)); + + // Timeline item should be tagged with the autonomous turn id. + const timelineEvent = events[timelineIndex]; + expect(timelineEvent.type).toBe("timeline"); + expect((timelineEvent as { turnId?: string }).turnId).toBe(autonomousTurnId); + + // Advance timers to complete the autonomous turn. + await vi.advanceTimersByTimeAsync(ACPAgentSession["AUTONOMOUS_TURN_TIMEOUT_MS"] + 10); + const turnCompleted = events.find((e) => e.type === "turn_completed"); + expect(turnCompleted).toBeDefined(); + expect((turnCompleted as { turnId?: string }).turnId).toBe(autonomousTurnId); + + vi.useRealTimers(); + }); + + test("completes an existing autonomous turn before starting a foreground turn", async () => { + vi.useFakeTimers({ shouldAdvanceTime: true }); + const session = createSession(); + asInternals(session).sessionId = "session-1"; + asInternals(session).connection = { + prompt: vi.fn(() => new Promise(() => {})), + }; + + const events: AgentStreamEvent[] = []; + session.subscribe((event) => { + events.push(event); + }); + + await session.sessionUpdate({ + sessionId: "session-1", + update: { + sessionUpdate: "agent_message_chunk", + messageId: "autonomous-msg", + content: { type: "text", text: "Autonomous update" }, + } as SessionUpdate, + }); + + const turnStarted = events.find((e) => e.type === "turn_started"); + expect(turnStarted).toBeDefined(); + const autonomousTurnId = (turnStarted as { turnId?: string }).turnId; + + // Starting a foreground turn should complete the autonomous turn first. + void session.startTurn("user prompt"); + await vi.runOnlyPendingTimersAsync(); + + const turnCompleted = events.find( + (e) => e.type === "turn_completed" && (e as { turnId?: string }).turnId === autonomousTurnId, + ); + expect(turnCompleted).toBeDefined(); + + vi.useRealTimers(); + }); + test("startTurn returns before the ACP prompt settles and completes later via subscribers", async () => { const session = createSession(); const events: Array<{ type: string; turnId?: string }> = []; diff --git a/packages/server/src/server/agent/providers/acp-agent.ts b/packages/server/src/server/agent/providers/acp-agent.ts index 6da37dad6..51e24c669 100644 --- a/packages/server/src/server/agent/providers/acp-agent.ts +++ b/packages/server/src/server/agent/providers/acp-agent.ts @@ -1331,6 +1331,9 @@ export class ACPAgentSession implements AgentSession, ACPClient { private readonly extensionCommandsParser?: ACPExtensionCommandsParser; private currentTurnUsage: AgentUsage | undefined; private activeForegroundTurnId: string | null = null; + private autonomousTurnId: string | null = null; + private autonomousTurnTimer: ReturnType | null = null; + private static readonly AUTONOMOUS_TURN_TIMEOUT_MS = 30_000; private fallbackAssistantMessageId: string | null = null; private closed = false; private historyPending = false; @@ -1472,6 +1475,7 @@ export class ACPAgentSession implements AgentSession, ACPClient { if (this.activeForegroundTurnId) { throw new Error("A foreground turn is already active"); } + this.completeAutonomousTurn(); const turnId = randomUUID(); const messageId = options?.messageId ?? randomUUID(); @@ -2115,7 +2119,7 @@ export class ACPAgentSession implements AgentSession, ACPClient { agentId: this.agentId, provider: this.provider, sessionId: this.sessionId, - turnId: this.activeForegroundTurnId ?? undefined, + turnId: this.activeForegroundTurnId ?? this.autonomousTurnId ?? undefined, rawEvent: params, events, }, @@ -2130,6 +2134,11 @@ export class ACPAgentSession implements AgentSession, ACPClient { return; } + if (events.length > 0 && !this.activeForegroundTurnId) { + this.startAutonomousTurn(); + this.resetAutonomousTurnTimer(); + } + for (const event of events) { this.pushEvent(event); } @@ -2663,23 +2672,25 @@ export class ACPAgentSession implements AgentSession, ACPClient { type: "timeline", provider: this.provider, item, - turnId: this.activeForegroundTurnId ?? undefined, + turnId: this.activeForegroundTurnId ?? this.autonomousTurnId ?? undefined, }; } private pushEvent(event: AgentStreamEvent): void { + const turnId = this.activeForegroundTurnId ?? this.autonomousTurnId; + const tagged = event.type === "timeline" && turnId ? { ...event, turnId } : event; this.logger.trace( { agentId: this.agentId, provider: this.provider, sessionId: this.sessionId, - turnId: getAgentStreamEventTurnId(event) ?? this.activeForegroundTurnId ?? undefined, - event, + turnId: getAgentStreamEventTurnId(tagged) ?? turnId ?? undefined, + event: tagged, }, "provider.acp.event_emit", ); for (const subscriber of this.subscribers) { - subscriber(event); + subscriber(tagged); } } @@ -2727,6 +2738,41 @@ export class ACPAgentSession implements AgentSession, ACPClient { this.pushEvent(event); } + private startAutonomousTurn(): void { + if (this.autonomousTurnId) { + return; + } + this.autonomousTurnId = randomUUID(); + this.pushEvent({ + type: "turn_started", + provider: this.provider, + turnId: this.autonomousTurnId, + }); + } + + private completeAutonomousTurn(): void { + if (!this.autonomousTurnId) { + return; + } + if (this.autonomousTurnTimer) { + clearTimeout(this.autonomousTurnTimer); + this.autonomousTurnTimer = null; + } + const turnId = this.autonomousTurnId; + this.autonomousTurnId = null; + this.pushEvent({ type: "turn_completed", provider: this.provider, turnId }); + } + + private resetAutonomousTurnTimer(): void { + if (this.autonomousTurnTimer) { + clearTimeout(this.autonomousTurnTimer); + } + this.autonomousTurnTimer = setTimeout(() => { + this.completeAutonomousTurn(); + }, ACPAgentSession.AUTONOMOUS_TURN_TIMEOUT_MS); + this.autonomousTurnTimer.unref?.(); + } + private isSubmittedUserMessageEcho( item: Extract, ): boolean {