puter/paseo
1
0
Fork 0
mirror of https://github.com/getpaseo/paseo.git synced 2026-07-29 12:01:31 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(acp): keep foreground agents running (#2148)

Browse Source 
ACP setup and out-of-prompt notifications do not define a turn lifecycle. Treating them as autonomous turns could complete the active run while its prompt was still streaming.
This commit is contained in:
Mohamed Boudra
2026-07-16 20:49:00 +02:00
committed by GitHub
parent a1cd50c2ae
commit d5baf1a7e6
2 changed files with 37 additions and 105 deletions
Download Patch File Download Diff File Expand all files Collapse all files

86
packages/server/src/server/agent/providers/acp-agent.test.ts
View File

@@ -2208,8 +2208,7 @@ 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 });
test("keeps ACP configuration notifications outside the turn lifecycle", async () => {
const session = createSession();
asInternals<ACPSessionInternals>(session).sessionId = "session-1";
@@ -2221,49 +2220,26 @@ describe("ACPAgentSession", () => {
await session.sessionUpdate({
sessionId: "session-1",
update: {
sessionUpdate: "agent_message_chunk",
messageId: "autonomous-msg",
content: { type: "text", text: "Autonomous update" },
sessionUpdate: "config_option_update",
configOptions: [
selectConfigOption("mode", ["plan", "yolo"], "yolo"),
selectConfigOption("model", ["kimi-code/kimi-for-coding"]),
selectConfigOption("thought_level", ["off", "on"], "on"),
],
} 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();
expect(events.map((event) => event.type)).toEqual([
"thread_started",
"mode_changed",
"model_changed",
"thinking_option_changed",
]);
});
test("completes an existing autonomous turn before starting a foreground turn", async () => {
vi.useFakeTimers({ shouldAdvanceTime: true });
test("forwards out-of-prompt ACP content without inventing a turn", async () => {
const session = createSession();
asInternals<ACPSessionInternals>(session).sessionId = "session-1";
asInternals<ACPSessionInternals>(session).connection = {
prompt: vi.fn(() => new Promise(() => {})),
};
const events: AgentStreamEvent[] = [];
session.subscribe((event) => {
@@ -2274,25 +2250,27 @@ describe("ACPAgentSession", () => {
sessionId: "session-1",
update: {
sessionUpdate: "agent_message_chunk",
messageId: "autonomous-msg",
content: { type: "text", text: "Autonomous update" },
messageId: "unscoped-message",
content: { type: "text", text: "Unscoped ACP 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();
expect(events).toEqual([
{
type: "thread_started",
provider: "claude-acp",
sessionId: "session-1",
},
{
type: "timeline",
provider: "claude-acp",
item: {
type: "assistant_message",
text: "Unscoped ACP update",
messageId: "unscoped-message",
},
},
]);
});
test("startTurn returns before the ACP prompt settles and completes later via subscribers", async () => {

56
packages/server/src/server/agent/providers/acp-agent.ts
View File

@@ -1331,9 +1331,6 @@ 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<typeof setTimeout> | null = null;
private static readonly AUTONOMOUS_TURN_TIMEOUT_MS = 30_000;
private fallbackAssistantMessageId: string | null = null;
private closed = false;
private historyPending = false;
@@ -1475,7 +1472,6 @@ 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();
@@ -2119,7 +2115,7 @@ export class ACPAgentSession implements AgentSession, ACPClient {
agentId: this.agentId,
provider: this.provider,
sessionId: this.sessionId,
turnId: this.activeForegroundTurnId ?? this.autonomousTurnId ?? undefined,
turnId: this.activeForegroundTurnId ?? undefined,
rawEvent: params,
events,
},
@@ -2134,11 +2130,6 @@ 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);
}
@@ -2672,25 +2663,23 @@ export class ACPAgentSession implements AgentSession, ACPClient {
type: "timeline",
provider: this.provider,
item,
turnId: this.activeForegroundTurnId ?? this.autonomousTurnId ?? undefined,
turnId: this.activeForegroundTurnId ?? 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(tagged) ?? turnId ?? undefined,
event: tagged,
turnId: getAgentStreamEventTurnId(event) ?? this.activeForegroundTurnId ?? undefined,
event,
},
"provider.acp.event_emit",
);
for (const subscriber of this.subscribers) {
subscriber(tagged);
subscriber(event);
}
}
@@ -2738,41 +2727,6 @@ 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<AgentTimelineItem, { type: "user_message" }>,
): boolean {