mirror of
https://github.com/agentclientprotocol/claude-agent-acp.git
synced 2026-10-27 04:30:59 +00:00
refactor: report a turn inserted when Claude Code starts it (#1227)
Part of opt-in ACP v2 support. It fixes the first issue the review of #1226 found in the agent's turn reporting. It's independent of #1226 and changes nothing for ACP v1. ## Problem A turn reported `inserted` (#1222) at activation: the prompt's echo, or for an echo-less command, its result. The echo can come after the turn's first output; the agent already handles deltas that stream before it (see `injectSessionEchoAt` in the tests). With #1226's v2 mapping, that means a v2 client sees: - a turn's output before its `user_message` and before `running`; - `/compact`'s `session/prompt` answered only after the whole command. ## Change On CLIs that report command lifecycles (`msg_lifecycle_v1`), the `command_lifecycle` "started" frame now reports the turn inserted. **Why that frame is early enough.** I read the order in the CLI bundled with SDK 0.3.287 rather than in a live capture: - **A command that starts a fresh turn:** the engine pushes "started" into its lifecycle buffer at turn start (`C.push(W(w.uuid,"started"))`, before `[engine] turn N start`). The turn's message loop then flushes that buffer before it yields anything the turn produced (`if(C.length>0)yield*te()`), including its echo and first `stream_event`. - **A prompt folded into a running turn:** its `queued_command` attachment (the echo) is yielded before "started", so activation at the echo already comes first and this frame changes nothing. **One guard.** While another turn is still active, insertion still waits for the echo. That covers a turn held for its background work, a steered turn, and a cancelled turn awaiting its trailing idle. The echo's hand-off ends that turn first, so a turn is never reported inserted before the previous one ends. **Older CLIs** without lifecycle frames keep insertion at activation. Only the time `inserted` is reported changes. Activation, output attribution and settlement don't, and ACP v1 ignores `inserted`. ## Testing `src/tests/turn-events.test.ts`: - a fresh turn whose answer streams between "started" and its echo is reported inserted before that output; - a prompt that starts while the turn before it is held is reported inserted only after that turn ended. Removing the new insertion point fails the first test, and removing the guard fails the second. The rest of the suite passes unchanged, including the recorded v1 scenario traffic. `npm run build`, `npm run check`, `npm run test:run` (2231 passed).
This commit is contained in:
Ben Brandt
@@ -4186,19 +4186,24 @@ export class ClaudeAcpAgent {
|
||||
}
|
||||
}
|
||||
resetTurnScratch();
|
||||
// Activation is when Claude Code takes the prompt in: its echo, or for an
|
||||
// echo-less command, its result. That inserts it into the conversation.
|
||||
// A turn can be settled by now: an echo hand-off awaits settling the
|
||||
// previous turn, and a cancel in that window settles the queued one.
|
||||
if (!turn.insertedReported && !turn.settled) {
|
||||
turn.insertedReported = true;
|
||||
turn.events.inserted(turn.promptUuid);
|
||||
}
|
||||
// Without a command_lifecycle "started" frame before it (see that
|
||||
// handler), activation is the first sign that Claude Code took the
|
||||
// prompt in: its echo, or for an echo-less command, its result.
|
||||
reportInserted(turn);
|
||||
// A request can already be open: Claude Code asks for a queued prompt
|
||||
// before the consumer reaches its echo.
|
||||
this.syncAwaitingUser(session);
|
||||
};
|
||||
|
||||
/** Report that Claude Code took the turn's prompt in, once. A turn can be
|
||||
* settled by now: an echo hand-off awaits settling the previous turn, and
|
||||
* a cancel in that window settles the queued one. */
|
||||
const reportInserted = (turn: Turn) => {
|
||||
if (turn.insertedReported || turn.settled) return;
|
||||
turn.insertedReported = true;
|
||||
turn.events.inserted(turn.promptUuid);
|
||||
};
|
||||
|
||||
/** Ensure there is an active turn before a user-turn result that carries no
|
||||
* echo to activate it, by promoting the queue head. Most turns are
|
||||
* activated by their replayed user message before their result, but some
|
||||
@@ -4856,10 +4861,10 @@ export class ClaudeAcpAgent {
|
||||
// prompt() stamps a uuid on every message. The frame is @internal and
|
||||
// absent from the SDKMessage union, so handle it BEFORE the exhaustive
|
||||
// switch: it must not reach `unreachable`'s error log, and a `case`
|
||||
// for it wouldn't typecheck. It feeds only the orphan accounting (see
|
||||
// Session.orphanCommands); turn settlement stays driven by
|
||||
// echoes/results/idle. (Raw-mode emission above still forwards these
|
||||
// frames.)
|
||||
// for it wouldn't typecheck. It feeds the orphan accounting (see
|
||||
// Session.orphanCommands) and reports a turn inserted at "started";
|
||||
// turn activation and settlement stay driven by echoes/results/idle.
|
||||
// (Raw-mode emission above still forwards these frames.)
|
||||
if ((message as { type: string }).type === "command_lifecycle") {
|
||||
const frame = message as unknown as { command_uuid: string; state: string };
|
||||
switch (frame.state) {
|
||||
@@ -4870,6 +4875,16 @@ export class ClaudeAcpAgent {
|
||||
if (queued) {
|
||||
queued.commandStarted = true;
|
||||
compaction.resume();
|
||||
// Claude Code took the prompt in: it drained into a turn. A
|
||||
// fresh turn reports that before anything it produces, its echo
|
||||
// and first stream events included, so this is the earliest
|
||||
// insertion point. (A prompt folded into a running turn reports
|
||||
// it after its echo, which already activated it.) While another
|
||||
// turn is still active (held for background work, steered, or
|
||||
// cancelled and awaiting its trailing idle), insertion waits
|
||||
// for this turn's echo, whose hand-off ends that turn first.
|
||||
const active = session.activeTurn;
|
||||
if (!active || active.settled || active === queued) reportInserted(queued);
|
||||
}
|
||||
// ...and promote an already-orphaned command: once dispatched,
|
||||
// a bare `cancelled` no longer means "dropped without running".
|
||||
|
||||
@@ -68,31 +68,45 @@ function result(sessionId: string) {
|
||||
};
|
||||
}
|
||||
|
||||
/** Claude Code's echo of a prompt it took in. */
|
||||
function echo(prompt: { uuid: string; message: unknown }, sessionId: string) {
|
||||
return {
|
||||
type: "user",
|
||||
message: prompt.message,
|
||||
parent_tool_use_id: null,
|
||||
uuid: prompt.uuid,
|
||||
session_id: sessionId,
|
||||
isReplay: true,
|
||||
};
|
||||
}
|
||||
|
||||
/** Scripts the query of the next session: each prompt runs the next turn script. */
|
||||
function scriptTurns(turns: TurnScript[], end: "stay open" | "end stream" = "stay open") {
|
||||
const promptUuids: string[] = [];
|
||||
scriptQuery(async function* (input, options) {
|
||||
for (const turn of turns) {
|
||||
const { value, done } = await input.next();
|
||||
if (done) return;
|
||||
promptUuids.push(value.uuid);
|
||||
yield echo(value, options.sessionId!);
|
||||
for await (const message of turn(options)) yield message;
|
||||
}
|
||||
if (end === "stay open") await input.next();
|
||||
});
|
||||
return promptUuids;
|
||||
}
|
||||
|
||||
/** Scripts the query of the next session as `run`, which reads the prompts. */
|
||||
function scriptQuery(
|
||||
run: (
|
||||
input: AsyncIterator<any>,
|
||||
options: QueryOptions,
|
||||
) => AsyncGenerator<Record<string, unknown>>,
|
||||
) {
|
||||
mockQuery.mockImplementation(
|
||||
({ prompt, options }: { prompt: AsyncIterable<any>; options: QueryOptions }) => {
|
||||
const sessionId = options.sessionId ?? "session";
|
||||
async function* run() {
|
||||
const input = prompt[Symbol.asyncIterator]();
|
||||
for (const turn of turns) {
|
||||
const { value, done } = await input.next();
|
||||
if (done) return;
|
||||
promptUuids.push(value.uuid);
|
||||
yield {
|
||||
type: "user",
|
||||
message: value.message,
|
||||
parent_tool_use_id: null,
|
||||
uuid: value.uuid,
|
||||
session_id: sessionId,
|
||||
isReplay: true,
|
||||
};
|
||||
for await (const message of turn(options)) yield message;
|
||||
}
|
||||
if (end === "stay open") await input.next();
|
||||
}
|
||||
return Object.assign(run(), {
|
||||
const sessionOptions = { ...options, sessionId: options.sessionId ?? "session" };
|
||||
return Object.assign(run(prompt[Symbol.asyncIterator](), sessionOptions), {
|
||||
initializationResult: async () => ({
|
||||
models: [{ value: "default", displayName: "Default", description: "" }],
|
||||
}),
|
||||
@@ -107,29 +121,72 @@ function scriptTurns(turns: TurnScript[], end: "stay open" | "end stream" = "sta
|
||||
});
|
||||
},
|
||||
);
|
||||
return promptUuids;
|
||||
}
|
||||
|
||||
/** Records the events of one turn as readable strings. */
|
||||
function recordEvents() {
|
||||
const log: string[] = [];
|
||||
/**
|
||||
* Records the events of one turn as readable strings, into `log` when given,
|
||||
* so several turns and the client's updates share one ordered trace.
|
||||
*/
|
||||
function recordEvents(log: string[] = [], label?: string) {
|
||||
const done = Promise.withResolvers<void>();
|
||||
const push = (entry: string) => log.push(label ? `${label} ${entry}` : entry);
|
||||
const events: TurnEvents = {
|
||||
inserted: (messageId) => log.push(`inserted ${messageId}`),
|
||||
awaitingUser: () => log.push("awaitingUser"),
|
||||
resumed: () => log.push("resumed"),
|
||||
inserted: (messageId) => push(`inserted ${messageId}`),
|
||||
awaitingUser: () => push("awaitingUser"),
|
||||
resumed: () => push("resumed"),
|
||||
ended: (outcome) => {
|
||||
log.push(`ended ${outcome.stopReason}`);
|
||||
push(`ended ${outcome.stopReason}`);
|
||||
done.resolve();
|
||||
},
|
||||
failed: (error) => {
|
||||
log.push(`failed ${error instanceof Error ? error.message : String(error)}`);
|
||||
push(`failed ${error instanceof Error ? error.message : String(error)}`);
|
||||
done.resolve();
|
||||
},
|
||||
};
|
||||
return { log, events, done: done.promise };
|
||||
}
|
||||
|
||||
function lifecycle(commandUuid: string, state: string, sessionId: string) {
|
||||
return {
|
||||
type: "command_lifecycle",
|
||||
command_uuid: commandUuid,
|
||||
state,
|
||||
uuid: randomUUID(),
|
||||
session_id: sessionId,
|
||||
};
|
||||
}
|
||||
|
||||
function stream(event: Record<string, unknown>, sessionId: string) {
|
||||
return {
|
||||
type: "stream_event",
|
||||
event,
|
||||
parent_tool_use_id: null,
|
||||
uuid: randomUUID(),
|
||||
session_id: sessionId,
|
||||
};
|
||||
}
|
||||
|
||||
/** The start of an answer that streams one text delta. */
|
||||
function streamedText(messageId: string, text: string, sessionId: string) {
|
||||
return [
|
||||
stream(
|
||||
{
|
||||
type: "message_start",
|
||||
message: { id: messageId, role: "assistant", content: [], usage: {} },
|
||||
},
|
||||
sessionId,
|
||||
),
|
||||
stream(
|
||||
{ type: "content_block_delta", index: 0, delta: { type: "text_delta", text } },
|
||||
sessionId,
|
||||
),
|
||||
];
|
||||
}
|
||||
|
||||
function system(subtype: string, sessionId: string, fields: Record<string, unknown> = {}) {
|
||||
return { type: "system", subtype, uuid: randomUUID(), session_id: sessionId, ...fields };
|
||||
}
|
||||
|
||||
describe("turn events", () => {
|
||||
let agent: ClaudeAcpAgent;
|
||||
let permissionRequests: RequestPermissionRequest[];
|
||||
@@ -186,6 +243,77 @@ describe("turn events", () => {
|
||||
expect(turn.log).toEqual([`inserted ${promptUuids[0]}`, "ended end_turn"]);
|
||||
});
|
||||
|
||||
it("reports a turn inserted when Claude Code starts it, before output that comes before its echo", async () => {
|
||||
const trace: string[] = [];
|
||||
onSessionUpdate = ({ update }) => {
|
||||
// The turn's answer; the session also sends advisories as chunks.
|
||||
if (update.sessionUpdate === "agent_message_chunk" && update.messageId === "msg_answer") {
|
||||
trace.push("chunk");
|
||||
}
|
||||
};
|
||||
let promptUuid = "";
|
||||
scriptQuery(async function* (input, options) {
|
||||
const { value } = await input.next();
|
||||
promptUuid = value.uuid;
|
||||
// A fresh turn reports "started" before anything it produces; its
|
||||
// answer can then stream before the echo of its prompt.
|
||||
yield lifecycle(value.uuid, "started", options.sessionId!);
|
||||
yield* streamedText("msg_answer", "Hello", options.sessionId!);
|
||||
yield echo(value, options.sessionId!);
|
||||
yield result(options.sessionId!);
|
||||
await input.next();
|
||||
});
|
||||
const sessionId = await newSession();
|
||||
const turn = recordEvents(trace);
|
||||
|
||||
await agent.startTurn(prompt(sessionId, "hello"), turn.events);
|
||||
await turn.done;
|
||||
|
||||
expect(trace).toEqual([`inserted ${promptUuid}`, "chunk", "ended end_turn"]);
|
||||
});
|
||||
|
||||
it("reports a turn inserted only after a held turn before it ended", async () => {
|
||||
const trace: string[] = [];
|
||||
scriptQuery(async function* (input, options) {
|
||||
const sessionId = options.sessionId!;
|
||||
const first = (await input.next()).value;
|
||||
yield lifecycle(first.uuid, "started", sessionId);
|
||||
yield echo(first, sessionId);
|
||||
yield system("task_started", sessionId, {
|
||||
task_id: "agent-1",
|
||||
tool_use_id: "toolu_agent-1",
|
||||
description: "Explore the project",
|
||||
subagent_type: "Explore",
|
||||
});
|
||||
// The result is held for the live background subagent.
|
||||
yield result(sessionId);
|
||||
yield system("session_state_changed", sessionId, { state: "idle" });
|
||||
const second = (await input.next()).value;
|
||||
// The next prompt starts a turn while the first is still held.
|
||||
yield lifecycle(second.uuid, "started", sessionId);
|
||||
yield echo(second, sessionId);
|
||||
yield result(sessionId);
|
||||
await input.next();
|
||||
});
|
||||
const sessionId = await newSession();
|
||||
const first = recordEvents(trace, "first");
|
||||
const second = recordEvents(trace, "second");
|
||||
|
||||
await agent.startTurn(prompt(sessionId, "explore"), first.events);
|
||||
await vi.waitFor(() =>
|
||||
expect(agent.sessions[sessionId]?.activeTurn?.deferredSettle).toBeDefined(),
|
||||
);
|
||||
await agent.startTurn(prompt(sessionId, "next"), second.events);
|
||||
await Promise.all([first.done, second.done]);
|
||||
|
||||
expect(trace).toEqual([
|
||||
expect.stringMatching(/^first inserted /),
|
||||
"first ended end_turn",
|
||||
expect.stringMatching(/^second inserted /),
|
||||
"second ended end_turn",
|
||||
]);
|
||||
});
|
||||
|
||||
it("reports waiting on a permission request, then resuming", async () => {
|
||||
scriptTurns([
|
||||
async function* (options) {
|
||||
|
||||
@@ -13,9 +13,14 @@ export interface TurnEvents {
|
||||
/**
|
||||
* Claude Code took the prompt in as the user message `messageId` of the
|
||||
* conversation. Reported at most once per turn, and never after
|
||||
* {@link ended} or {@link failed}. For a prompt Claude Code echoes, that is
|
||||
* the echo; for a command it does not echo (such as `/compact`), it is the
|
||||
* command's result, after any output the command streamed.
|
||||
* {@link ended} or {@link failed}.
|
||||
*
|
||||
* When Claude Code reports command lifecycles (`msg_lifecycle_v1`), that is
|
||||
* when the prompt starts a turn, before the turn's output. Otherwise, and
|
||||
* while another turn is still active (held for its background work,
|
||||
* steered, or cancelled but not yet ended), it is the prompt's echo, after
|
||||
* that turn ended; for a command Claude Code does not echo (such as
|
||||
* `/compact`), it is the command's result, after any output it streamed.
|
||||
*/
|
||||
inserted(messageId: string): void;
|
||||
/**
|
||||
|
||||
Reference in New Issue
Block a user