Files
claude-agent-acp/src/async-tasks.ts
Ben Brandt 6597626d86 fix: list a subagent's background Bash commands (#1540)
Claude Code sends no `tool_use_result` with a subagent's tool results.
The adapter recognizes a background Bash command only from the
`backgroundTaskId` in that structured result, so it never recognized one
that a subagent started. The command's terminal was reported exited
(code 0) as soon as the result came, while the command kept running, and
the command was missing from the subagent's `state_update.background`.

The real SDK stream (CLI 2.1.318) for a subagent running `sleep 30` in
the background:

```
task_started  task_id=beure6fer task_type=local_bash tool_use_id=toolu_018M… owned_by_subagent=true is_backgrounded=true
user          parent_tool_use_id=toolu_01HY… tool_result for toolu_018M…, no tool_use_result
```

The local transcripts agree: none of the 35 subagent background Bash
results across CLI 2.1.293 to 2.1.318 has a `toolUseResult`.

When a Bash result has no structured task id, the adapter now uses the
live background `local_bash` task whose `task_started` named that Bash
call. The existing subagent test now sends the stream the CLI sends. It
fails without the fix.
2026-11-06 19:12:42 +01:00

1138 lines
42 KiB
TypeScript

import { randomUUID } from "node:crypto";
import type { ClientCapabilities } from "@agentclientprotocol/sdk/unstable";
import type { AcpSessionNotification, AsyncTaskState } from "./acp-subagents.js";
import { AIR_ASYNC_TASKS_CAPABILITY, clientSupportsAirCapability } from "./air-extension.js";
import { noticeOrTranscriptUpdate } from "./session-notices.js";
type Publish = (notification: AcpSessionNotification) => Promise<void>;
/** Moves an update to its owner session, or drops it with `null`. */
type Route = (notification: AcpSessionNotification) => AcpSessionNotification | null;
type TerminalSource = "event" | "level" | "shutdown";
type AsyncTask = {
id: string;
name: string;
taskType: string;
description: string;
showInTranscript: boolean;
/** A Monitor tool call or an SDK `local_monitor` task: published as a monitor the client cannot stop. */
monitor: boolean;
outputFilePath?: string;
toolCallId?: string;
/**
* The route to the session of the tool call that started the task. The
* spawn sets it, so every update of the task goes to one session. A task
* without a route stays in the root session.
*/
route?: Route;
announced: boolean;
/**
* Whether the spawn waits for the tool call id. The updates of a held task
* wait in {@link AsyncTask.deferred} and follow the spawn in order.
*/
held: boolean;
deferred: (() => Promise<void>)[];
ignored: boolean;
startedObserved: boolean;
panelOnlyRecovery: boolean;
stopping: boolean;
stopAnnounced: boolean;
state: AsyncTaskState;
terminalSummary?: string;
terminalSource?: TerminalSource;
/** The optional fields that the client holds now, as JSON. */
published: Map<string, string>;
/**
* Whether the task stands for a model turn that Claude Code started on its
* own, see {@link AsyncTaskRuntime.autonomousTurnStarted}. The SDK knows no
* such task, so its background task list never ends it.
*/
autonomousTurn: boolean;
};
type TaskIdentity = { taskId?: unknown; task_id?: unknown };
const MAX_TOOL_RESULT_PATHS = 500;
export type AsyncTaskStarted = TaskIdentity & {
taskType?: unknown;
task_type?: unknown;
description?: unknown;
subagentType?: unknown;
subagent_type?: unknown;
isBackgrounded?: unknown;
is_backgrounded?: unknown;
workflowName?: unknown;
workflow_name?: unknown;
skipTranscript?: unknown;
skip_transcript?: unknown;
outputFilePath?: unknown;
output_file?: unknown;
toolCallId?: unknown;
tool_use_id?: unknown;
};
export type AsyncTaskUpdated = TaskIdentity & { patch?: TaskPatch };
type TaskPatch = {
status?: unknown;
description?: unknown;
isBackgrounded?: unknown;
is_backgrounded?: unknown;
error?: unknown;
outputFilePath?: unknown;
output_file?: unknown;
};
export type AsyncTaskNotification = TaskIdentity & {
status?: unknown;
toolCallId?: unknown;
tool_use_id?: unknown;
summary?: unknown;
outputFilePath?: unknown;
output_file?: unknown;
};
type TaskProgress = TaskIdentity & {
description?: unknown;
toolCallId?: unknown;
tool_use_id?: unknown;
summary?: unknown;
lastToolName?: unknown;
last_tool_name?: unknown;
usage?: unknown;
outputFilePath?: unknown;
output_file?: unknown;
};
type BackgroundTaskLevel = TaskIdentity & {
taskType?: unknown;
task_type?: unknown;
description?: unknown;
};
export function clientSupportsAsyncTasks(capabilities?: ClientCapabilities | null): boolean {
return clientSupportsAirCapability(capabilities, AIR_ASYNC_TASKS_CAPABILITY);
}
/** Publishes Claude's non-agent background work as a separate AIR task lifecycle. */
export class AsyncTaskRuntime {
/**
* One registry owns both active tasks and terminal tombstones. Keeping an
* unannounced terminal record is intentional: the Bash result proving that
* a command was backgrounded can arrive after its terminal SDK edge.
*/
private readonly tasks = new Map<string, AsyncTask>();
/**
* The task output paths in the text of each tool result, by tool call
* id. A task whose id and tool call are known takes its output path from
* here. The tool result can come before or after the SDK names the task.
* The oldest entries are dropped.
*/
private readonly outputPathsByToolCall = new Map<string, string[]>();
/** The task of the autonomous turn that runs now. */
private autonomousTurn?: AsyncTask;
/** The summary of the last task notification, which usually starts the next autonomous turn. */
private lastNotificationSummary?: string;
constructor(
readonly enabled: boolean,
private readonly sessionId: string,
private readonly publish: Publish,
/** `notices`: the client can present `notice` updates, so the stop
* acknowledgement need not be a transcript line. `routeOf`: the route of
* the tasks that a tool call in another session started, for example in
* a native subagent session. `toolNameOf`: the name of the tool of a tool
* call, or `undefined` when the tool call is unknown. A task that a
* `Monitor` tool call started is published as a monitor: the SDK reports
* it as `local_bash`, so the adapter names it a monitor. */
private readonly options: {
notices?: boolean;
routeOf?: (toolCallId: string) => Route | undefined;
toolNameOf?: (toolCallId: string) => string | undefined;
} = {},
) {}
async taskStarted(message: AsyncTaskStarted): Promise<void> {
if (!this.enabled) return;
const taskId = taskIdOf(message);
if (!taskId) return;
const task = this.task(taskId);
if (isSubagentTask(message)) {
this.ignore(task);
return;
}
const previousOutputFilePath = task.outputFilePath;
const previousToolCallId = task.toolCallId;
const wasAnnounced = task.announced;
this.mergeStarted(task, message);
if (
isBackgroundTask(
field(message, "isBackgrounded", "is_backgrounded"),
field(message, "taskType", "task_type"),
) ||
task.held
) {
await this.announce(task);
}
if (
wasAnnounced &&
(task.outputFilePath !== previousOutputFilePath || task.toolCallId !== previousToolCallId)
) {
await this.publishMetadata(task);
}
}
async taskBackgrounded(message: AsyncTaskStarted): Promise<void> {
if (!this.enabled) return;
const taskId = taskIdOf(message);
if (!taskId) return;
const task = this.task(taskId);
if (isSubagentTask(message)) {
this.ignore(task);
return;
}
const previousOutputFilePath = task.outputFilePath;
const previousToolCallId = task.toolCallId;
const wasAnnounced = task.announced;
this.mergeStarted(task, message);
await this.announce(task);
if (
wasAnnounced &&
(task.outputFilePath !== previousOutputFilePath || task.toolCallId !== previousToolCallId)
) {
await this.publishMetadata(task);
}
}
async taskUpdated(message: AsyncTaskUpdated): Promise<void>;
async taskUpdated(taskId: string, patch: TaskPatch): Promise<void>;
async taskUpdated(taskOrId: string | AsyncTaskUpdated, legacyPatch?: TaskPatch): Promise<void> {
if (!this.enabled) return;
const taskId = typeof taskOrId === "string" ? nonBlankString(taskOrId) : taskIdOf(taskOrId);
const patch = typeof taskOrId === "string" ? legacyPatch : taskOrId.patch;
if (!taskId || !patch) return;
const task = this.task(taskId);
const previousOutputFilePath = task.outputFilePath;
this.mergePatch(task, patch);
if (field(patch, "isBackgrounded", "is_backgrounded") === true) {
await this.announce(task);
}
const state = taskState(patch.status);
if (!state) {
if (task.announced && task.outputFilePath !== previousOutputFilePath) {
await this.publishMetadata(task);
}
return;
}
if (state === "running" || state === "paused") {
if (isTerminal(task.state)) {
if (task.announced && task.outputFilePath !== previousOutputFilePath) {
await this.publishMetadata(task);
}
return;
}
if (task.state !== state) {
task.state = state;
const error = nonBlankString(patch.error);
await this.whenAnnounced(task, () => this.publishState(task, state, error));
} else if (task.announced && task.outputFilePath !== previousOutputFilePath) {
await this.publishMetadata(task);
}
return;
}
await this.finish(task, state, nonBlankString(patch.error), "event");
}
async taskProgress(message: TaskProgress): Promise<void> {
if (!this.enabled) return;
const taskId = taskIdOf(message);
if (!taskId) return;
const task = this.tasks.get(taskId);
if (!task || task.ignored || !(task.announced || task.held)) return;
const previousOutputFilePath = task.outputFilePath;
const previousToolCallId = task.toolCallId;
this.mergeOutputFilePath(task, message);
this.mergeToolCallId(task, message);
const outputChanged = task.outputFilePath !== previousOutputFilePath;
if (isTerminal(task.state)) {
if (outputChanged) await this.publishMetadata(task);
return;
}
const usage = taskUsage(message.usage);
const description = nonBlankString(message.description);
const summary = nonBlankString(message.summary);
const lastToolName = nonBlankString(field(message, "lastToolName", "last_tool_name"));
const progress = {
...(description ? { description } : {}),
...(summary ? { summary } : {}),
...(lastToolName ? { lastToolName } : {}),
...(usage ? { usage } : {}),
...(outputChanged && task.outputFilePath ? { outputFilePath: task.outputFilePath } : {}),
...(task.toolCallId !== previousToolCallId && task.toolCallId
? { toolCallId: task.toolCallId }
: {}),
};
await this.whenAnnounced(task, () => this.publishProgress(task, progress));
// The progress brought the tool call id of a held task: send the spawn,
// then the progress beats that waited for it.
if (task.held && task.toolCallId) await this.announce(task);
}
async taskNotification(message: AsyncTaskNotification): Promise<void>;
async taskNotification(
taskId: string,
status: unknown,
summary?: unknown,
outputFilePath?: unknown,
): Promise<void>;
async taskNotification(
taskOrId: string | AsyncTaskNotification,
legacyStatus?: unknown,
legacySummary?: unknown,
legacyOutputFilePath?: unknown,
): Promise<void> {
if (!this.enabled) return;
const message: AsyncTaskNotification =
typeof taskOrId === "string"
? {
taskId: taskOrId,
status: legacyStatus,
summary: legacySummary,
outputFilePath: legacyOutputFilePath,
}
: taskOrId;
const taskId = taskIdOf(message);
const state = taskState(message.status);
if (!taskId || !state || state === "running" || state === "paused") return;
this.lastNotificationSummary = nonBlankString(message.summary) ?? this.lastNotificationSummary;
const task = this.task(taskId);
const wasTerminal = isTerminal(task.state);
const correctsLevelState = wasTerminal && task.terminalSource === "level";
const previousOutputFilePath = task.outputFilePath;
this.mergeOutputFilePath(task, message);
this.mergeToolCallId(task, message);
await this.finish(task, state, nonBlankString(message.summary), "event");
if (
wasTerminal &&
!correctsLevelState &&
task.announced &&
task.outputFilePath !== previousOutputFilePath
) {
await this.publishMetadata(task);
}
}
/**
* Reconciles the SDK's replace-semantics background task level. It does not
* create unknown tasks because this level normally precedes task_started and
* carries no tool-call attribution. It can, however, promote a known
* foreground task and terminate an announced task whose edge was lost.
*/
async backgroundTasksChanged(
tasksOrMessage: readonly BackgroundTaskLevel[] | { tasks?: unknown },
): Promise<void> {
if (!this.enabled) return;
const value = Array.isArray(tasksOrMessage)
? tasksOrMessage
: isRecord(tasksOrMessage)
? tasksOrMessage.tasks
: undefined;
if (!Array.isArray(value)) return;
const live = new Set<string>();
for (const item of value) {
if (!isRecord(item)) continue;
const taskId = taskIdOf(item);
if (!taskId) continue;
live.add(taskId);
const task = this.task(taskId);
if (task.ignored || isTerminal(task.state)) continue;
const levelTaskType = field(item, "taskType", "task_type");
if (levelTaskType === "local_agent") {
this.ignore(task);
continue;
}
if (!task.startedObserved) {
// The replace-level proves liveness but carries neither transcript
// policy nor full identity. Recover it for the Async Tasks panel only;
// creating a transcript card here would be irreversible if a late
// task_started says skip_transcript=true.
task.panelOnlyRecovery = true;
task.showInTranscript = false;
}
this.mergeLevel(task, item);
await this.announce(task);
}
for (const task of this.tasks.values()) {
if (
(task.announced || task.held) &&
!task.ignored &&
!task.autonomousTurn &&
!isTerminal(task.state) &&
!live.has(task.id)
) {
// This replace-semantics level is itself the authoritative liveness
// boundary. Close immediately so a lost terminal edge cannot leave a
// permanent running card. Since SDK 0.3.292 a finishing task's level
// follows its terminal edges, so this is a lost or late edge; a late
// task_notification still corrects the best-effort stopped state to
// completed/failed.
await this.finish(task, "stopped", undefined, "level");
}
}
}
async finishAll(state: Extract<AsyncTaskState, "failed" | "stopped">): Promise<void> {
this.autonomousTurn = undefined;
const errors: unknown[] = [];
for (const task of this.tasks.values()) {
if ((task.announced || task.held) && !task.ignored && !isTerminal(task.state)) {
try {
await this.finish(task, state, undefined, "shutdown");
} catch (error) {
errors.push(error);
}
}
}
if (errors.length === 1) throw errors[0];
if (errors.length > 1) throw new AggregateError(errors, "Failed to finish async tasks");
}
canStop(taskId: string): boolean {
const task = this.tasks.get(taskId);
return (
task?.announced === true &&
!task.monitor &&
!task.ignored &&
!task.stopping &&
!isTerminal(task.state)
);
}
claimStop(taskId: string): boolean {
if (!this.canStop(taskId)) return false;
this.tasks.get(taskId)!.stopping = true;
return true;
}
releaseStop(taskId: string): void {
const task = this.tasks.get(taskId);
if (task && !isTerminal(task.state)) task.stopping = false;
}
async taskStopped(taskId: string): Promise<void> {
const task = this.tasks.get(taskId);
if (!task || !task.announced || task.ignored || task.stopAnnounced) return;
task.stopAnnounced = true;
if (this.autonomousTurn === task) this.autonomousTurn = undefined;
// No terminal summary: a stopped task leaves the Async Tasks panel at once,
// so anything said there is said to nobody.
await this.finish(task, "stopped", undefined, "event");
// This acknowledgement is the only one the user gets -- the SDK injects
// nothing into the model's context for a stopped shell task. A client on
// the notice contract shows it as a live `notice`; otherwise the
// transcript is where it has to land.
//
// Terminal state deliberately does not gate this. The SDK's own
// `task_notification` routinely wins the race against the `stopTask`
// response that brought us here, so by now the task is usually already
// terminal: gating on that would silence the very stop the user asked for.
// `stopAnnounced` is what keeps this to one line per task.
//
// `showInTranscript` does not gate it either: that flag decides whether the
// task owns a transcript *card* (a background Bash task is already drawn as
// its tool call), not whether the agent may answer a direct user action.
await this.send(
task,
noticeOrTranscriptUpdate(
{ severity: "info", title: "Task stopped by user", description: `${task.name}.` },
this.options.notices ?? false,
),
);
}
/**
* Publishes a model turn that Claude Code started with no prompt open, for
* example to answer a task notification. ACP v1 has no session state, and
* the client already got the stop reason of its last prompt, so this task is
* the only sign that the agent works. The tool calls of the turn are in the
* transcript already, so the task is not shown there.
*
* [cause] names what started the turn when Claude Code says so, for example
* "A scheduled wakeup fired." It replaces the description of a task that is
* already open; without it, an open task is left as it is.
*/
async autonomousTurnStarted(cause?: string): Promise<void> {
if (!this.enabled) return;
if (this.autonomousTurn) {
if (cause) await this.publishProgress(this.autonomousTurn, { description: cause });
return;
}
const task = this.task(`autonomous-turn-${randomUUID()}`);
task.autonomousTurn = true;
task.startedObserved = true;
task.name = "Claude is working";
task.taskType = "turn";
task.description =
cause ?? this.lastNotificationSummary ?? "Claude Code started a turn without a prompt.";
task.showInTranscript = false;
this.lastNotificationSummary = undefined;
this.autonomousTurn = task;
await this.announce(task, { withoutToolCall: true });
}
/** Whether the task of an autonomous turn is open. */
get autonomousTurnOpen(): boolean {
return this.autonomousTurn !== undefined;
}
/** Ends the task of the autonomous turn that runs now, if there is one. */
async autonomousTurnEnded(state: "completed" | "failed"): Promise<void> {
const task = this.autonomousTurn;
if (!task) return;
this.autonomousTurn = undefined;
await this.finish(task, state, undefined, "event");
}
/** Whether the task stands for an autonomous turn, which a stop interrupts. */
isAutonomousTurn(taskId: string): boolean {
return this.tasks.get(taskId)?.autonomousTurn === true;
}
/** Whether a task of the tool call went to the background, so the client gets it. */
isBackgroundedToolCall(toolCallId: string): boolean {
for (const task of this.tasks.values()) {
if (task.toolCallId === toolCallId && !task.ignored && (task.announced || task.held)) {
return true;
}
}
return false;
}
/**
* Reads the tool results in `content` for the output path of a known task.
* Only a path that ends with `tasks/<task id>.output` in the result of
* the task's own tool call counts. A tool result never creates a task.
*/
async toolResults(content: unknown): Promise<void> {
if (!this.enabled || !Array.isArray(content)) return;
for (const result of content.flatMap(toolResultBlock)) {
const text = textContent(result.content);
const paths = text ? taskOutputPaths(text) : [];
if (paths.length === 0) continue;
this.outputPathsByToolCall.delete(result.toolUseId);
this.outputPathsByToolCall.set(result.toolUseId, paths);
if (this.outputPathsByToolCall.size > MAX_TOOL_RESULT_PATHS) {
const oldest = this.outputPathsByToolCall.keys().next().value;
if (oldest !== undefined) this.outputPathsByToolCall.delete(oldest);
}
for (const task of this.tasks.values()) {
if (task.toolCallId !== result.toolUseId || task.ignored || task.outputFilePath) continue;
this.mergeToolResultPath(task);
if (task.outputFilePath && task.announced) await this.publishMetadata(task);
}
}
}
/**
* Sends the spawn of each held task without a tool call id. The prompt
* result ends the model turn, and the result of the tool call that started a
* task comes before it. So a held task is never held for longer than one turn.
*/
async releaseHeld(): Promise<void> {
if (!this.enabled) return;
for (const task of this.tasks.values()) {
if (task.held) await this.announce(task, { withoutToolCall: true });
}
}
clear(): void {
this.tasks.clear();
this.outputPathsByToolCall.clear();
this.autonomousTurn = undefined;
this.lastNotificationSummary = undefined;
}
/**
* Marks a task that the client never sees. Only a task that is not announced
* can become ignored, so the client never holds a task without its end.
*/
private ignore(task: AsyncTask): void {
if (task.announced) return;
task.ignored = true;
task.held = false;
task.deferred = [];
}
private task(taskId: string): AsyncTask {
let task = this.tasks.get(taskId);
if (task) return task;
task = {
id: taskId,
name: "Background task",
taskType: "task",
description: "Background task",
showInTranscript: true,
monitor: false,
announced: false,
held: false,
deferred: [],
ignored: false,
startedObserved: false,
panelOnlyRecovery: false,
stopping: false,
stopAnnounced: false,
state: "running",
published: new Map(),
autonomousTurn: false,
};
this.tasks.set(taskId, task);
return task;
}
private mergeStarted(task: AsyncTask, message: AsyncTaskStarted): void {
task.startedObserved = true;
const rawTaskType = field(message, "taskType", "task_type");
if (rawTaskType !== undefined) task.taskType = friendlyTaskType(rawTaskType);
if (rawTaskType === "local_monitor") task.monitor = true;
const description = nonBlankString(message.description);
if (description) task.description = description;
else if (task.description === "Background task")
task.description = defaultDescription(task.taskType);
task.name = nonBlankString(field(message, "workflowName", "workflow_name")) ?? task.description;
const skipTranscript = field(message, "skipTranscript", "skip_transcript");
if (task.panelOnlyRecovery) {
// A level-only recovery was already announced without a transcript
// entry. Keep that decision monotonic: a late task_started can enrich
// the panel, but cannot retroactively create or mutate transcript UI.
task.showInTranscript = false;
} else if (skipTranscript !== undefined) {
task.showInTranscript = skipTranscript !== true;
}
this.mergeOutputFilePath(task, message);
const toolCallId = nonBlankString(field(message, "toolCallId", "tool_use_id"));
if (toolCallId) task.toolCallId = toolCallId;
this.mergeToolResultPath(task);
}
/** Keeps the first tool call id that a later SDK message brings. */
private mergeToolCallId(
task: AsyncTask,
value: { toolCallId?: unknown; tool_use_id?: unknown },
): void {
task.toolCallId ??= nonBlankString(field(value, "toolCallId", "tool_use_id"));
this.mergeToolResultPath(task);
}
/**
* Takes the output path from the result of the task's tool call when no
* structured source gave one. A structured path later replaces it.
*/
private mergeToolResultPath(task: AsyncTask): void {
if (task.outputFilePath || !task.toolCallId) return;
const paths = this.outputPathsByToolCall.get(task.toolCallId);
if (paths) task.outputFilePath = outputFilePathOf(paths, task.id);
}
/** Sends an update of an announced task. A held task sends it after its spawn. */
private async whenAnnounced(task: AsyncTask, send: () => Promise<void>): Promise<void> {
if (task.announced) await send();
else if (task.held) task.deferred.push(send);
}
private mergePatch(task: AsyncTask, patch: TaskPatch): void {
const description = nonBlankString(patch.description);
if (description) {
task.description = description;
task.name = description;
}
this.mergeOutputFilePath(task, patch);
}
private mergeLevel(task: AsyncTask, level: BackgroundTaskLevel): void {
const rawTaskType = field(level, "taskType", "task_type");
if (rawTaskType !== undefined) task.taskType = friendlyTaskType(rawTaskType);
if (rawTaskType === "local_monitor") task.monitor = true;
const description = nonBlankString(level.description);
if (description) {
task.description = description;
task.name = description;
}
}
private mergeOutputFilePath(
task: AsyncTask,
value: { outputFilePath?: unknown; output_file?: unknown },
): void {
const outputFilePath = nonBlankString(field(value, "outputFilePath", "output_file"));
if (outputFilePath) task.outputFilePath = outputFilePath;
}
/**
* Sends the spawn of the task. The spawn names the tool call that started the
* task, so it waits for the tool call id: the task is held until the id
* arrives. The id comes as `tool_use_id` of the SDK `task_started`,
* `task_progress`, or `task_notification`, or from the Bash result of a
* backgrounded command. A held task gets its spawn without the id when it
* ends first, or when the prompt result ends the model turn
* ({@link releaseHeld}). No guess by the command text binds a task.
*
* A task that a `Monitor` tool call started is published as a monitor. The
* SDK reports it as `local_bash`, so the adapter names it a monitor instead of
* a shell. It has no transcript card because Monitor streams its output to
* the model only, and no Stop because stopping it only ends the agent's wait
* (the agent's own TaskStop still works). The tool_use streams before the
* tool runs, so the tool name is known when the spawn has the tool call id.
*/
private async announce(
task: AsyncTask,
options: { withoutToolCall?: boolean } = {},
): Promise<void> {
if (task.announced || task.ignored) return;
if (!task.toolCallId && !options.withoutToolCall && !isTerminal(task.state)) {
task.held = true;
return;
}
if (task.toolCallId && this.options.toolNameOf?.(task.toolCallId) === "Monitor") {
task.monitor = true;
}
if (task.monitor) {
task.taskType = "monitor";
task.showInTranscript = false;
}
if (task.toolCallId) task.route ??= this.options.routeOf?.(task.toolCallId);
await this.send(task, {
sessionUpdate: "async_task_spawned",
asyncTaskId: task.id,
name: task.name,
taskType: task.taskType,
description: task.description,
showInTranscript: task.showInTranscript,
canStop: !task.monitor,
...(task.outputFilePath ? { outputFilePath: task.outputFilePath } : {}),
...(task.toolCallId ? { toolCallId: task.toolCallId } : {}),
});
task.announced = true;
task.held = false;
recordPublished(task, {
description: task.description,
outputFilePath: task.outputFilePath,
toolCallId: task.toolCallId,
});
for (const send of task.deferred.splice(0)) await send();
if (isTerminal(task.state)) {
await this.publishState(task, task.state, task.terminalSummary);
}
}
private async finish(
task: AsyncTask,
state: Extract<AsyncTaskState, "completed" | "failed" | "stopped">,
summary: string | undefined,
source: TerminalSource,
): Promise<void> {
if (task.ignored) return;
if (isTerminal(task.state)) {
// A level event can precede a late terminal edge (older CLIs sent the
// level first). Correct its best-effort stopped state if that edge
// later arrives.
if (task.terminalSource !== "level" || source !== "event") return;
}
const previous = {
state: task.state,
terminalSummary: task.terminalSummary,
terminalSource: task.terminalSource,
};
task.state = state;
task.terminalSummary = summary;
task.terminalSource = source;
try {
if (task.announced) await this.publishState(task, state, summary);
// The task ended before its tool call id arrived. The spawn goes out now
// without the id, so the task still appears.
else if (task.held) await this.announce(task);
} catch (error) {
task.state = previous.state;
task.terminalSummary = previous.terminalSummary;
task.terminalSource = previous.terminalSource;
throw error;
}
}
private async publishMetadata(task: AsyncTask): Promise<void> {
if (task.outputFilePath || task.toolCallId) {
if (isTerminal(task.state)) {
await this.publishState(task, task.state, task.terminalSummary);
} else {
await this.publishProgress(task, {
...(task.outputFilePath ? { outputFilePath: task.outputFilePath } : {}),
...(task.toolCallId ? { toolCallId: task.toolCallId } : {}),
});
}
}
}
/**
* Publishes the progress fields that changed since the last report of the
* task. A beat with nothing new is not sent.
*/
private async publishProgress(
task: AsyncTask,
update: {
description?: string;
summary?: string;
lastToolName?: string;
usage?: { totalTokens: number; toolUses: number; durationMs: number };
outputFilePath?: string;
toolCallId?: string;
},
): Promise<void> {
const changed = changedFields(task, update);
if (Object.keys(changed).length === 0) return;
await this.send(task, {
sessionUpdate: "async_task_progress",
asyncTaskId: task.id,
...changed,
});
recordPublished(task, changed);
}
/** Publishes a state, with the output path and the tool call only when they changed. */
private async publishState(
task: AsyncTask,
state: AsyncTaskState,
summary?: string,
): Promise<void> {
const changed = changedFields(task, {
outputFilePath: task.outputFilePath,
toolCallId: task.toolCallId,
});
await this.send(task, {
sessionUpdate: "async_task_state_update",
asyncTaskId: task.id,
state,
...(summary ? { summary } : {}),
...changed,
});
recordPublished(task, changed);
}
/** Publishes an update of the task in the session that owns the task. */
private async send(task: AsyncTask, update: AcpSessionNotification["update"]): Promise<void> {
const notification: AcpSessionNotification = { sessionId: this.sessionId, update };
const routed = task.route ? task.route(notification) : notification;
if (routed) await this.publish(routed);
}
}
/**
* The fields whose value differs from what the client holds. An undefined
* value is not a field. The caller records the fields with
* {@link recordPublished} after the send succeeds, so a retry after a failed
* send carries them again.
*/
function changedFields<T extends Record<string, unknown>>(task: AsyncTask, fields: T): Partial<T> {
const changed: Partial<T> = {};
for (const [key, value] of Object.entries(fields)) {
if (value === undefined) continue;
if (task.published.get(key) === JSON.stringify(value)) continue;
(changed as Record<string, unknown>)[key] = value;
}
return changed;
}
/** Records the fields that the client holds now. An undefined value is not a field. */
function recordPublished(task: AsyncTask, fields: Record<string, unknown>): void {
for (const [key, value] of Object.entries(fields)) {
if (value !== undefined) task.published.set(key, JSON.stringify(value));
}
}
/**
* Recovers a background Bash task from the structured `backgroundTaskId` of
* its tool result. The text of a tool result never gives a task id: a
* foreground command can print any text.
*
* Claude Code sends no `tool_use_result` with a subagent's tool results. For
* those, `startedTaskOf` gives the background `local_bash` task whose
* `task_started` named the Bash call, which comes before the result.
*/
export function backgroundBashTaskFromToolResult(
content: unknown,
toolUseResult: unknown,
toolUses: Record<string, { name: string; input: unknown }>,
startedTaskOf?: (toolUseId: string) => string | undefined,
): (AsyncTaskStarted & { taskId: string; toolCallId: string }) | undefined {
if (!Array.isArray(content)) return undefined;
const toolResults = content.flatMap(toolResultBlock);
const bashResults = toolResults.filter((result) => toolUses[result.toolUseId]?.name === "Bash");
if (bashResults.length === 0) return undefined;
let backgroundResults = structuredBackgroundResults(toolUseResult, toolUses);
if (backgroundResults.length === 0 && startedTaskOf) {
backgroundResults = bashResults.flatMap((result) => {
const taskId = result.isError ? undefined : startedTaskOf(result.toolUseId);
return taskId ? [{ taskId, toolUseId: result.toolUseId }] : [];
});
}
const distinctTaskIds = new Set(backgroundResults.map((result) => result.taskId));
if (distinctTaskIds.size !== 1) return undefined;
const taskId = [...distinctTaskIds][0];
if (!taskId) return undefined;
const hintedToolUseIds = new Set(
backgroundResults
.filter((result) => result.taskId === taskId && result.toolUseId)
.map((result) => result.toolUseId!),
);
let matches: typeof bashResults;
if (hintedToolUseIds.size > 0) {
matches = bashResults.filter((result) => hintedToolUseIds.has(result.toolUseId));
} else {
const pathMatches = bashResults.filter(
(result) => asyncTaskOutputFilePath(result.content, taskId) !== undefined,
);
matches = pathMatches.length > 0 ? pathMatches : bashResults.length === 1 ? bashResults : [];
}
if (matches.length !== 1) return undefined;
const match = matches[0];
const toolUse = toolUses[match.toolUseId];
const input = isRecord(toolUse?.input) ? toolUse.input : undefined;
return {
taskId,
taskType: "local_bash",
description: nonBlankString(input?.command),
isBackgrounded: true,
outputFilePath: asyncTaskOutputFilePath(match.content, taskId),
toolCallId: match.toolUseId,
};
}
/**
* The Bash tool calls in `content` whose command went to the background. Only
* structured data counts: the task from the structured tool result, a task
* that the runtime knows for the tool call, or a `run_in_background` input of
* a call that did not fail.
*/
export function backgroundedBashToolCallIds(
content: unknown,
toolUses: Record<string, { name: string; input: unknown }>,
runtime: AsyncTaskRuntime,
structuredTask?: AsyncTaskStarted,
): Set<string> {
const ids = new Set<string>();
if (!runtime.enabled || !Array.isArray(content)) return ids;
const structuredToolCallId = structuredTask
? nonBlankString(field(structuredTask, "toolCallId", "tool_use_id"))
: undefined;
for (const result of content.flatMap(toolResultBlock)) {
const toolUse = toolUses[result.toolUseId];
if (toolUse?.name !== "Bash") continue;
const input = isRecord(toolUse.input) ? toolUse.input : undefined;
if (
result.toolUseId === structuredToolCallId ||
runtime.isBackgroundedToolCall(result.toolUseId) ||
(input?.run_in_background === true && !result.isError)
) {
ids.add(result.toolUseId);
}
}
return ids;
}
function structuredBackgroundResults(
value: unknown,
toolUses: Record<string, { name: string; input: unknown }>,
hintedToolUseId?: string,
seen = new Set<unknown>(),
): { taskId: string; toolUseId?: string }[] {
if (typeof value !== "object" || value === null || seen.has(value)) return [];
seen.add(value);
if (Array.isArray(value)) {
return value.flatMap((item) =>
structuredBackgroundResults(item, toolUses, hintedToolUseId, seen),
);
}
const directTaskId = nonBlankString(field(value, "backgroundTaskId", "background_task_id"));
const directToolUseId =
nonBlankString(field(value, "toolUseId", "tool_use_id")) ?? hintedToolUseId;
const results: { taskId: string; toolUseId?: string }[] = directTaskId
? [{ taskId: directTaskId, toolUseId: directToolUseId }]
: [];
for (const [key, child] of Object.entries(value)) {
if (key === "backgroundTaskId" || key === "background_task_id") continue;
results.push(
...structuredBackgroundResults(child, toolUses, toolUses[key] ? key : directToolUseId, seen),
);
}
return results;
}
function toolResultBlock(
value: unknown,
): { toolUseId: string; content?: unknown; isError: boolean }[] {
if (!isRecord(value) || value.type !== "tool_result") return [];
const toolUseId = nonBlankString(field(value, "toolUseId", "tool_use_id"));
return toolUseId
? [
{
toolUseId,
content: value.content,
isError: field(value, "isError", "is_error") === true,
},
]
: [];
}
function asyncTaskOutputFilePath(content: unknown, taskId: string): string | undefined {
const text = textContent(content);
return text ? outputFilePathOf(taskOutputPaths(text), taskId) : undefined;
}
/**
* The end of a task output path: `tasks/<id>.output`, followed by a sentence
* period, whitespace or the end of the text. Claude Code writes the output of
* a background task there.
*/
const TASK_OUTPUT_SUFFIX = /[/\\]tasks[/\\][^\s/\\]+?\.output(?=\.(?!\w)|\s|$)/g;
/** An absolute POSIX, Windows drive or UNC path. */
const ABSOLUTE_PATH = /^(?:\/|[A-Za-z]:[\\/]|\\\\)/;
/**
* The task output paths in a tool result text. Each text of Claude Code puts
* the path after "Output is being written to: ", so a path starts after the
* nearest `": "` before its end. A drive colon has no space after it, so a
* path can hold spaces and a drive. The search for the start stops at a line
* break and at the end of the previous path end, so a path never crosses a
* line or holds another path. Each character is read at most once, so the
* parse is linear in the text length. A path that is not absolute does not
* count.
*/
function taskOutputPaths(text: string): string[] {
const paths: string[] = [];
let previousEnd = 0;
for (const match of text.matchAll(TASK_OUTPUT_SUFFIX)) {
const end = match.index + match[0].length;
let start = -1;
for (let i = match.index - 1; i >= previousEnd; i--) {
const char = text[i];
if (char === "\n" || char === "\r") break;
if (char === ":" && text[i + 1] === " ") {
start = i + 2;
break;
}
}
previousEnd = end;
if (start < 0) continue;
const path = text.slice(start, end);
if (ABSOLUTE_PATH.test(path)) paths.push(path);
}
return paths;
}
/** The output path of the task `taskId` among the paths, if one names it. */
function outputFilePathOf(paths: readonly string[], taskId: string): string | undefined {
const posix = `/tasks/${taskId}.output`;
const windows = `\\tasks\\${taskId}.output`;
return paths.find((path) => path.endsWith(posix) || path.endsWith(windows));
}
function textContent(value: unknown): string | undefined {
if (typeof value === "string") return value;
if (Array.isArray(value)) {
const parts = value.map(textContent).filter((part): part is string => part !== undefined);
return parts.length > 0 ? parts.join("") : undefined;
}
if (!isRecord(value)) return undefined;
if (typeof value.text === "string") return value.text;
return value.content === value ? undefined : textContent(value.content);
}
function friendlyTaskType(value: unknown): string {
if (value === "local_bash") return "shell";
if (value === "local_workflow") return "workflow";
if (value === "local_monitor" || value === "mcp") return "monitor";
return nonBlankString(value) ?? "task";
}
function defaultDescription(taskType: string): string {
return taskType === "task" ? "Background task" : taskType[0].toUpperCase() + taskType.slice(1);
}
function isBackgroundTask(isBackgrounded: unknown, taskType: unknown): boolean {
if (isBackgrounded === true) return true;
if (isBackgrounded === false) return false;
return taskType !== "local_bash" && taskType !== "local_agent";
}
function isSubagentTask(message: AsyncTaskStarted): boolean {
return (
field(message, "subagentType", "subagent_type") !== undefined ||
field(message, "taskType", "task_type") === "local_agent"
);
}
function taskState(value: unknown): AsyncTaskState | undefined {
if (value === "pending" || value === "running") return "running";
if (value === "paused") return "paused";
if (value === "completed") return "completed";
if (value === "failed") return "failed";
if (value === "killed" || value === "cancelled" || value === "stopped") return "stopped";
return undefined;
}
function isTerminal(state: AsyncTaskState): state is "completed" | "failed" | "stopped" {
return state === "completed" || state === "failed" || state === "stopped";
}
function taskUsage(
value: unknown,
): { totalTokens: number; toolUses: number; durationMs: number } | undefined {
if (!isRecord(value)) return undefined;
const totalTokens = numberField(value, "totalTokens", "total_tokens");
const toolUses = numberField(value, "toolUses", "tool_uses");
const durationMs = numberField(value, "durationMs", "duration_ms");
if (totalTokens === undefined || toolUses === undefined || durationMs === undefined)
return undefined;
return { totalTokens, toolUses, durationMs };
}
function taskIdOf(value: TaskIdentity): string | undefined {
return nonBlankString(field(value, "taskId", "task_id"));
}
function field(value: object, camelCase: string, snakeCase: string): unknown {
const record = value as Record<string, unknown>;
return record[camelCase] !== undefined ? record[camelCase] : record[snakeCase];
}
function numberField(
value: Record<string, unknown>,
camelCase: string,
snakeCase: string,
): number | undefined {
const result = field(value, camelCase, snakeCase);
return typeof result === "number" ? result : undefined;
}
function nonBlankString(value: unknown): string | undefined {
return typeof value === "string" && value.trim().length > 0 ? value.trim() : undefined;
}
function isRecord(value: unknown): value is Record<string, unknown> {
return typeof value === "object" && value !== null && !Array.isArray(value);
}