mirror of
https://github.com/agentclientprotocol/claude-agent-acp.git
synced 2026-10-27 04:30:59 +00:00
fix: IJAI-3744 keep a session running while its background tasks are open (#1399)
## Summary A Claude Code session can start a background shell command or a background sub-agent, and then end its turn. When a task ends, Claude Code wakes the session for a short turn. The session index reported `idle` at each turn end, so a client saw one completion for each task. Now the index finds the open background tasks in the transcript. A turn end with an open task stays `running`, also after 10 minutes. The session gets one `idle`, when the last task ends. The rules are the same as in the native Claude provider of AIR (IJAI-3744). ## Rules - A task opens with a tool result that has `backgroundTaskId`, or `isAsync` with an `agentId`. - A task closes with its `<task-notification>` (a user record or a queue `remove`), or with a `TaskStop` result. - The wait ends when no live process holds the session, or when the registry status is `idle`. - A task can start before the tail window. A later scan applies the tail on top of the earlier task set. When it cannot, the index reads the whole transcript, but only for a finished turn of a live session. ## Not in this PR - A background task that a sub-agent starts. Its start record is only in the transcript of the sub-agent. - A session that this connection runs. Its state comes from the SDK. ## Tests - `background-tasks.test.ts`: the record orders of the native provider tests, in a small transcript and in one longer than the tail window. The tests also cover a task before the tail window, the end of the wait, and the count of whole-file reads. - A real check with Claude Code 2.1.306 in a terminal. With two background shells or two background sub-agents, `main` gave three `idle` and this change gave one. --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Ilya Kirillov
@@ -1370,6 +1370,7 @@ Each row carries `_meta.jetbrains.air` with flat row fields of the RFDs. `archiv
|
||||
- a session this connection runs: the SDK `session_state_changed` state; `error` instead of `idle` when its last turn failed (an error result, `is_error` or an `error_*` subtype, that was not a cancel). A session whose query failed here (not a cancel) is reported as any other session, but `error` until its transcript shows a prompt or a turn end after the failure, unless a busy or waiting interactive CLI holds it;
|
||||
- a session no live Claude Code process holds: `idle`, or `error` when its last turn ended with an API error (an assistant record with `isApiErrorMessage: true`, also when the CLI's turn-end records follow it, with no prompt after it);
|
||||
- a session that an interactive CLI holds (not an `sdk-*` entrypoint) whose registry status is newer than the transcript: `busy` and `shell` are `running`, `waiting` is `requires_action`, `idle` is `idle`, or `error` after such an API error;
|
||||
- any other held session whose last turn ended with an open background task: `running`, also after 10 minutes, until the registry status is `idle`. Claude Code wakes the session again when the task ends, so the session gets one `idle` at the last turn end, not one for each task. A background shell command (`toolUseResult.backgroundTaskId`) or an async sub-agent (`isAsync` with an `agentId`) opens a task. Its `<task-notification>` with a `<status>`, as a user record or a `queue-operation` `remove`, or the result of a `TaskStop` call closes it. A notification without a `<status>`, such as the check-in of a quiet shell or a `Monitor` event, keeps it open. Sidechain records open no task. The tail gives the tasks on top of those of an earlier scan when the file only grew within the tail; else the whole transcript is read, and only for such a held session. A background task that a sub-agent starts is not seen: it is only in the transcript of the sub-agent;
|
||||
- any other held session: `idle` after a finished turn (an assistant `end_turn`, a user interrupt, or the CLI turn-end records), `error` after a turn that ended with an API error, `running` for an unfinished turn written in the last 10 minutes, and no state otherwise.
|
||||
- The live processes come from `<config>/sessions/<pid>.json`, with the liveness rules of Claude Desktop: `kill(pid, 0)` (or `EPERM`), a matching `procStart`, no foreign `pidDomain`, and records older than 24 hours only with a matching `procStart`.
|
||||
The adapter never reads the `.key` files there and never changes the directory.
|
||||
|
||||
@@ -316,6 +316,12 @@ import {
|
||||
} from "./exit-plan.js";
|
||||
import { DEFAULT_AGENT_ID } from "./session-config-ids.js";
|
||||
import { parseToolResultMeta } from "./tool-result-meta.js";
|
||||
import {
|
||||
markerTagText,
|
||||
type PersistedTaskNotification,
|
||||
TASK_NOTIFICATION_TAG,
|
||||
taskNotificationsOf,
|
||||
} from "./task-notifications.js";
|
||||
import { formatUsageResponse, isUsageCommandText, parseUsageResponse } from "./usage-markdown.js";
|
||||
|
||||
export { DEFAULT_AGENT_ID } from "./session-config-ids.js";
|
||||
@@ -2123,7 +2129,6 @@ const LOCAL_COMMAND_MARKERS = [
|
||||
// model that a background task stopped; live, the SDK `task_notification`
|
||||
// frame reports the same stop, so replay restores the task state from it (see
|
||||
// taskNotificationsOf) instead of showing it.
|
||||
const TASK_NOTIFICATION_TAG = "task-notification";
|
||||
const INJECTED_CONTEXT_MARKERS = ["system-reminder", TASK_NOTIFICATION_TAG].map((tag) => ({
|
||||
open: `<${tag}>`,
|
||||
close: `</${tag}>`,
|
||||
@@ -2162,16 +2167,6 @@ function stripMarkerTags(text: string): string {
|
||||
return result + text.slice(copiedUpTo);
|
||||
}
|
||||
|
||||
function markerTagText(text: string, tag: string): string | undefined {
|
||||
const open = `<${tag}>`;
|
||||
const close = `</${tag}>`;
|
||||
const start = text.indexOf(open);
|
||||
if (start === -1) return undefined;
|
||||
const end = text.indexOf(close, start + open.length);
|
||||
if (end === -1) return undefined;
|
||||
return text.slice(start + open.length, end);
|
||||
}
|
||||
|
||||
function hasMarkerTag(text: string, tag: string): boolean {
|
||||
return markerTagText(text, tag) !== undefined;
|
||||
}
|
||||
@@ -2232,57 +2227,6 @@ export function isLocalCommandMetadata(content: unknown): boolean {
|
||||
return stripLocalCommandMetadata(content) === null;
|
||||
}
|
||||
|
||||
/** The fields of a persisted `<task-notification>`, in the shape of the SDK
|
||||
* `task_notification` frame that reported the same stop live. */
|
||||
type PersistedTaskNotification = {
|
||||
task_id: string;
|
||||
status: string;
|
||||
tool_use_id?: string;
|
||||
summary?: string;
|
||||
output_file?: string;
|
||||
};
|
||||
|
||||
/**
|
||||
* Returns each `<task-notification>` in user-message content (a string or its
|
||||
* text blocks). The `<result>` body is model context that the live frame does
|
||||
* not carry, so the scan stops at it and a tag quoted in it cannot match.
|
||||
*/
|
||||
function taskNotificationsOf(content: unknown): PersistedTaskNotification[] {
|
||||
const texts =
|
||||
typeof content === "string"
|
||||
? [content]
|
||||
: Array.isArray(content)
|
||||
? content.flatMap((block) =>
|
||||
block?.type === "text" && typeof block.text === "string" ? [block.text as string] : [],
|
||||
)
|
||||
: [];
|
||||
const open = `<${TASK_NOTIFICATION_TAG}>`;
|
||||
const close = `</${TASK_NOTIFICATION_TAG}>`;
|
||||
const notifications: PersistedTaskNotification[] = [];
|
||||
for (const text of texts) {
|
||||
let start = text.indexOf(open);
|
||||
while (start !== -1) {
|
||||
const end = text.indexOf(close, start + open.length);
|
||||
if (end === -1) break;
|
||||
const body = text.slice(start + open.length, end).split("<result>", 1)[0];
|
||||
const field = (tag: string) => markerTagText(body, tag)?.trim() || undefined;
|
||||
const taskId = field("task-id");
|
||||
const status = field("status");
|
||||
if (taskId && status) {
|
||||
notifications.push({
|
||||
task_id: taskId,
|
||||
status,
|
||||
tool_use_id: field("tool-use-id"),
|
||||
summary: field("summary"),
|
||||
output_file: field("output-file"),
|
||||
});
|
||||
}
|
||||
start = text.indexOf(open, end + close.length);
|
||||
}
|
||||
}
|
||||
return notifications;
|
||||
}
|
||||
|
||||
/**
|
||||
* True for a user record that the CLI wrote to tell the model that a
|
||||
* background task stopped. A `subkind` marks the delivery of a prompt, for
|
||||
|
||||
@@ -11,6 +11,10 @@
|
||||
* - an interactive CLI that is not driven by an SDK holds it, and its registry
|
||||
* status is newer than the transcript: the registry status (an idle one is
|
||||
* `error` after such a turn);
|
||||
* - a turn that ended with an open background task is `running`, also after
|
||||
* 10 minutes: Claude Code wakes the session again when the task ends. The
|
||||
* wait ends when the registry status is `idle`, because a task can end
|
||||
* without a transcript record (a stop in the task UI);
|
||||
* - otherwise the transcript tail: a finished turn is `idle` (`error` after
|
||||
* an API error), an unfinished turn written in the last 10 minutes is
|
||||
* `running`, and anything else has no known state.
|
||||
@@ -72,6 +76,14 @@ function usesRegistryStatus(record: LiveRecord, transcriptMtimeMs: number): bool
|
||||
);
|
||||
}
|
||||
|
||||
/** Whether the last turn ended while a background task of the session is
|
||||
* open. */
|
||||
function waitsForBackgroundTasks(facts: TranscriptFacts): boolean {
|
||||
if (facts.turnState !== "finished") return false;
|
||||
const openTasks = facts.backgroundTasks?.open ?? [];
|
||||
return openTasks.length > 0;
|
||||
}
|
||||
|
||||
function iso(ms: number | undefined): string | undefined {
|
||||
return ms === undefined || !Number.isFinite(ms) ? undefined : new Date(ms).toISOString();
|
||||
}
|
||||
@@ -103,6 +115,9 @@ export function deriveActivity(input: {
|
||||
// A busy or waiting CLI wins; an idle one shows how its turn ended.
|
||||
state = registryState(live.status);
|
||||
if (state === "idle" && failed) state = "error";
|
||||
} else if (waitsForBackgroundTasks(facts) && live.status !== "idle") {
|
||||
// Claude Code wakes the session again when an open task ends.
|
||||
state = "running";
|
||||
} else if (failed) {
|
||||
state = "error";
|
||||
} else if (facts.turnState === "finished") {
|
||||
|
||||
223
src/session-index/background-tasks.ts
Normal file
223
src/session-index/background-tasks.ts
Normal file
@@ -0,0 +1,223 @@
|
||||
/**
|
||||
* The background tasks of a session that did not end yet, from its transcript.
|
||||
*
|
||||
* A Claude Code session can start a background shell command or a background
|
||||
* sub-agent and end its turn at once. When the task ends, Claude Code wakes
|
||||
* the session with a `<task-notification>` input for a short turn. So a turn
|
||||
* end with an open task is no end of the work: the session list keeps such a
|
||||
* session `running` (see `activity.ts`).
|
||||
*
|
||||
* The records, as Claude Code 2.1.299 writes them, in the rules of AIR's
|
||||
* native Claude provider:
|
||||
*
|
||||
* - a tool result opens a task when its `toolUseResult` has a
|
||||
* `backgroundTaskId` (a shell command), or `isAsync: true` with an
|
||||
* `agentId` (a sub-agent);
|
||||
* - a task closes when its notification reaches the session: as a user
|
||||
* record, or as a `queue-operation` `remove` when a running turn takes it
|
||||
* in. A notification that waits in the queue (`enqueue`) keeps the task
|
||||
* open, because the session still runs a turn for it;
|
||||
* - only a notification with a `<status>` closes a task. A notification
|
||||
* without one comes while the task still runs: the check-in of a quiet
|
||||
* shell ("is still running after …", since Claude Code 2.1.303) or a
|
||||
* `Monitor` event;
|
||||
* - a `TaskStop` result closes the task of its `toolUseResult.task_id`:
|
||||
* Claude Code writes no notification for it. The result does not name its
|
||||
* tool, so its `tool_use_id` must match a `TaskStop` call. A `task_id` in
|
||||
* the result of another tool keeps the task open;
|
||||
* - a resumed CLI writes a `stopped` notification for each task of the
|
||||
* previous process;
|
||||
* - a sidechain record belongs to a sub-agent and opens or closes no task of
|
||||
* the session.
|
||||
*
|
||||
* A background shell that does not end, for example a dev server, keeps its
|
||||
* task open, so its session stays `running` while the CLI holds it.
|
||||
*
|
||||
* A task that a sub-agent starts is not seen. Its start record is only in the
|
||||
* transcript of the sub-agent, so the session can be `idle` while the
|
||||
* sub-agent waits for it.
|
||||
*/
|
||||
|
||||
import { createReadStream } from "node:fs";
|
||||
import { createInterface } from "node:readline";
|
||||
import { TASK_NOTIFICATION_TAG, taskNotificationsOf } from "../task-notifications.js";
|
||||
|
||||
export type BackgroundTasks = {
|
||||
/** The ids of the tasks that started and did not end, in start order. */
|
||||
open: string[];
|
||||
/** The `tool_use` ids of the `TaskStop` calls without a result. */
|
||||
stopCalls: string[];
|
||||
/** Whether the sets hold the records of the whole file. A set from the
|
||||
* tail window only does not know a task that started before it. */
|
||||
complete: boolean;
|
||||
};
|
||||
|
||||
/** The fields of a transcript record that bear on a task. */
|
||||
type TranscriptRecord = {
|
||||
type?: string;
|
||||
isSidechain?: boolean;
|
||||
/** Of a `queue-operation` record: `enqueue`, `remove`, ... */
|
||||
operation?: string;
|
||||
/** Of a `queue-operation` record: the queued input. */
|
||||
content?: string;
|
||||
message?: { content?: string | ContentBlock[] };
|
||||
toolUseResult?: ToolUseResult;
|
||||
};
|
||||
|
||||
type ContentBlock = {
|
||||
type?: string;
|
||||
text?: string;
|
||||
/** Of a `tool_use` block. */
|
||||
id?: string;
|
||||
name?: string;
|
||||
/** Of a `tool_result` block. */
|
||||
tool_use_id?: string;
|
||||
};
|
||||
|
||||
type ToolUseResult = {
|
||||
/** The task of a background shell command. */
|
||||
backgroundTaskId?: string;
|
||||
/** With `agentId`: an async sub-agent. */
|
||||
isAsync?: boolean;
|
||||
agentId?: string;
|
||||
/** The task that a `TaskStop` call stopped. Another tool can give one. */
|
||||
task_id?: string;
|
||||
};
|
||||
|
||||
const NOTIFICATION_TAG = `<${TASK_NOTIFICATION_TAG}>`;
|
||||
|
||||
/** Text that a line must hold to bear on a task, so that the other lines
|
||||
* are not parsed. */
|
||||
const TASK_MARKERS = ["backgroundTaskId", '"isAsync"', NOTIFICATION_TAG, "TaskStop", "task_id"];
|
||||
|
||||
function parseRecord(line: string): TranscriptRecord | undefined {
|
||||
try {
|
||||
return JSON.parse(line) as TranscriptRecord;
|
||||
} catch {
|
||||
return undefined;
|
||||
}
|
||||
}
|
||||
|
||||
function blocksOf(record: TranscriptRecord): ContentBlock[] {
|
||||
const content = record.message?.content;
|
||||
return Array.isArray(content) ? content : [];
|
||||
}
|
||||
|
||||
/** The input of a user record: its string content, else its first text
|
||||
* block. */
|
||||
function inputTextOf(record: TranscriptRecord): string | undefined {
|
||||
const content = record.message?.content;
|
||||
if (typeof content === "string") return content;
|
||||
return blocksOf(record).find((block) => block.type === "text")?.text;
|
||||
}
|
||||
|
||||
/** The task that a `<task-notification>` input ends: only an input that
|
||||
* starts with the tag, after whitespace, as AIR's `TaskNotification` reads
|
||||
* it. Only a notification with a `<status>` ends a task (see
|
||||
* `taskNotificationsOf`). */
|
||||
export function endedTaskId(input: string | undefined): string | undefined {
|
||||
if (!input?.trimStart().startsWith(NOTIFICATION_TAG)) return undefined;
|
||||
return taskNotificationsOf(input)[0]?.task_id;
|
||||
}
|
||||
|
||||
/** The task that a tool result started: a background shell command, or an
|
||||
* async sub-agent. */
|
||||
function startedTaskId(record: TranscriptRecord): string | undefined {
|
||||
if (record.type !== "user") return undefined;
|
||||
const result = record.toolUseResult;
|
||||
if (result?.backgroundTaskId) return result.backgroundTaskId;
|
||||
if (result?.isAsync) return result.agentId;
|
||||
return undefined;
|
||||
}
|
||||
|
||||
/** The task whose notification reached the session: as a user record, or
|
||||
* as a queue `remove` when a running turn takes it in. */
|
||||
function deliveredTaskId(record: TranscriptRecord): string | undefined {
|
||||
if (record.type === "user") return endedTaskId(inputTextOf(record));
|
||||
if (record.type === "queue-operation" && record.operation === "remove") {
|
||||
return endedTaskId(record.content);
|
||||
}
|
||||
return undefined;
|
||||
}
|
||||
|
||||
/** Collects the tasks of the lines it is given, in file order. */
|
||||
class TaskTracker {
|
||||
private readonly open: Set<string>;
|
||||
private readonly stopCalls: Set<string>;
|
||||
|
||||
constructor(after?: BackgroundTasks) {
|
||||
this.open = new Set(after?.open);
|
||||
this.stopCalls = new Set(after?.stopCalls);
|
||||
}
|
||||
|
||||
add(line: string): void {
|
||||
const mayBearOnTask = TASK_MARKERS.some((marker) => line.includes(marker));
|
||||
if (!mayBearOnTask) return;
|
||||
const record = parseRecord(line);
|
||||
// A sidechain record belongs to a sub-agent, which receives the
|
||||
// notifications of its own tasks.
|
||||
if (!record || record.isSidechain) return;
|
||||
this.apply(record);
|
||||
}
|
||||
|
||||
result({ complete }: { complete: boolean }): BackgroundTasks {
|
||||
return { open: [...this.open], stopCalls: [...this.stopCalls], complete };
|
||||
}
|
||||
|
||||
private apply(record: TranscriptRecord): void {
|
||||
const started = startedTaskId(record);
|
||||
if (started) this.open.add(started);
|
||||
const delivered = deliveredTaskId(record);
|
||||
if (delivered) this.open.delete(delivered);
|
||||
this.applyTaskStop(record);
|
||||
}
|
||||
|
||||
/** A `task_id` stops its task only in the result of a `TaskStop` call:
|
||||
* another tool can give a `task_id` too. */
|
||||
private applyTaskStop(record: TranscriptRecord): void {
|
||||
let isTaskStopResult = false;
|
||||
for (const block of blocksOf(record)) {
|
||||
if (block.type === "tool_use" && block.name === "TaskStop" && block.id) {
|
||||
this.stopCalls.add(block.id);
|
||||
}
|
||||
const resultId = block.tool_use_id;
|
||||
if (resultId && this.stopCalls.has(resultId)) {
|
||||
this.stopCalls.delete(resultId);
|
||||
isTaskStopResult = true;
|
||||
}
|
||||
}
|
||||
const stopped = record.toolUseResult?.task_id;
|
||||
if (isTaskStopResult && stopped) this.open.delete(stopped);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* The tasks of the lines of `text`, applied on the tasks `after`.
|
||||
*
|
||||
* `after` can be what an earlier scan found for a shorter copy of the file,
|
||||
* and `text` can repeat records of that copy. The result is still the same:
|
||||
* the last record that opens or closes a task decides, and `text` holds the
|
||||
* last records.
|
||||
*/
|
||||
export function backgroundTasksOf(
|
||||
text: string,
|
||||
options: { complete: boolean; after?: BackgroundTasks },
|
||||
): BackgroundTasks {
|
||||
const tracker = new TaskTracker(options.after);
|
||||
for (const line of text.split("\n")) tracker.add(line);
|
||||
return tracker.result({ complete: options.complete });
|
||||
}
|
||||
|
||||
/** The tasks of the first `size` bytes of `filePath`, from all its records.
|
||||
* Reads the whole file, so only for a session that a live process holds. */
|
||||
export async function searchBackgroundTasks(
|
||||
filePath: string,
|
||||
size: number,
|
||||
): Promise<BackgroundTasks> {
|
||||
const tracker = new TaskTracker();
|
||||
if (size === 0) return tracker.result({ complete: true });
|
||||
const stream = createReadStream(filePath, { start: 0, end: size - 1, encoding: "utf8" });
|
||||
const lines = createInterface({ input: stream, crlfDelay: Infinity });
|
||||
for await (const line of lines) tracker.add(line);
|
||||
return tracker.result({ complete: true });
|
||||
}
|
||||
@@ -476,12 +476,18 @@ export class SessionIndexService {
|
||||
private readonly deleteSession: (sessionId: string) => Promise<void>;
|
||||
|
||||
constructor(private readonly deps: SessionIndexDeps) {
|
||||
this.index = new SessionIndex();
|
||||
this.registry = deps.registry ?? new LiveSessionRegistry();
|
||||
this.index = new SessionIndex({ isLive: (sessionId) => this.isLive(sessionId) });
|
||||
this.now = deps.now ?? Date.now;
|
||||
this.deleteSession = deps.deleteSession ?? ((id) => sdkDeleteSession(id));
|
||||
}
|
||||
|
||||
/** Whether a live process holds the session, by the live registry. */
|
||||
private async isLive(sessionId: string): Promise<boolean> {
|
||||
const live = await this.registry.snapshot();
|
||||
return live.has(sessionId);
|
||||
}
|
||||
|
||||
/** `session/list` of a `sessionIndex` client. `own` reports the sessions
|
||||
* that this connection runs. */
|
||||
async list(
|
||||
|
||||
@@ -52,6 +52,7 @@ import {
|
||||
type SdkTitles,
|
||||
type TranscriptFacts,
|
||||
} from "./transcript-scan.js";
|
||||
import { searchBackgroundTasks } from "./background-tasks.js";
|
||||
import { DirListings, statFiles } from "./dir-listing.js";
|
||||
import { isInLinkedWorktree, worktreeCounterparts } from "./worktrees.js";
|
||||
|
||||
@@ -400,7 +401,15 @@ function scopeOf(
|
||||
};
|
||||
}
|
||||
|
||||
export type SessionIndexOptions = {
|
||||
/** Whether a live process holds the session. Only for such a session the
|
||||
* whole transcript is searched for its background tasks. */
|
||||
isLive?: (sessionId: string) => Promise<boolean>;
|
||||
};
|
||||
|
||||
export class SessionIndex {
|
||||
constructor(private readonly options: SessionIndexOptions = {}) {}
|
||||
|
||||
private readonly listings = new DirListings();
|
||||
private readonly metadata = new Lru<string, CacheEntry>(METADATA_CACHE_SIZE);
|
||||
/** The metadata reads in flight, by path: a list and a subscription that
|
||||
@@ -992,14 +1001,15 @@ export class SessionIndex {
|
||||
this.reads++;
|
||||
const headTail = await readHeadTail(candidate.filePath, candidate.size);
|
||||
if (isSidechainTranscript(headTail.head)) return null;
|
||||
const facts = await scanTranscriptFile(
|
||||
const scanned = await scanTranscriptFile(
|
||||
candidate.filePath,
|
||||
candidate.size,
|
||||
headTail,
|
||||
previous,
|
||||
candidate.ino,
|
||||
);
|
||||
if (!facts.hasMessages) return null;
|
||||
if (!scanned.hasMessages) return null;
|
||||
const facts = await this.withAllBackgroundTasks(candidate, scanned);
|
||||
// The last relocation names the session's cwd, as the SDK reads it; the
|
||||
// messages before it keep the old one.
|
||||
const fileCwd = await this.recoverCwd(candidate.dirName, [
|
||||
@@ -1032,6 +1042,24 @@ export class SessionIndex {
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* `facts` with the background tasks of the whole transcript, when the
|
||||
* tasks of its tail are not complete: a task can start before the tail.
|
||||
* Only a finished turn of a live session waits for the tasks, so only
|
||||
* then the whole transcript is read.
|
||||
*/
|
||||
private async withAllBackgroundTasks(
|
||||
candidate: TranscriptCandidate,
|
||||
facts: TranscriptFacts,
|
||||
): Promise<TranscriptFacts> {
|
||||
if (facts.turnState !== "finished") return facts;
|
||||
if (facts.backgroundTasks?.complete) return facts;
|
||||
const isLive = await this.options.isLive?.(candidate.sessionId);
|
||||
if (!isLive) return facts;
|
||||
const backgroundTasks = await searchBackgroundTasks(candidate.filePath, candidate.size);
|
||||
return { ...facts, backgroundTasks };
|
||||
}
|
||||
|
||||
/**
|
||||
* The titles the SDK gives the listed transcript (`customTitle` and
|
||||
* `summary` of its `getSessionInfo`), by the SDK's rule from the head and
|
||||
|
||||
@@ -7,9 +7,13 @@
|
||||
*
|
||||
* A last message longer than the tail window leaves no message in it: then
|
||||
* the tail window grows, up to {@link MAX_TAIL_SIZE}, until it holds one.
|
||||
*
|
||||
* The open background tasks come from the tail, on top of the tasks of an
|
||||
* earlier scan (see `background-tasks.ts`).
|
||||
*/
|
||||
|
||||
import * as fs from "node:fs/promises";
|
||||
import { backgroundTasksOf, type BackgroundTasks } from "./background-tasks.js";
|
||||
import { firstPrompt, isUserPrompt, mediaPrompt, promptOf } from "./first-prompt.js";
|
||||
|
||||
const CHUNK_SIZE = 64 * 1024;
|
||||
@@ -56,6 +60,9 @@ export type TranscriptFacts = {
|
||||
* worktree and the session is in it), false when it says otherwise.
|
||||
* Undefined when no such record was read. See {@link agentWorktreeOf}. */
|
||||
agentWorktree?: boolean;
|
||||
/** The background tasks that did not end. From the tail, on top of what
|
||||
* an earlier scan found when the file only grew within the tail. */
|
||||
backgroundTasks?: BackgroundTasks;
|
||||
/** Whether the head or the tail holds a user or an assistant message. A
|
||||
* transcript without one is a metadata-only stub. */
|
||||
hasMessages: boolean;
|
||||
@@ -77,7 +84,7 @@ const INHERITED_FACTS = ["model", "tailCwd", "lastTurnEndedAt", "agentWorktree"]
|
||||
/** What an earlier scan of the same file found, at `size`. */
|
||||
export type PreviousScan = { size: number } & Pick<
|
||||
TranscriptFacts,
|
||||
"lastPromptAt" | "promptSearched" | (typeof INHERITED_FACTS)[number]
|
||||
"lastPromptAt" | "promptSearched" | "backgroundTasks" | (typeof INHERITED_FACTS)[number]
|
||||
>;
|
||||
|
||||
/** The last `window` bytes of a file of `size` bytes, without the first,
|
||||
@@ -157,6 +164,8 @@ export async function scanTranscriptFile(
|
||||
}
|
||||
}
|
||||
}
|
||||
const previousTasks = grownWithinTail ? previous.backgroundTasks : undefined;
|
||||
facts.backgroundTasks = tailBackgroundTasks(read.tail, size, previousTasks);
|
||||
// No worktree record in the tail (or before it, by an earlier scan): the
|
||||
// one `claude --worktree` writes at the start may still be in the head.
|
||||
if (facts.agentWorktree === undefined) {
|
||||
@@ -174,7 +183,8 @@ export async function scanTranscriptFile(
|
||||
if (!facts.hasMessages || complete(facts) || size <= CHUNK_SIZE) return facts;
|
||||
const key = `${filePath}\0${size}\0${identity ?? ""}`;
|
||||
const searched = fullySearched.get(key);
|
||||
if (searched) return searched;
|
||||
// The tasks can differ by the earlier scan.
|
||||
if (searched) return { ...searched, backgroundTasks: facts.backgroundTasks };
|
||||
let result = facts;
|
||||
const handle = await fs.open(filePath, "r");
|
||||
try {
|
||||
@@ -218,6 +228,7 @@ export async function scanTranscriptFile(
|
||||
// A turn end the wider tail lacks may lie before it: the
|
||||
// narrow scan inherited it from an earlier scan of this file.
|
||||
lastTurnEndedAt: wider.lastTurnEndedAt ?? facts.lastTurnEndedAt,
|
||||
backgroundTasks: facts.backgroundTasks,
|
||||
hasMessages: facts.hasMessages,
|
||||
lastPromptAt,
|
||||
// Searched to the start of the file, or found.
|
||||
@@ -237,6 +248,22 @@ export async function scanTranscriptFile(
|
||||
return result;
|
||||
}
|
||||
|
||||
/**
|
||||
* The background tasks of the tail of a file of `size` bytes. `previous`:
|
||||
* the tasks of an earlier scan, when the file only grew within the tail
|
||||
* since. The set is complete when the tail is the whole file, or when it
|
||||
* holds all records after a complete earlier set.
|
||||
*/
|
||||
function tailBackgroundTasks(
|
||||
tail: string,
|
||||
size: number,
|
||||
previous: BackgroundTasks | undefined,
|
||||
): BackgroundTasks {
|
||||
if (size <= CHUNK_SIZE) return backgroundTasksOf(tail, { complete: true });
|
||||
if (previous?.complete) return backgroundTasksOf(tail, { complete: true, after: previous });
|
||||
return backgroundTasksOf(tail, { complete: false });
|
||||
}
|
||||
|
||||
const RELOCATED_MARKER = '"relocated"';
|
||||
|
||||
/** The cwd of the last `relocated` record of the tail: where the session
|
||||
|
||||
75
src/task-notifications.ts
Normal file
75
src/task-notifications.ts
Normal file
@@ -0,0 +1,75 @@
|
||||
/**
|
||||
* The `<task-notification>` inputs that Claude Code writes to a transcript
|
||||
* when a background task stops. The replay of a session (`acp-agent.ts`) and
|
||||
* the session index (`session-index/background-tasks.ts`) read them here, so
|
||||
* both read the same fields.
|
||||
*/
|
||||
|
||||
export const TASK_NOTIFICATION_TAG = "task-notification";
|
||||
|
||||
/** The text between the first `<tag>` and the next `</tag>` after it. */
|
||||
export function markerTagText(text: string, tag: string): string | undefined {
|
||||
const open = `<${tag}>`;
|
||||
const close = `</${tag}>`;
|
||||
const start = text.indexOf(open);
|
||||
if (start === -1) return undefined;
|
||||
const end = text.indexOf(close, start + open.length);
|
||||
if (end === -1) return undefined;
|
||||
return text.slice(start + open.length, end);
|
||||
}
|
||||
|
||||
/** The fields of a persisted `<task-notification>`, in the shape of the SDK
|
||||
* `task_notification` frame that reported the same stop live. */
|
||||
export type PersistedTaskNotification = {
|
||||
task_id: string;
|
||||
status: string;
|
||||
tool_use_id?: string;
|
||||
summary?: string;
|
||||
output_file?: string;
|
||||
};
|
||||
|
||||
/**
|
||||
* Returns each `<task-notification>` in user-message content (a string or its
|
||||
* text blocks). The `<result>` body is model context that the live frame does
|
||||
* not carry, so the scan stops at it and a tag quoted in it cannot match.
|
||||
*
|
||||
* Only a notification with a `<task-id>` and a `<status>` is returned. A
|
||||
* notification without a status comes while its task still runs: the
|
||||
* check-in of a quiet shell ("is still running after …") or a `Monitor`
|
||||
* event.
|
||||
*/
|
||||
export function taskNotificationsOf(content: unknown): PersistedTaskNotification[] {
|
||||
const texts =
|
||||
typeof content === "string"
|
||||
? [content]
|
||||
: Array.isArray(content)
|
||||
? content.flatMap((block) =>
|
||||
block?.type === "text" && typeof block.text === "string" ? [block.text as string] : [],
|
||||
)
|
||||
: [];
|
||||
const open = `<${TASK_NOTIFICATION_TAG}>`;
|
||||
const close = `</${TASK_NOTIFICATION_TAG}>`;
|
||||
const notifications: PersistedTaskNotification[] = [];
|
||||
for (const text of texts) {
|
||||
let start = text.indexOf(open);
|
||||
while (start !== -1) {
|
||||
const end = text.indexOf(close, start + open.length);
|
||||
if (end === -1) break;
|
||||
const body = text.slice(start + open.length, end).split("<result>", 1)[0];
|
||||
const field = (tag: string) => markerTagText(body, tag)?.trim() || undefined;
|
||||
const taskId = field("task-id");
|
||||
const status = field("status");
|
||||
if (taskId && status) {
|
||||
notifications.push({
|
||||
task_id: taskId,
|
||||
status,
|
||||
tool_use_id: field("tool-use-id"),
|
||||
summary: field("summary"),
|
||||
output_file: field("output-file"),
|
||||
});
|
||||
}
|
||||
start = text.indexOf(open, end + close.length);
|
||||
}
|
||||
}
|
||||
return notifications;
|
||||
}
|
||||
587
src/tests/background-tasks.test.ts
Normal file
587
src/tests/background-tasks.test.ts
Normal file
@@ -0,0 +1,587 @@
|
||||
/**
|
||||
* The `state` of a session list row while a background task of the session
|
||||
* is open, against a real projects directory and live registry in a
|
||||
* temporary `CLAUDE_CONFIG_DIR`. The record orders are those of AIR's native
|
||||
* Claude provider tests (`ClaudeThreadsStoreTest`), from Claude Code 2.1.299.
|
||||
*/
|
||||
import { afterEach, beforeEach, describe, expect, it } from "vitest";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import * as fs from "node:fs/promises";
|
||||
import * as fsSync from "node:fs";
|
||||
import * as os from "node:os";
|
||||
import * as path from "node:path";
|
||||
import type { SessionInfo } from "@agentclientprotocol/sdk/unstable";
|
||||
import type { ActivityState } from "../session-index/activity.js";
|
||||
import { endedTaskId, searchBackgroundTasks } from "../session-index/background-tasks.js";
|
||||
import { encodeProjectPath } from "../session-index/project-dirs.js";
|
||||
import { SessionIndex } from "../session-index/session-index.js";
|
||||
import { SessionIndexService } from "../session-index/service.js";
|
||||
|
||||
let configDir: string;
|
||||
let workspace: string;
|
||||
let previousConfigDir: string | undefined;
|
||||
let sessionId: string;
|
||||
let transcript: string;
|
||||
|
||||
beforeEach(async () => {
|
||||
previousConfigDir = process.env.CLAUDE_CONFIG_DIR;
|
||||
configDir = fsSync.realpathSync(await fs.mkdtemp(path.join(os.tmpdir(), "bg-tasks-cfg-")));
|
||||
workspace = fsSync.realpathSync(await fs.mkdtemp(path.join(os.tmpdir(), "bg-tasks-ws-")));
|
||||
process.env.CLAUDE_CONFIG_DIR = configDir;
|
||||
sessionId = randomUUID();
|
||||
const dir = path.join(configDir, "projects", encodeProjectPath(workspace));
|
||||
await fs.mkdir(dir, { recursive: true });
|
||||
transcript = path.join(dir, `${sessionId}.jsonl`);
|
||||
});
|
||||
|
||||
afterEach(async () => {
|
||||
if (previousConfigDir === undefined) delete process.env.CLAUDE_CONFIG_DIR;
|
||||
else process.env.CLAUDE_CONFIG_DIR = previousConfigDir;
|
||||
await fs.rm(configDir, { recursive: true, force: true });
|
||||
await fs.rm(workspace, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
type JsonValue = string | number | boolean | null | JsonValue[] | { [key: string]: JsonValue };
|
||||
/** The fields of a transcript record. */
|
||||
type RecordFields = { [key: string]: JsonValue };
|
||||
|
||||
/** The fields that `sessionInfoOf` writes in the `_meta` of a row. The SDK
|
||||
* types `_meta` as a map of any values. */
|
||||
type RowMeta = { jetbrains?: { air?: { state?: ActivityState } } };
|
||||
|
||||
const record = (timestamp: string, fields: RecordFields) =>
|
||||
JSON.stringify({ sessionId, cwd: workspace, isSidechain: false, timestamp, ...fields });
|
||||
|
||||
const user = (timestamp: string, text = "Run it") =>
|
||||
record(timestamp, { type: "user", message: { role: "user", content: text } });
|
||||
|
||||
const assistant = (timestamp: string, text = "Done") =>
|
||||
record(timestamp, {
|
||||
type: "assistant",
|
||||
message: { role: "assistant", content: [{ type: "text", text }], stop_reason: "end_turn" },
|
||||
});
|
||||
|
||||
const toolUse = (timestamp: string, id = "tu_1", name = "Bash") =>
|
||||
record(timestamp, {
|
||||
type: "assistant",
|
||||
message: {
|
||||
role: "assistant",
|
||||
content: [{ type: "tool_use", id, name, input: {} }],
|
||||
stop_reason: "tool_use",
|
||||
},
|
||||
});
|
||||
|
||||
const backgroundShellResult = (timestamp: string, taskId: string, extra: RecordFields = {}) =>
|
||||
record(timestamp, {
|
||||
...extra,
|
||||
type: "user",
|
||||
message: {
|
||||
role: "user",
|
||||
content: [
|
||||
{
|
||||
tool_use_id: "tu_1",
|
||||
type: "tool_result",
|
||||
content: `Command running in background with ID: ${taskId}`,
|
||||
is_error: false,
|
||||
},
|
||||
],
|
||||
},
|
||||
toolUseResult: { backgroundTaskId: taskId, stdout: "", stderr: "", interrupted: false },
|
||||
});
|
||||
|
||||
const asyncAgentResult = (timestamp: string, agentId: string, toolUseId: string) =>
|
||||
record(timestamp, {
|
||||
type: "user",
|
||||
message: {
|
||||
role: "user",
|
||||
content: [
|
||||
{
|
||||
tool_use_id: toolUseId,
|
||||
type: "tool_result",
|
||||
content: [{ type: "text", text: "Async agent launched successfully." }],
|
||||
},
|
||||
],
|
||||
},
|
||||
toolUseResult: {
|
||||
isAsync: true,
|
||||
status: "async_launched",
|
||||
agentId,
|
||||
description: "Background work",
|
||||
prompt: "Work",
|
||||
outputFile: `/tmp/${agentId}.output`,
|
||||
canReadOutputFile: true,
|
||||
},
|
||||
});
|
||||
|
||||
const notificationText = (taskId: string, status = "completed") =>
|
||||
`<task-notification>\n<task-id>${taskId}</task-id>\n<status>${status}</status>\n</task-notification>`;
|
||||
|
||||
const notification = (timestamp: string, taskId: string, status = "completed") =>
|
||||
record(timestamp, {
|
||||
type: "user",
|
||||
message: { role: "user", content: notificationText(taskId, status) },
|
||||
});
|
||||
|
||||
const queueOperation = (
|
||||
timestamp: string,
|
||||
taskId: string,
|
||||
operation = "enqueue",
|
||||
content = notificationText(taskId),
|
||||
) => record(timestamp, { type: "queue-operation", operation, content });
|
||||
|
||||
/** The notification that Claude Code 2.1.303 sends for a quiet background
|
||||
* shell that still runs. It has no `<status>`. */
|
||||
const checkInText = (taskId: string) =>
|
||||
`<task-notification>\n<task-id>${taskId}</task-id>\n<tool-use-id>tu_1</tool-use-id>\n` +
|
||||
`<output-file>/tmp/${taskId}.output</output-file>\n` +
|
||||
`<summary>Background command "npm run dev" is still running after 10m, with no new output for 10m</summary>\n` +
|
||||
`</task-notification>\nThis is a check-in, not a completion.`;
|
||||
|
||||
const checkIn = (timestamp: string, taskId: string) =>
|
||||
record(timestamp, { type: "user", message: { role: "user", content: checkInText(taskId) } });
|
||||
|
||||
const taskStopResult = (timestamp: string, taskId: string, toolUseId = "tu_stop") =>
|
||||
record(timestamp, {
|
||||
type: "user",
|
||||
message: {
|
||||
role: "user",
|
||||
content: [
|
||||
{
|
||||
tool_use_id: toolUseId,
|
||||
type: "tool_result",
|
||||
content: `Successfully stopped task: ${taskId}`,
|
||||
},
|
||||
],
|
||||
},
|
||||
toolUseResult: {
|
||||
message: `Successfully stopped task: ${taskId} (sleep 300)`,
|
||||
task_id: taskId,
|
||||
task_type: "local_bash",
|
||||
command: "sleep 300",
|
||||
},
|
||||
});
|
||||
|
||||
/** A finished turn with a long answer: the records after it fit in the tail
|
||||
* window, the records before it do not. */
|
||||
const padding = (timestamp: string) => [
|
||||
user(timestamp, "Explain it"),
|
||||
assistant(timestamp, "x".repeat(70 * 1024)),
|
||||
];
|
||||
|
||||
/** Registers this process as a live interactive CLI that holds the session.
|
||||
* Its status is older than the transcript, so the transcript decides. */
|
||||
async function registerHolder(status = "busy") {
|
||||
await fs.mkdir(path.join(configDir, "sessions"), { recursive: true });
|
||||
await fs.writeFile(
|
||||
path.join(configDir, "sessions", `${process.pid}.json`),
|
||||
JSON.stringify({
|
||||
pid: process.pid,
|
||||
sessionId,
|
||||
kind: "interactive",
|
||||
entrypoint: "cli",
|
||||
status,
|
||||
statusUpdatedAt: 1,
|
||||
updatedAt: Date.now(),
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
const toJsonl = (lines: string[]) => lines.map((line) => `${line}\n`).join("");
|
||||
|
||||
const newService = () => new SessionIndexService({ logError: () => {} });
|
||||
|
||||
function rowState(row: SessionInfo | undefined): ActivityState | undefined {
|
||||
const meta = row?._meta as RowMeta | null | undefined;
|
||||
return meta?.jetbrains?.air?.state;
|
||||
}
|
||||
|
||||
async function stateOf(service: SessionIndexService): Promise<ActivityState | undefined> {
|
||||
const { sessions } = await service.list({ cwd: workspace }, () => undefined);
|
||||
const row = sessions.find((session) => session.sessionId === sessionId);
|
||||
return rowState(row);
|
||||
}
|
||||
|
||||
async function stateWithNewService(): Promise<ActivityState | undefined> {
|
||||
const service = newService();
|
||||
try {
|
||||
return await stateOf(service);
|
||||
} finally {
|
||||
service.dispose();
|
||||
}
|
||||
}
|
||||
|
||||
/** The background tasks in the facts of the only row, by an index that
|
||||
* takes the session as live or not. */
|
||||
async function listedBackgroundTasks(isLive: boolean) {
|
||||
const index = new SessionIndex({ isLive: async () => isLive });
|
||||
const { rows } = await index.list({ cwd: workspace, limit: 10, archived: "unarchived" });
|
||||
return rows[0]?.facts.backgroundTasks;
|
||||
}
|
||||
|
||||
type Step = [lines: string[], state: ActivityState];
|
||||
|
||||
/**
|
||||
* Appends the lines of each step to the transcript and checks the state of
|
||||
* the row after the step. One service lists the session after each step,
|
||||
* so its scan takes over what the scan of the step before found. A new
|
||||
* service gives the same state from a first scan.
|
||||
*/
|
||||
async function expectStateAfterEachStep(...steps: Step[]) {
|
||||
await registerHolder();
|
||||
const service = newService();
|
||||
try {
|
||||
for (const [index, [lines, expected]] of steps.entries()) {
|
||||
await fs.appendFile(transcript, toJsonl(lines));
|
||||
const takenOver = await stateOf(service);
|
||||
expect(takenOver, `taken over after step ${index}`).toBe(expected);
|
||||
const firstScan = await stateWithNewService();
|
||||
expect(firstScan, `first scan after step ${index}`).toBe(expected);
|
||||
}
|
||||
} finally {
|
||||
service.dispose();
|
||||
}
|
||||
}
|
||||
|
||||
/** The record orders of AIR's native Claude provider: each runs in a small
|
||||
* transcript, and behind a padding that pushes the transcript past the tail
|
||||
* window. */
|
||||
describe.each([
|
||||
["a small transcript", [] as string[]],
|
||||
["a transcript longer than the tail window", padding("2026-02-08T00:59:00.000Z")],
|
||||
])("background tasks in %s", (_, before) => {
|
||||
// IJAI-3744
|
||||
it("keeps two background sub-agents running until the last notification reaches the session", async () => {
|
||||
await expectStateAfterEachStep(
|
||||
[
|
||||
[
|
||||
...before,
|
||||
user("2026-02-08T01:00:00.000Z", "Run two agents in the background"),
|
||||
toolUse("2026-02-08T01:00:01.000Z", "tu_a", "Agent"),
|
||||
asyncAgentResult("2026-02-08T01:00:01.100Z", "agent-a", "tu_a"),
|
||||
toolUse("2026-02-08T01:00:01.200Z", "tu_b", "Agent"),
|
||||
asyncAgentResult("2026-02-08T01:00:01.300Z", "agent-b", "tu_b"),
|
||||
assistant("2026-02-08T01:00:02.000Z", "launched"),
|
||||
],
|
||||
"running",
|
||||
],
|
||||
// The notification of agent B waits in the queue until the session takes it.
|
||||
[[queueOperation("2026-02-08T01:00:05.000Z", "agent-b")], "running"],
|
||||
[
|
||||
[
|
||||
notification("2026-02-08T01:00:05.100Z", "agent-b"),
|
||||
assistant("2026-02-08T01:00:06.000Z", "B finished, A still runs"),
|
||||
],
|
||||
"running",
|
||||
],
|
||||
[
|
||||
[
|
||||
queueOperation("2026-02-08T01:00:20.000Z", "agent-a"),
|
||||
notification("2026-02-08T01:00:20.100Z", "agent-a"),
|
||||
assistant("2026-02-08T01:00:21.000Z", "All done"),
|
||||
],
|
||||
"idle",
|
||||
],
|
||||
);
|
||||
});
|
||||
|
||||
// IJAI-3744
|
||||
it("closes a task whose notification a running turn takes in", async () => {
|
||||
await expectStateAfterEachStep([
|
||||
[
|
||||
...before,
|
||||
user("2026-02-08T01:00:00.000Z"),
|
||||
toolUse("2026-02-08T01:00:01.000Z"),
|
||||
backgroundShellResult("2026-02-08T01:00:02.000Z", "task-1"),
|
||||
queueOperation("2026-02-08T01:00:03.000Z", "task-1"),
|
||||
queueOperation("2026-02-08T01:00:04.000Z", "task-1", "remove"),
|
||||
assistant("2026-02-08T01:00:05.000Z"),
|
||||
],
|
||||
"idle",
|
||||
]);
|
||||
});
|
||||
|
||||
// IJAI-3744
|
||||
it("keeps a quiet shell open at its check-ins", async () => {
|
||||
await expectStateAfterEachStep(
|
||||
[
|
||||
[
|
||||
...before,
|
||||
user("2026-02-08T01:00:00.000Z", "Start the dev server"),
|
||||
toolUse("2026-02-08T01:00:01.000Z"),
|
||||
backgroundShellResult("2026-02-08T01:00:02.000Z", "task-1"),
|
||||
assistant("2026-02-08T01:00:03.000Z", "Started"),
|
||||
],
|
||||
"running",
|
||||
],
|
||||
// The check-in wakes the session for a short turn.
|
||||
[
|
||||
[
|
||||
queueOperation("2026-02-08T01:10:00.000Z", "task-1", "enqueue", checkInText("task-1")),
|
||||
queueOperation("2026-02-08T01:10:00.000Z", "task-1", "dequeue", ""),
|
||||
checkIn("2026-02-08T01:10:00.000Z", "task-1"),
|
||||
assistant("2026-02-08T01:10:01.000Z", "The server still runs"),
|
||||
],
|
||||
"running",
|
||||
],
|
||||
// A running turn takes the next check-in in.
|
||||
[
|
||||
[
|
||||
user("2026-02-08T01:20:00.000Z", "Next"),
|
||||
queueOperation("2026-02-08T01:20:01.000Z", "task-1", "enqueue", checkInText("task-1")),
|
||||
queueOperation("2026-02-08T01:20:02.000Z", "task-1", "remove", checkInText("task-1")),
|
||||
assistant("2026-02-08T01:20:03.000Z"),
|
||||
],
|
||||
"running",
|
||||
],
|
||||
[
|
||||
[
|
||||
notification("2026-02-08T01:30:00.000Z", "task-1"),
|
||||
assistant("2026-02-08T01:30:01.000Z", "The server stopped"),
|
||||
],
|
||||
"idle",
|
||||
],
|
||||
);
|
||||
});
|
||||
|
||||
// IJAI-3744
|
||||
it("closes a task at its TaskStop result", async () => {
|
||||
await expectStateAfterEachStep(
|
||||
[
|
||||
[
|
||||
...before,
|
||||
user("2026-02-08T01:00:00.000Z"),
|
||||
toolUse("2026-02-08T01:00:01.000Z"),
|
||||
backgroundShellResult("2026-02-08T01:00:02.000Z", "task-1"),
|
||||
assistant("2026-02-08T01:00:03.000Z", "Started"),
|
||||
],
|
||||
"running",
|
||||
],
|
||||
[
|
||||
[
|
||||
user("2026-02-08T01:00:10.000Z", "Stop it"),
|
||||
toolUse("2026-02-08T01:00:11.000Z", "tu_stop", "TaskStop"),
|
||||
],
|
||||
"running",
|
||||
],
|
||||
// The scan matches the result to the TaskStop call of an earlier step.
|
||||
[
|
||||
[
|
||||
taskStopResult("2026-02-08T01:00:12.000Z", "task-1"),
|
||||
assistant("2026-02-08T01:00:13.000Z", "Stopped"),
|
||||
],
|
||||
"idle",
|
||||
],
|
||||
);
|
||||
});
|
||||
|
||||
// IJAI-3744
|
||||
it("keeps a task open at a task_id in the result of another tool", async () => {
|
||||
await expectStateAfterEachStep([
|
||||
[
|
||||
...before,
|
||||
user("2026-02-08T01:00:00.000Z"),
|
||||
toolUse("2026-02-08T01:00:01.000Z"),
|
||||
backgroundShellResult("2026-02-08T01:00:02.000Z", "task-1"),
|
||||
toolUse("2026-02-08T01:00:03.000Z", "tu_output", "TaskOutput"),
|
||||
taskStopResult("2026-02-08T01:00:04.000Z", "task-1", "tu_output"),
|
||||
assistant("2026-02-08T01:00:05.000Z", "Still running"),
|
||||
],
|
||||
"running",
|
||||
]);
|
||||
});
|
||||
|
||||
// IJAI-3744
|
||||
it("closes the tasks of the previous process in a resumed session", async () => {
|
||||
await expectStateAfterEachStep(
|
||||
[
|
||||
[
|
||||
...before,
|
||||
user("2026-02-08T01:00:00.000Z"),
|
||||
toolUse("2026-02-08T01:00:01.000Z"),
|
||||
backgroundShellResult("2026-02-08T01:00:02.000Z", "task-1"),
|
||||
assistant("2026-02-08T01:00:03.000Z", "Started"),
|
||||
],
|
||||
"running",
|
||||
],
|
||||
// The resumed CLI reports the task of the gone process as stopped, and
|
||||
// delivers that with the next prompt.
|
||||
[
|
||||
[
|
||||
queueOperation("2026-02-08T02:00:00.000Z", "task-1"),
|
||||
notification("2026-02-08T02:00:00.100Z", "task-1", "stopped"),
|
||||
user("2026-02-08T02:00:00.200Z", "Say hi"),
|
||||
assistant("2026-02-08T02:00:01.000Z", "Hi"),
|
||||
],
|
||||
"idle",
|
||||
],
|
||||
);
|
||||
});
|
||||
|
||||
// IJAI-3744
|
||||
it("keeps no turn open for a background task of a sub-agent", async () => {
|
||||
await expectStateAfterEachStep([
|
||||
[
|
||||
...before,
|
||||
user("2026-02-08T01:00:00.000Z"),
|
||||
backgroundShellResult("2026-02-08T01:00:01.000Z", "task-of-sub-agent", {
|
||||
isSidechain: true,
|
||||
}),
|
||||
assistant("2026-02-08T01:00:02.000Z"),
|
||||
],
|
||||
"idle",
|
||||
]);
|
||||
});
|
||||
});
|
||||
|
||||
describe("a background task before the tail window", () => {
|
||||
// IJAI-3744
|
||||
it("keeps the session running over later turns", async () => {
|
||||
await expectStateAfterEachStep(
|
||||
[
|
||||
[
|
||||
user("2026-02-08T01:00:00.000Z"),
|
||||
toolUse("2026-02-08T01:00:01.000Z"),
|
||||
backgroundShellResult("2026-02-08T01:00:02.000Z", "task-1"),
|
||||
assistant("2026-02-08T01:00:03.000Z", "Started"),
|
||||
...padding("2026-02-08T01:00:04.000Z"),
|
||||
],
|
||||
"running",
|
||||
],
|
||||
// Within the tail window of the scan before: the scan takes over its tasks.
|
||||
[
|
||||
[user("2026-02-08T01:00:10.000Z", "Next"), assistant("2026-02-08T01:00:11.000Z")],
|
||||
"running",
|
||||
],
|
||||
// Past the tail window of the scan before: the whole file is searched again.
|
||||
[padding("2026-02-08T01:00:12.000Z"), "running"],
|
||||
[
|
||||
[
|
||||
notification("2026-02-08T01:00:20.000Z", "task-1"),
|
||||
assistant("2026-02-08T01:00:21.000Z", "All done"),
|
||||
],
|
||||
"idle",
|
||||
],
|
||||
);
|
||||
});
|
||||
|
||||
// IJAI-3744
|
||||
it("is taken over from the scan before without a whole-file search", async () => {
|
||||
let wholeFileSearches = 0;
|
||||
// The index asks for the live state only before it searches the whole file.
|
||||
const index = new SessionIndex({
|
||||
isLive: async () => {
|
||||
wholeFileSearches++;
|
||||
return true;
|
||||
},
|
||||
});
|
||||
const tasksAfter = async (lines: string[]) => {
|
||||
await fs.appendFile(transcript, toJsonl(lines));
|
||||
const { rows } = await index.list({ cwd: workspace, limit: 10, archived: "unarchived" });
|
||||
return rows[0]?.facts.backgroundTasks?.open;
|
||||
};
|
||||
const started = await tasksAfter([
|
||||
user("2026-02-08T01:00:00.000Z"),
|
||||
toolUse("2026-02-08T01:00:01.000Z"),
|
||||
backgroundShellResult("2026-02-08T01:00:02.000Z", "task-1"),
|
||||
assistant("2026-02-08T01:00:03.000Z", "Started"),
|
||||
...padding("2026-02-08T01:00:04.000Z"),
|
||||
]);
|
||||
expect(started).toEqual(["task-1"]);
|
||||
expect(wholeFileSearches).toBe(1);
|
||||
// Within the tail window of the scan before.
|
||||
const takenOver = await tasksAfter([
|
||||
user("2026-02-08T01:00:10.000Z", "Next"),
|
||||
assistant("2026-02-08T01:00:11.000Z"),
|
||||
]);
|
||||
expect(takenOver).toEqual(["task-1"]);
|
||||
expect(wholeFileSearches).toBe(1);
|
||||
// Past the tail window of the scan before.
|
||||
const searchedAgain = await tasksAfter(padding("2026-02-08T01:00:12.000Z"));
|
||||
expect(searchedAgain).toEqual(["task-1"]);
|
||||
expect(wholeFileSearches).toBe(2);
|
||||
const ended = await tasksAfter([
|
||||
notification("2026-02-08T01:00:20.000Z", "task-1"),
|
||||
assistant("2026-02-08T01:00:21.000Z", "All done"),
|
||||
]);
|
||||
expect(ended).toEqual([]);
|
||||
expect(wholeFileSearches).toBe(2);
|
||||
});
|
||||
|
||||
it("is not searched for a session that no live process holds", async () => {
|
||||
const lines = [
|
||||
user("2026-02-08T01:00:00.000Z"),
|
||||
toolUse("2026-02-08T01:00:01.000Z"),
|
||||
backgroundShellResult("2026-02-08T01:00:02.000Z", "task-1"),
|
||||
...padding("2026-02-08T01:00:04.000Z"),
|
||||
];
|
||||
await fs.writeFile(transcript, toJsonl(lines));
|
||||
const fromTail = await listedBackgroundTasks(false);
|
||||
expect(fromTail).toEqual({ open: [], stopCalls: [], complete: false });
|
||||
const fromWholeFile = await listedBackgroundTasks(true);
|
||||
expect(fromWholeFile).toEqual({ open: ["task-1"], stopCalls: [], complete: true });
|
||||
});
|
||||
});
|
||||
|
||||
describe("the end of the wait for a background task", () => {
|
||||
const lines = () => [
|
||||
user("2026-02-08T01:00:00.000Z"),
|
||||
toolUse("2026-02-08T01:00:01.000Z"),
|
||||
backgroundShellResult("2026-02-08T01:00:02.000Z", "task-1"),
|
||||
assistant("2026-02-08T01:00:03.000Z", "Started"),
|
||||
];
|
||||
|
||||
beforeEach(async () => {
|
||||
await fs.writeFile(transcript, toJsonl(lines()));
|
||||
});
|
||||
|
||||
// IJAI-3744
|
||||
it("does not come after 10 minutes", async () => {
|
||||
await registerHolder();
|
||||
const old = new Date(Date.now() - 11 * 60_000);
|
||||
await fs.utimes(transcript, old, old);
|
||||
expect(await stateWithNewService()).toBe("running");
|
||||
});
|
||||
|
||||
// IJAI-3744
|
||||
it("comes when the registry status is idle", async () => {
|
||||
await registerHolder("idle");
|
||||
expect(await stateWithNewService()).toBe("idle");
|
||||
});
|
||||
|
||||
// IJAI-3744
|
||||
it("comes when no live process holds the session", async () => {
|
||||
expect(await stateWithNewService()).toBe("idle");
|
||||
});
|
||||
});
|
||||
|
||||
describe("the task notification", () => {
|
||||
it("is read only at the start of the input", () => {
|
||||
expect(endedTaskId(notificationText("task-1"))).toBe("task-1");
|
||||
expect(endedTaskId(`\n ${notificationText("task-1")}`)).toBe("task-1");
|
||||
expect(endedTaskId(`Look: ${notificationText("task-1")}`)).toBeUndefined();
|
||||
const status = "<status>completed</status></task-notification>";
|
||||
expect(endedTaskId(`<task-notification><task-id> </task-id>${status}`)).toBeUndefined();
|
||||
expect(endedTaskId(`<task-notification><task-id>task-1${status}`)).toBeUndefined();
|
||||
expect(endedTaskId(undefined)).toBeUndefined();
|
||||
});
|
||||
|
||||
it("ends no task without a status", () => {
|
||||
expect(endedTaskId(checkInText("task-1"))).toBeUndefined();
|
||||
const monitorEvent =
|
||||
"<task-notification>\n<task-id>task-1</task-id>\n" +
|
||||
'<summary>Monitor event: "test log"</summary>\n<event>FAILED</event>\n</task-notification>';
|
||||
expect(endedTaskId(monitorEvent)).toBeUndefined();
|
||||
});
|
||||
|
||||
it("is read as a text block of a user record", async () => {
|
||||
const content = [{ type: "text", text: notificationText("task-1") }];
|
||||
const lines = [
|
||||
backgroundShellResult("2026-02-08T01:00:00.000Z", "task-1"),
|
||||
record("2026-02-08T01:00:01.000Z", { type: "user", message: { role: "user", content } }),
|
||||
];
|
||||
await fs.writeFile(transcript, toJsonl(lines));
|
||||
const { size } = await fs.stat(transcript);
|
||||
const tasks = await searchBackgroundTasks(transcript, size);
|
||||
expect(tasks.open).toEqual([]);
|
||||
});
|
||||
});
|
||||
@@ -2,8 +2,8 @@ import { describe, expect, it } from "vitest";
|
||||
import * as fs from "node:fs/promises";
|
||||
import * as os from "node:os";
|
||||
import * as path from "node:path";
|
||||
import { deriveActivity, selectCost } from "../session-index/activity.js";
|
||||
import { LiveSessionRegistry } from "../session-index/live-registry.js";
|
||||
import { deriveActivity, selectCost, type ActivityState } from "../session-index/activity.js";
|
||||
import { LiveSessionRegistry, type LiveRecord } from "../session-index/live-registry.js";
|
||||
import {
|
||||
encodeProjectPath,
|
||||
isSessionId,
|
||||
@@ -271,6 +271,15 @@ describe("activity", () => {
|
||||
hasMessages: true,
|
||||
...overrides,
|
||||
});
|
||||
/** An interactive CLI that holds the session. */
|
||||
const live = (status: string, entrypoint = "cli", statusUpdatedAt = now): LiveRecord => ({
|
||||
pid: 1,
|
||||
sessionId: "s",
|
||||
kind: "interactive",
|
||||
entrypoint,
|
||||
status,
|
||||
statusUpdatedAt,
|
||||
});
|
||||
|
||||
it("reports the SDK state of a session this connection runs", () => {
|
||||
expect(
|
||||
@@ -316,14 +325,6 @@ describe("activity", () => {
|
||||
expect(again.lastTurnError).toBeUndefined();
|
||||
expect(deriveActivity({ facts: again, ...at })?.state).toBe("idle");
|
||||
// A busy or waiting CLI wins; an idle one shows the error.
|
||||
const live = (status: string) => ({
|
||||
pid: 1,
|
||||
sessionId: "s",
|
||||
kind: "interactive",
|
||||
entrypoint: "cli",
|
||||
status,
|
||||
statusUpdatedAt: now,
|
||||
});
|
||||
const held = { facts: errorFacts, transcriptMtimeMs: now - 1, now };
|
||||
expect(deriveActivity({ live: live("busy"), ...held })?.state).toBe("running");
|
||||
expect(deriveActivity({ live: live("waiting"), ...held })?.state).toBe("requires_action");
|
||||
@@ -433,6 +434,34 @@ describe("activity", () => {
|
||||
}),
|
||||
).toBeUndefined();
|
||||
});
|
||||
|
||||
// IJAI-3744
|
||||
describe("with an open background task", () => {
|
||||
const tasks = (open: string[]) => ({ open, stopCalls: [], complete: true });
|
||||
const waiting = facts({ turnState: "finished", backgroundTasks: tasks(["task-1"]) });
|
||||
/** A status older than the transcript, so the transcript decides. */
|
||||
const held = live("busy", "cli", 1);
|
||||
it.each<[string, LiveRecord, TranscriptFacts, ActivityState | undefined]>([
|
||||
["an SDK-driven CLI is busy", live("busy", "sdk-ts"), waiting, "running"],
|
||||
["an SDK-driven CLI is idle", live("idle", "sdk-ts"), waiting, "idle"],
|
||||
["a newer registry status is waiting", live("waiting"), waiting, "requires_action"],
|
||||
["a newer registry status is idle", live("idle"), waiting, "idle"],
|
||||
["the turn failed", held, { ...waiting, lastTurnError: true }, "running"],
|
||||
// An unfinished turn that has not written for 10 minutes has no known state.
|
||||
["the turn is in progress", held, { ...waiting, turnState: "unfinished" }, undefined],
|
||||
["the task ended", held, { ...waiting, backgroundTasks: tasks([]) }, "idle"],
|
||||
])("when %s", (_, record, transcriptFacts, expected) => {
|
||||
// The wait does not end after 10 minutes.
|
||||
const transcriptMtimeMs = now - 11 * 60_000;
|
||||
const state = deriveActivity({
|
||||
live: record,
|
||||
facts: transcriptFacts,
|
||||
transcriptMtimeMs,
|
||||
now,
|
||||
})?.state;
|
||||
expect(state).toBe(expected);
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
describe("cost", () => {
|
||||
|
||||
Reference in New Issue
Block a user