mirror of
https://github.com/getpaseo/paseo.git
synced 2026-10-24 10:12:44 +00:00
fix(plugin): restore ACP history replayed during session load (#7717)
ACP agents replay a loaded session's session/update notifications before session/load resolves, but AcpRuntime.open registered nativeSessionId only after the response, so the session filter dropped every replayed update and restored sessions opened with an empty timeline. Register the native id before the request, drain the notification lane once the response arrives so the replay completes before session.opened, and follow the open request's history contract: replayed updates rebuild the timeline under history "replay" and are dropped under history "skip". Replayed user_message_chunk updates now restore user messages through the chunk fallback scheme, so multi-chunk messages keep their boundaries; live user chunks stay dropped because admitPrompt already echoes the prompt. Co-authored-by: nistream <nistream@users.noreply.github.com>
This commit is contained in:
Ni Xiao
@@ -374,13 +374,19 @@ class AcpRuntime {
|
||||
/**
|
||||
* Fallback ids for streamed chunks that carry no `messageId`, one per
|
||||
* chunk kind. Consecutive chunks of a kind continue the same timeline
|
||||
* item; a chunk of the other kind, a tool call, a plan, a user message,
|
||||
* or a turn boundary ends it (#4699).
|
||||
* item; a chunk of another kind, a tool call, a plan, or a turn boundary
|
||||
* ends it (#4699).
|
||||
*/
|
||||
private readonly fallbackChunkIds = new Map<
|
||||
"agent_message_chunk" | "agent_thought_chunk",
|
||||
"agent_message_chunk" | "agent_thought_chunk" | "user_message_chunk",
|
||||
string
|
||||
>();
|
||||
/**
|
||||
* Active while a persisted session loads. ACP agents replay the loaded
|
||||
* session's updates before session/load resolves; the value follows the
|
||||
* open request's history contract.
|
||||
*/
|
||||
private replayWindow: "replay" | "skip" | null = null;
|
||||
private closing = false;
|
||||
private processFailed = false;
|
||||
private configTransaction = false;
|
||||
@@ -499,15 +505,23 @@ class AcpRuntime {
|
||||
};
|
||||
let response: Pick<NewSessionResponse, "modes" | "configOptions">;
|
||||
if (nativeSessionId) {
|
||||
response = await this.call(
|
||||
this.connection.loadSession({
|
||||
sessionId: nativeSessionId,
|
||||
cwd: input.config.cwd,
|
||||
mcpServers,
|
||||
_meta: metadata,
|
||||
}),
|
||||
);
|
||||
// ACP agents replay the loaded session's updates before session/load resolves,
|
||||
// so the native id must be registered first for them to pass the session filter.
|
||||
this.nativeSessionId = nativeSessionId;
|
||||
this.replayWindow = input.history === "replay" ? "replay" : "skip";
|
||||
try {
|
||||
response = await this.call(
|
||||
this.connection.loadSession({
|
||||
sessionId: nativeSessionId,
|
||||
cwd: input.config.cwd,
|
||||
mcpServers,
|
||||
_meta: metadata,
|
||||
}),
|
||||
);
|
||||
await this.drainNotifications();
|
||||
} finally {
|
||||
this.replayWindow = null;
|
||||
}
|
||||
} else {
|
||||
const newSession = await this.call(
|
||||
this.connection.newSession({
|
||||
@@ -898,6 +912,14 @@ class AcpRuntime {
|
||||
|
||||
private sessionUpdate(notification: SessionNotification): void {
|
||||
if (notification.sessionId !== this.nativeSessionId) return;
|
||||
if (this.replayWindow === "skip") return;
|
||||
if (
|
||||
this.replayWindow === "replay" &&
|
||||
notification.update.sessionUpdate === "user_message_chunk"
|
||||
) {
|
||||
this.reduceUserMessageChunk(notification.update);
|
||||
return;
|
||||
}
|
||||
this.reduceUpdate(notification.update);
|
||||
}
|
||||
|
||||
@@ -911,13 +933,14 @@ class AcpRuntime {
|
||||
this.fallbackChunkIds.clear();
|
||||
return;
|
||||
}
|
||||
// A chunk of one kind ends the other kind's item, whatever its content
|
||||
// or id: reasoning, then text, then reasoning again are three items.
|
||||
// A chunk of one kind ends every other kind's open item, whatever its
|
||||
// content or id: reasoning, then text, then reasoning again are three items.
|
||||
this.fallbackChunkIds.delete(
|
||||
update.sessionUpdate === "agent_message_chunk"
|
||||
? "agent_thought_chunk"
|
||||
: "agent_message_chunk",
|
||||
);
|
||||
this.fallbackChunkIds.delete("user_message_chunk");
|
||||
if (update.content.type !== "text") return;
|
||||
const id = this.resolveChunkId(update.sessionUpdate, update.messageId);
|
||||
const text = `${this.messages.get(id) ?? ""}${update.content.text}`;
|
||||
@@ -945,8 +968,31 @@ class AcpRuntime {
|
||||
this.reduceStateUpdate(update);
|
||||
}
|
||||
|
||||
// Historical user messages arrive only as replayed chunks; a live prompt
|
||||
// already echoes its own user message when it is admitted.
|
||||
private reduceUserMessageChunk(
|
||||
update: Extract<SessionUpdate, { sessionUpdate: "user_message_chunk" }>,
|
||||
): void {
|
||||
this.fallbackChunkIds.delete("agent_message_chunk");
|
||||
this.fallbackChunkIds.delete("agent_thought_chunk");
|
||||
if (update.content.type !== "text") return;
|
||||
const id = this.resolveChunkId("user_message_chunk", update.messageId);
|
||||
const text = `${this.messages.get(id) ?? ""}${update.content.text}`;
|
||||
this.messages.set(id, text);
|
||||
this.emit({
|
||||
type: "timeline.item",
|
||||
sessionId: this.options.boundarySessionId,
|
||||
item: {
|
||||
type: "user_message",
|
||||
id,
|
||||
text,
|
||||
...(update.messageId ? { messageId: update.messageId } : {}),
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
private resolveChunkId(
|
||||
kind: "agent_message_chunk" | "agent_thought_chunk",
|
||||
kind: "agent_message_chunk" | "agent_thought_chunk" | "user_message_chunk",
|
||||
messageId: string | null | undefined,
|
||||
): string {
|
||||
if (messageId) {
|
||||
|
||||
@@ -1500,3 +1500,187 @@ describe("ACP streamed message boundaries", () => {
|
||||
},
|
||||
);
|
||||
});
|
||||
|
||||
function userChunk(text: string, messageId?: string): SessionUpdate {
|
||||
return {
|
||||
sessionUpdate: "user_message_chunk",
|
||||
content: { type: "text", text },
|
||||
...(messageId === undefined ? {} : { messageId }),
|
||||
};
|
||||
}
|
||||
|
||||
async function openReplaySession(
|
||||
replay: SessionUpdate[],
|
||||
history: "replay" | "skip",
|
||||
): Promise<{ connection: ProviderConnection; events: ProviderEvent[] }> {
|
||||
const harness = connectorHarness({
|
||||
capabilities: { loadSession: true },
|
||||
handleRequest(instance, request) {
|
||||
if (request.method !== "session/load") return false;
|
||||
for (const update of replay) {
|
||||
instance.notify("session/update", { sessionId: "connector-session", update });
|
||||
}
|
||||
instance.respond(request, {
|
||||
sessionId: "connector-session",
|
||||
modes: null,
|
||||
configOptions: [],
|
||||
});
|
||||
return true;
|
||||
},
|
||||
});
|
||||
const registration = runAcpProvider({
|
||||
id: "replay-acp",
|
||||
label: "Replay ACP",
|
||||
connector: harness.connector,
|
||||
});
|
||||
const connection = await registration.connect({
|
||||
versions: [1],
|
||||
capabilities: ["prompt.message", "session.persistence"],
|
||||
});
|
||||
const events: ProviderEvent[] = [];
|
||||
connection.onEvent((event) => events.push(event));
|
||||
const input = openInput();
|
||||
input.persistence = { version: 1, data: { sessionId: "connector-session" } };
|
||||
input.history = history;
|
||||
await connection.send(input);
|
||||
await waitForEvent(events, (event) => event.type === "session.ready");
|
||||
return { connection, events };
|
||||
}
|
||||
|
||||
function timelineProjection(events: ProviderEvent[]): Array<Record<string, string>> {
|
||||
return events.flatMap((event) => {
|
||||
if (event.type !== "timeline.item") return [];
|
||||
const item = event.item;
|
||||
if (
|
||||
item.type === "user_message" ||
|
||||
item.type === "assistant_message" ||
|
||||
item.type === "reasoning"
|
||||
) {
|
||||
return [{ type: item.type, id: item.id, text: item.text }];
|
||||
}
|
||||
if (item.type === "tool_call") return [{ type: item.type, id: item.id, status: item.status }];
|
||||
return [{ type: item.type, id: item.id }];
|
||||
});
|
||||
}
|
||||
|
||||
describe("ACP session history replay", () => {
|
||||
it("restores the timeline replayed before session/load resolves", async () => {
|
||||
const { connection, events } = await openReplaySession(
|
||||
[
|
||||
userChunk("What is the ", "user-1"),
|
||||
userChunk("temperature?", "user-1"),
|
||||
textChunk("Checking the sensor", "agent_thought_chunk"),
|
||||
{
|
||||
sessionUpdate: "tool_call",
|
||||
toolCallId: "tool-1",
|
||||
title: "Read sensor",
|
||||
kind: "read",
|
||||
status: "in_progress",
|
||||
},
|
||||
{ sessionUpdate: "tool_call_update", toolCallId: "tool-1", status: "completed" },
|
||||
textChunk("It is 25°C"),
|
||||
],
|
||||
"replay",
|
||||
);
|
||||
try {
|
||||
expect(timelineProjection(events)).toEqual([
|
||||
{ type: "user_message", id: "user-1", text: "What is the " },
|
||||
{ type: "user_message", id: "user-1", text: "What is the temperature?" },
|
||||
{ type: "reasoning", id: "agent_thought_chunk:1", text: "Checking the sensor" },
|
||||
{ type: "tool_call", id: "tool-1", status: "running" },
|
||||
{ type: "tool_call", id: "tool-1", status: "completed" },
|
||||
{ type: "assistant_message", id: "agent_message_chunk:2", text: "It is 25°C" },
|
||||
]);
|
||||
const lastReplayed = events.findLastIndex((event) => event.type === "timeline.item");
|
||||
const opened = events.findIndex((event) => event.type === "session.opened");
|
||||
expect(lastReplayed).toBeLessThan(opened);
|
||||
} finally {
|
||||
await connection.close();
|
||||
}
|
||||
});
|
||||
|
||||
it("continues anonymous user chunks into one message and starts a new one after an agent reply", async () => {
|
||||
const { connection, events } = await openReplaySession(
|
||||
[
|
||||
userChunk("first "),
|
||||
userChunk("question"),
|
||||
textChunk("answer"),
|
||||
userChunk("second question"),
|
||||
],
|
||||
"replay",
|
||||
);
|
||||
try {
|
||||
expect(timelineProjection(events)).toEqual([
|
||||
{ type: "user_message", id: "user_message_chunk:1", text: "first " },
|
||||
{ type: "user_message", id: "user_message_chunk:1", text: "first question" },
|
||||
{ type: "assistant_message", id: "agent_message_chunk:2", text: "answer" },
|
||||
{ type: "user_message", id: "user_message_chunk:3", text: "second question" },
|
||||
]);
|
||||
} finally {
|
||||
await connection.close();
|
||||
}
|
||||
});
|
||||
|
||||
it("does not duplicate the live prompt echo when the agent echoes it as a user chunk", async () => {
|
||||
const harness = connectorHarness({
|
||||
capabilities: { loadSession: true },
|
||||
promptUpdates: (promptText) => [
|
||||
{ sessionUpdate: "user_message_chunk", content: { type: "text", text: promptText } },
|
||||
{ sessionUpdate: "agent_message_chunk", content: { type: "text", text: "Acknowledged" } },
|
||||
],
|
||||
});
|
||||
const registration = runAcpProvider({
|
||||
id: "echo-acp",
|
||||
label: "Echo ACP",
|
||||
connector: harness.connector,
|
||||
});
|
||||
const connection = await registration.connect({
|
||||
versions: [1],
|
||||
capabilities: ["prompt.message", "session.persistence"],
|
||||
});
|
||||
const events: ProviderEvent[] = [];
|
||||
connection.onEvent((event) => events.push(event));
|
||||
const input = openInput();
|
||||
input.persistence = { version: 1, data: { sessionId: "connector-session" } };
|
||||
input.history = "replay";
|
||||
await connection.send(input);
|
||||
await waitForEvent(events, (event) => event.type === "session.ready");
|
||||
|
||||
await connection.send({
|
||||
type: "session.prompt",
|
||||
sessionId: "session-1",
|
||||
prompt: {
|
||||
clientMessageId: "echo-1",
|
||||
delivery: "auto",
|
||||
input: { type: "message", content: [{ type: "text", text: "hello" }] },
|
||||
},
|
||||
});
|
||||
try {
|
||||
await waitForEvent(
|
||||
events,
|
||||
(event) =>
|
||||
event.type === "session.turn" &&
|
||||
event.turnId === "acp:echo-1" &&
|
||||
event.state === "completed",
|
||||
);
|
||||
expect(timelineProjection(events)).toEqual([
|
||||
{ type: "user_message", id: "echo-1", text: "hello" },
|
||||
{ type: "assistant_message", id: "agent_message_chunk:1", text: "Acknowledged" },
|
||||
]);
|
||||
} finally {
|
||||
await connection.close();
|
||||
}
|
||||
});
|
||||
|
||||
it("drops replayed updates when the open request asks to skip history", async () => {
|
||||
const { connection, events } = await openReplaySession(
|
||||
[userChunk("What is in the README?", "user-1"), textChunk("It describes the project.")],
|
||||
"skip",
|
||||
);
|
||||
try {
|
||||
expect(timelineProjection(events)).toEqual([]);
|
||||
} finally {
|
||||
await connection.close();
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user