jerry/claude-agent-acp
1
0
Fork 0
mirror of https://github.com/agentclientprotocol/claude-agent-acp.git synced 2026-10-27 04:30:59 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: Send message ids for assistant and user messages (#750)

Browse Source
This commit is contained in:
Ben Brandt
2026-06-05 12:53:21 +02:00
committed by GitHub
parent ec14211056
commit 18516a3088
2 changed files with 359 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

118
src/acp-agent.ts
View File

@@ -199,6 +199,24 @@ type Session = {
* `tool_result` time so a long-running session doesn't accumulate every
* tool call for its whole lifetime. */
toolUseCache: ToolUseCache;
/** Maps the ACP `messageId` we expose to clients (see `messageIdForGrouping`)
* to the SDK message uuid that the Agent SDK's rewind/resume APIs key on
* (`Query.rewindFiles` takes a user-message uuid; `resumeSessionAt` takes an
* `SDKAssistantMessage.uuid`). For assistant turns the two differ — the ACP
* id is the Anthropic API message id (`msg_…`), available at `message_start`
* so streamed chunks can carry it, while the uuid only arrives on the
* consolidated message — so a client can only ask to rewind/fork by the id it
* was given, and we need this table to translate it back.
*
* Populated as a byproduct of the message loop (the consolidated message
* carries both ids) and of `replaySessionHistory` on load, so no extra
* `getSessionMessages` read is needed at rewind time. Last-write-wins
* naturally yields the turn-boundary uuid when one `msg_…` spans several
* content-block messages.
*
* NOT READ YET — recorded now so the mapping exists if/when we wire up
* fork/rewind. */
messageIdToUuid: Map<string, string>;
};
/** Compute a stable fingerprint of the session-defining params so we can
@@ -821,6 +839,17 @@ export class ClaudeAcpAgent implements Agent {
// `compacting` status sets it again, so every distinct compaction (e.g.
// repeated auto-compactions in a long turn) is still shown.
let compactionInProgress = false;
// Holds the Anthropic API message id of the assistant message currently
// being streamed, captured from `message_start` so every streamed chunk can
// be tagged with it. We use the API message id rather than the
// per-`stream_event` uuid because the same id is also present on the
// consolidated assistant message and in the persisted transcript — so a turn
// keeps the same ACP `messageId` whether it is streamed live or replayed
// from history. The per-event uuid is unique per event and never persisted.
// A single value suffices because every streaming partial arrives with
// `parent_tool_use_id === null` (subagent work is folded into tool-result
// messages, never surfaced as partial streams).
let currentStreamMessageId: string | undefined;
const userMessage = promptToClaude(params);
@@ -1247,6 +1276,12 @@ export class ClaudeAcpAgent implements Agent {
break;
}
case "stream_event": {
// `message_start` carries the Anthropic API message id; capture it
// so the streamed chunks that follow (whose delta events don't carry
// it) can all be tagged with the same, replay-stable id.
if (message.event.type === "message_start") {
currentStreamMessageId = message.event.message.id || undefined;
}
if (
message.parent_tool_use_id === null &&
(message.event.type === "message_start" || message.event.type === "message_delta")
@@ -1308,6 +1343,7 @@ export class ClaudeAcpAgent implements Agent {
clientCapabilities: this.clientCapabilities,
cwd: session.cwd,
taskState: session.taskState,
messageId: currentStreamMessageId,
},
)) {
await this.client.sessionUpdate(notification);
@@ -1320,6 +1356,15 @@ export class ClaudeAcpAgent implements Agent {
break;
}
// Record the ACP messageId -> SDK uuid mapping for this message. The
// consolidated message carries both ids, so this is where we learn
// the uuid that the SDK's rewind/resume APIs key on for the id we
// hand clients. Not read yet (see Session.messageIdToUuid).
const mappedMessageId = messageIdForGrouping(message);
if (mappedMessageId && typeof message.uuid === "string" && message.uuid.length > 0) {
session.messageIdToUuid.set(mappedMessageId, message.uuid);
}
// Check for prompt replay
if (message.type === "user" && "uuid" in message && message.uuid) {
if (message.uuid === promptUuid) {
@@ -1383,6 +1428,7 @@ export class ClaudeAcpAgent implements Agent {
parentToolUseId: message.parent_tool_use_id,
cwd: session.cwd,
taskState: session.taskState,
messageId: messageIdForGrouping(message),
},
)) {
await this.client.sessionUpdate(notification);
@@ -1445,6 +1491,7 @@ export class ClaudeAcpAgent implements Agent {
parentToolUseId: message.parent_tool_use_id,
cwd: session.cwd,
taskState: session.taskState,
messageId: messageIdForGrouping(message),
},
)) {
await this.client.sessionUpdate(notification);
@@ -1786,6 +1833,16 @@ export class ClaudeAcpAgent implements Agent {
const messages = await getSessionMessages(sessionId);
for (const message of messages) {
// Backfill the ACP messageId -> SDK uuid mapping for messages we didn't
// observe live (resumed/loaded sessions), so rewind/resume can translate
// a client-supplied id without an extra getSessionMessages read. Not read
// yet (see Session.messageIdToUuid).
const replayMessageId = messageIdForGrouping(message);
const replaySession = this.sessions[sessionId];
if (replaySession && replayMessageId && message.uuid) {
replaySession.messageIdToUuid.set(replayMessageId, message.uuid);
}
// @ts-expect-error - untyped in SDK but we handle all of these
let content: unknown = message.message.content;
// @ts-expect-error - untyped in SDK but we handle all of these
@@ -1808,6 +1865,7 @@ export class ClaudeAcpAgent implements Agent {
clientCapabilities: this.clientCapabilities,
cwd: this.sessions[sessionId]?.cwd,
taskState: this.sessions[sessionId]?.taskState,
messageId: replayMessageId,
},
)) {
await this.client.sessionUpdate(notification);
@@ -2514,6 +2572,7 @@ export class ClaudeAcpAgent implements Agent {
inferContextWindowFromModel(models.currentModelId) ?? DEFAULT_CONTEXT_WINDOW,
taskState,
toolUseCache: {},
messageIdToUuid: new Map(),
};
return {
@@ -3108,6 +3167,54 @@ export function promptToClaude(prompt: PromptRequest): SDKUserMessage {
};
}
/**
* Resolves the ACP `messageId` for a Claude SDK message (live) or a persisted
* transcript message (replay) so chunk grouping is identical in both views.
*
* Assistant turns are keyed by the Anthropic API message id (`message.id`),
* which is identical at `message_start`, on the consolidated assistant message,
* and in the persisted transcript — unlike the per-`stream_event` uuid, which is
* unique per event and never persisted. User messages have no API id, but they
* are never streamed, so their (stable) SDK uuid is used instead. ACP message
* ids are opaque strings, so no particular format is required.
*/
export function messageIdForGrouping(message: {
type?: string;
uuid?: string | null;
message?: unknown;
}): string | undefined {
if (message.type === "assistant") {
const inner = message.message;
const apiId =
inner && typeof inner === "object" && "id" in inner
? (inner as { id?: unknown }).id
: undefined;
if (typeof apiId === "string" && apiId.length > 0) {
return apiId;
}
}
return typeof message.uuid === "string" && message.uuid.length > 0 ? message.uuid : undefined;
}
/**
* Stamps an ACP `messageId` onto a session update, but only on the message/
* thought chunk variants that carry one — tool_call/plan/etc. updates never do.
* No-op when `messageId` is falsy, so callers can pass it through unconditionally.
*/
function applyMessageId(
update: SessionNotification["update"],
messageId: string | undefined,
): void {
if (
messageId &&
(update.sessionUpdate === "agent_message_chunk" ||
update.sessionUpdate === "user_message_chunk" ||
update.sessionUpdate === "agent_thought_chunk")
) {
update.messageId = messageId;
}
}
/**
* Convert an SDKAssistantMessage (Claude) to a SessionNotification (ACP).
* Only handles text, image, and thinking chunks for now.
@@ -3125,6 +3232,12 @@ export function toAcpNotifications(
parentToolUseId?: string | null;
cwd?: string;
taskState?: TaskState;
// Opaque id identifying the message these chunks belong to (ACP message ids
// are opaque strings — no particular format is required). Attached to
// user/agent message and thought chunks so clients can group streamed chunks
// into a single message. Omit it (leave undefined) when unknown — never send
// an explicit `null`.
messageId?: string;
},
): SessionNotification[] {
const taskState = options?.taskState ?? new Map();
@@ -3138,6 +3251,7 @@ export function toAcpNotifications(
text: content,
},
};
applyMessageId(update, options?.messageId);
if (options?.parentToolUseId) {
update._meta = {
@@ -3419,6 +3533,7 @@ export function toAcpNotifications(
},
};
}
applyMessageId(update, options?.messageId);
output.push({ sessionId, update });
}
}
@@ -3436,6 +3551,7 @@ export function streamEventToAcpNotifications(
clientCapabilities?: ClientCapabilities;
cwd?: string;
taskState?: TaskState;
messageId?: string;
},
): SessionNotification[] {
const event = message.event;
@@ -3453,6 +3569,7 @@ export function streamEventToAcpNotifications(
parentToolUseId: message.parent_tool_use_id,
cwd: options?.cwd,
taskState: options?.taskState,
messageId: options?.messageId,
},
);
case "content_block_delta":
@@ -3468,6 +3585,7 @@ export function streamEventToAcpNotifications(
parentToolUseId: message.parent_tool_use_id,
cwd: options?.cwd,
taskState: options?.taskState,
messageId: options?.messageId,
},
);
// No content. `ping` is a Messages-API keep-alive event that the SDK's

242
src/tests/acp-agent.test.ts
View File

@@ -33,10 +33,16 @@ import {
claudeCliPath,
describeAlwaysAllow,
streamEventToAcpNotifications,
messageIdForGrouping,
type SDKMessageFilter,
} from "../acp-agent.js";
import { Pushable } from "../utils.js";
import { deleteSession, query, SDKAssistantMessage } from "@anthropic-ai/claude-agent-sdk";
import {
deleteSession,
getSessionMessages,
query,
SDKAssistantMessage,
} from "@anthropic-ai/claude-agent-sdk";
import { randomUUID } from "crypto";
vi.mock("@anthropic-ai/claude-agent-sdk", async (importOriginal) => {
@@ -1318,6 +1324,67 @@ describe.skipIf(!process.env.RUN_INTEGRATION_TESTS)("SDK behavior", () => {
const { value } = await q.next();
expect(value).toMatchObject({ type: "system", session_id: sessionId });
}, 10000);
// Pins the SDK invariant our `messageId` plumbing relies on: the Anthropic
// API message id is available at `message_start` (before any delta), is the
// same on the consolidated assistant message, and is recoverable from the
// persisted transcript — so a turn keeps one stable id across streaming and
// replay. The per-`stream_event` uuid is NOT used because it is unique per
// event and never persisted; this test would fail if a future SDK regressed
// any of those properties.
it("uses the API message id as a stable anchor across streaming and replay", async () => {
const sessionId = randomUUID();
const q = query({
prompt: "Reply with exactly these words and nothing else: hello there my friend",
options: {
systemPrompt: { type: "preset", preset: "claude_code" },
sessionId,
includePartialMessages: true,
maxTurns: 1,
allowedTools: [],
},
});
let messageStartApiId: string | undefined;
let consolidatedApiId: string | undefined;
let sawDelta = false;
let allPartialsTopLevel = true;
for await (const message of q) {
if (message.type === "assistant") {
consolidatedApiId = message.message.id;
}
if (message.type !== "stream_event") continue;
// Every streaming partial must belong to the top-level agent
// (parent_tool_use_id === null). Subagent work is folded into tool-result
// messages rather than surfaced as partial streams, which is what lets us
// track a single anchor without keying by parent_tool_use_id.
if (message.parent_tool_use_id !== null) allPartialsTopLevel = false;
if (message.event.type === "message_start") {
messageStartApiId = message.event.message.id;
} else if (message.event.type === "content_block_delta") {
sawDelta = true;
}
}
// The API message id is present at message_start (before deltas), so we can
// tag every streamed chunk with it, and it is identical on the consolidated
// assistant message.
expect(messageStartApiId).toBeTruthy();
expect(sawDelta).toBe(true);
expect(allPartialsTopLevel).toBe(true);
expect(consolidatedApiId).toBe(messageStartApiId);
// ...and the SAME id is recoverable from the persisted transcript, so chunks
// grouped live keep their id when the session is replayed.
const persisted = await getSessionMessages(sessionId);
const replayedAssistant = persisted.find((m) => m.type === "assistant");
expect(replayedAssistant).toBeDefined();
expect((replayedAssistant!.message as { id?: string }).id).toBe(messageStartApiId);
// The helper used in production must derive that same id from the replayed
// message.
expect(messageIdForGrouping(replayedAssistant!)).toBe(messageStartApiId);
}, 30000);
});
describe("permission requests", () => {
@@ -1549,6 +1616,7 @@ describe("stop reason propagation", () => {
contextWindowSize: 200000,
taskState: new Map(),
toolUseCache: {},
messageIdToUuid: new Map(),
};
}
@@ -1694,6 +1762,7 @@ describe("stop reason propagation", () => {
contextWindowSize: 200000,
taskState: new Map(),
toolUseCache: {},
messageIdToUuid: new Map(),
};
const response = await agent.prompt({
@@ -1854,6 +1923,7 @@ describe("session/close", () => {
contextWindowSize: 200000,
taskState: new Map(),
toolUseCache: {},
messageIdToUuid: new Map(),
};
return agent.sessions[sessionId]!;
}
@@ -1939,6 +2009,7 @@ describe("session/delete", () => {
contextWindowSize: 200000,
taskState: new Map(),
toolUseCache: {},
messageIdToUuid: new Map(),
};
return agent.sessions[sessionId]!;
}
@@ -2041,6 +2112,7 @@ describe("getOrCreateSession param change detection", () => {
contextWindowSize: 200000,
taskState: new Map(),
toolUseCache: {},
messageIdToUuid: new Map(),
};
return agent.sessions[sessionId]!;
}
@@ -2277,6 +2349,7 @@ describe("usage_update computation", () => {
contextWindowSize: 200000,
taskState: new Map(),
toolUseCache: {},
messageIdToUuid: new Map(),
};
}
@@ -3221,6 +3294,7 @@ describe("emitRawSDKMessages", () => {
contextWindowSize: 200000,
taskState: new Map(),
toolUseCache: {},
messageIdToUuid: new Map(),
};
}
@@ -3450,6 +3524,7 @@ describe("result origin handling", () => {
contextWindowSize: 200000,
taskState: new Map(),
toolUseCache: {},
messageIdToUuid: new Map(),
};
}
@@ -3626,6 +3701,7 @@ describe("memory_recall handling", () => {
contextWindowSize: 200000,
taskState: new Map(),
toolUseCache: {},
messageIdToUuid: new Map(),
};
}
@@ -3857,6 +3933,7 @@ describe("post-error recovery", () => {
contextWindowSize: 200000,
taskState: new Map(),
toolUseCache: {},
messageIdToUuid: new Map(),
};
return { interrupt };
}
@@ -4003,6 +4080,7 @@ describe("session/cancel wedge recovery (issue #680)", () => {
contextWindowSize: 200000,
taskState: new Map(),
toolUseCache: {},
messageIdToUuid: new Map(),
};
return { interrupt };
}
@@ -4142,4 +4220,166 @@ describe("streamEventToAcpNotifications", () => {
expect(result).toEqual([]);
expect(errors).toEqual([]);
});
it("attaches the supplied messageId to streamed text chunks", () => {
const messageId = randomUUID();
const message = {
type: "stream_event",
parent_tool_use_id: null,
uuid: randomUUID(),
session_id: "test-session",
event: {
type: "content_block_delta",
index: 0,
delta: { type: "text_delta", text: "hello" },
},
} as Parameters<typeof streamEventToAcpNotifications>[0];
const result = streamEventToAcpNotifications(
message,
"test",
{},
{} as AgentSideConnection,
console,
{ messageId },
);
expect(result).toEqual([
{
sessionId: "test",
update: {
sessionUpdate: "agent_message_chunk",
content: { type: "text", text: "hello" },
messageId,
},
},
]);
});
});
describe("toAcpNotifications messageId", () => {
const messageId = "11111111-2222-3333-4444-555555555555";
it("sets messageId on agent message chunks from string content", () => {
const result = toAcpNotifications(
"hello world",
"assistant",
"test",
{},
{} as AgentSideConnection,
console,
{ messageId },
);
expect(result).toEqual([
{
sessionId: "test",
update: {
sessionUpdate: "agent_message_chunk",
content: { type: "text", text: "hello world" },
messageId,
},
},
]);
});
it("sets messageId on user message chunks and thought chunks", () => {
const userResult = toAcpNotifications(
[{ type: "text", text: "hi" }],
"user",
"test",
{},
{} as AgentSideConnection,
console,
{ messageId },
);
expect(userResult[0].update).toMatchObject({
sessionUpdate: "user_message_chunk",
messageId,
});
const thoughtResult = toAcpNotifications(
[{ type: "thinking", thinking: "hmm", signature: "" }],
"assistant",
"test",
{},
{} as AgentSideConnection,
console,
{ messageId },
);
expect(thoughtResult[0].update).toMatchObject({
sessionUpdate: "agent_thought_chunk",
messageId,
});
});
it("omits messageId when none is supplied", () => {
const result = toAcpNotifications(
"hello",
"assistant",
"test",
{},
{} as AgentSideConnection,
console,
);
expect(result[0].update).not.toHaveProperty("messageId");
});
it("never sets messageId on non-chunk updates (tool_call)", () => {
const result = toAcpNotifications(
[
{
type: "tool_use",
id: "toolu_abc",
name: "Read",
input: { file_path: "/tmp/x" },
},
],
"assistant",
"test",
{},
{} as AgentSideConnection,
console,
{ messageId, registerHooks: false },
);
expect(result[0].update.sessionUpdate).toBe("tool_call");
expect(result[0].update).not.toHaveProperty("messageId");
});
});
describe("messageIdForGrouping", () => {
it("uses the Anthropic API message id for assistant messages", () => {
const message = {
type: "assistant",
uuid: "de242400-cdb3-4af7-9856-d3b114b20af9",
message: { id: "msg_018DQGVuZbGYwVnvDakAP9Do", role: "assistant" },
};
// The API id is identical at message_start, on the consolidated message,
// and in the persisted transcript — so it stays stable across replay,
// unlike the per-message uuid.
expect(messageIdForGrouping(message)).toBe("msg_018DQGVuZbGYwVnvDakAP9Do");
});
it("falls back to the uuid for assistant messages without an API id", () => {
const message = {
type: "assistant",
uuid: "de242400-cdb3-4af7-9856-d3b114b20af9",
message: { role: "assistant" },
};
expect(messageIdForGrouping(message)).toBe("de242400-cdb3-4af7-9856-d3b114b20af9");
});
it("uses the uuid for user messages (they carry no API id and aren't streamed)", () => {
const message = {
type: "user",
uuid: "11111111-2222-3333-4444-555555555555",
message: { id: "msg_should_be_ignored", role: "user" },
};
expect(messageIdForGrouping(message)).toBe("11111111-2222-3333-4444-555555555555");
});
it("returns undefined when there is no usable id", () => {
expect(messageIdForGrouping({ type: "system", message: {} })).toBeUndefined();
expect(messageIdForGrouping({ type: "assistant", uuid: "", message: {} })).toBeUndefined();
});
});