Show and stop agent background work (#6220)

* feat(server): add an in-memory store for agent background work

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* feat(claude): report background tasks as background work

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* feat(server): track agent background work in the agent manager

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* feat(protocol): add agent background work messages

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* feat(server): serve and push agent background work

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* feat(client): expose agent background work on agent handles

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* docs(sdk): document agent background work

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* test(claude): cover background shells in background work end to end

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix(client): resync background work subscriptions on every snapshot

subscribe() now lists the agent's background work each time the event
observation is established, so the listener starts with the current list
and catches up after a reconnect. A list answered after a newer update
is dropped.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix(server): answer background work lists without loading dormant agents

Background work lives only in memory, so a stored agent that is not
loaded has none. The list handler now answers [] from storage instead of
resuming the agent, and still rejects unknown, internal, and archived
agents. A new test pins that old apps never receive
agent.background_work.update.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix(server): read Claude background tasks defensively

The user's claude binary can differ from the SDK typings, so a task's
description is read as unknown and an empty or missing one becomes null.
Drop the monitor kind: Claude Monitor watches report as local_bash, so
kind is "shell" or "other" today. The store now copies only the wire
fields of each input.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* test(server): prove internal background work was emitted before checking globals

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* docs(sdk): describe background work resync and feature gating

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* test(claude): cover background work reported by a retired query

The daemon stops reading a Claude query it no longer owns, so work that a
retired process still had queued cannot reach the background work list.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix(client): let background work subscribers handle a failed list read

When the list read that follows each subscription snapshot failed, the
error was only logged and the caller got neither the list nor a failure.
subscribe() now takes an optional onError that receives it; without one
the error is still logged. The read is not retried, since the next
reconnect reads the list again.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* feat(agents): show background work as active

Keep background activity visible after a foreground turn ends while preserving independent send and interrupt behavior.

* fix(agents): protect background work subscriptions

Keep internal work out of public updates and restore observation after a subscription failure.

* feat(agents): show Codex and OpenCode background work

Keep provider-owned shell activity visible after a turn ends while leaving foreground turn controls independent.

* feat(agents): stop individual background work

Wait for provider confirmation so stopping one background item leaves sibling work and foreground turn controls independent.

* refactor(agents): name background stop parameters

* test(website): isolate browser tests from live metadata

* test(agents): clarify background stop journeys

* fix(sdk): report background work subscription failures

Report setup failures so subscribers can replace an observation that the host released.

* fix(app): restore background work after reconnect

Wait for the host handshake before selecting the retained client so an offline pane starts observing when the connection returns.

* fix(agents): retain background shells across rewind

Preserve known live shell ownership when conversation history is rewound, and revoke it when the provider runtime exits.

* fix(website): launch test seeder across platforms

Invoke Wrangler through Node so local website test startup does not depend on an executable npm command shim.

* feat(agents): clarify background task details and tracks

Keep launch descriptions and commands distinct so background work remains identifiable, and allow composer tracks to scroll when space is limited.

* fix(agents): preserve active background work during archive

Keep background-running children out of bulk cleanup and warn before an individual archive. Preserve first taps on composer tracks while the keyboard is open.

* fix(codex): retain background terminals across rewind

Keep live terminal inventories attached to their originating threads so a fork does not hide or misroute their stop actions.

* fix(codex): scope background terminal identities to threads

Prevent commands on separate threads with the same provider process ID from sharing a stop target.

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
Co-authored-by: Mohamed Boudra <boudra.moha@gmail.com>
This commit is contained in:
fxlelouarn
2026-11-06 19:49:49 +07:00
committed by GitHub
co-authored by Claude Opus 5.5 Mohamed Boudra
parent c8bd4717c9
commit 77798b2e46
99 changed files with 5681 additions and 149 deletions
@@ -13075,3 +13075,172 @@ test("an out-of-band command racing idle cleanup waits for the resumed runtime",
expect(h.commands).toEqual(["/maintenance"]);
expect(h.prompts).toEqual([]);
});
test("folds provider background work and clears it when the agent closes", async () => {
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-background-work-"));
const storage = new AgentStorage(join(workdir, "agents"), logger);
let activeSession: TestAgentSession | null = null;
class BackgroundWorkClient extends TestAgentClient {
override async createSession(config: AgentSessionConfig): Promise<AgentSession> {
activeSession = new TestAgentSession(config);
return activeSession;
}
}
const manager = new AgentManager({
clients: { codex: new BackgroundWorkClient() },
registry: storage,
logger,
idFactory: () => "00000000-0000-4000-8000-000000000201",
});
const snapshot = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, {
workspaceId: undefined,
});
const events: AgentManagerEvent[] = [];
manager.subscribe((event) => events.push(event), { agentId: snapshot.id, replayState: false });
activeSession?.pushEvent({
type: "background_work",
provider: "codex",
items: [{ id: "bash-1", kind: "shell", description: "sleep 20" }],
});
await vi.waitFor(() =>
expect(manager.listAgentBackgroundWork(snapshot.id)).toEqual([
{ id: "bash-1", kind: "shell", description: "sleep 20", startedAt: expect.any(String) },
]),
);
expect(events).toContainEqual({
type: "background_work",
agentId: snapshot.id,
items: [expect.objectContaining({ id: "bash-1" })],
});
expect(manager.getAgent(snapshot.id)?.lifecycle).toBe("idle");
expect(manager.getAgentBackgroundWorkSummary(snapshot.id)).toEqual({ count: 1 });
expect(
events.some((event) => event.type === "agent_stream" && event.event.type === "background_work"),
).toBe(false);
await manager.closeAgent(snapshot.id);
expect(events).toContainEqual({ type: "background_work", agentId: snapshot.id, items: [] });
expect(manager.getAgentBackgroundWorkSummary(snapshot.id)).toBeUndefined();
});
test("does not deliver an internal agent's background work to global subscribers", async () => {
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-background-work-internal-"));
const storage = new AgentStorage(join(workdir, "agents"), logger);
let activeSession: TestAgentSession | null = null;
class BackgroundWorkClient extends TestAgentClient {
override async createSession(config: AgentSessionConfig): Promise<AgentSession> {
activeSession = new TestAgentSession(config);
return activeSession;
}
}
const manager = new AgentManager({
clients: { codex: new BackgroundWorkClient() },
registry: storage,
logger,
idFactory: () => "00000000-0000-4000-8000-000000000202",
});
const events: AgentManagerEvent[] = [];
manager.subscribe((event) => events.push(event), { replayState: false });
const agent = await manager.createAgent(
{ provider: "codex", cwd: workdir, internal: true },
undefined,
{ workspaceId: undefined },
);
const agentEvents: AgentManagerEvent[] = [];
manager.subscribe((event) => agentEvents.push(event), {
agentId: agent.id,
replayState: false,
});
activeSession?.pushEvent({
type: "background_work",
provider: "codex",
items: [{ id: "bash-1", kind: "shell", description: null }],
});
// The per-agent subscriber proves the event was emitted before the global one is checked.
await vi.waitFor(() =>
expect(agentEvents).toContainEqual(
expect.objectContaining({ type: "background_work", agentId: agent.id }),
),
);
expect(events.some((event) => event.type === "background_work")).toBe(false);
await expect(
manager.stopAgentBackgroundWork({ agentId: agent.id, itemId: "bash-1", stopToken: "token" }),
).rejects.toThrow("Unknown agent");
});
test("stops only a currently owned public background item with its current token", async () => {
const workdir = mkdtempSync(join(tmpdir(), "agent-background-stop-"));
const storage = new AgentStorage(join(workdir, "agents"), logger);
const stopped: string[] = [];
let activeSession: TestAgentSession | null = null;
class StoppableSession extends TestAgentSession {
async stopBackgroundWork(id: string) {
stopped.push(id);
}
}
class BackgroundWorkClient extends TestAgentClient {
override async createSession(config: AgentSessionConfig): Promise<AgentSession> {
activeSession = new StoppableSession(config);
return activeSession;
}
}
const manager = new AgentManager({
clients: { codex: new BackgroundWorkClient() },
registry: storage,
logger,
idFactory: () => "00000000-0000-4000-8000-000000000204",
});
const agent = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, {
workspaceId: undefined,
});
activeSession?.pushEvent({
type: "background_work",
provider: "codex",
items: [
{ id: "one", kind: "shell", description: null, canStop: true },
{ id: "two", kind: "other", description: null },
],
});
await vi.waitFor(() => expect(manager.listAgentBackgroundWork(agent.id)).toHaveLength(2));
const [one] = manager.listAgentBackgroundWork(agent.id);
await expect(
manager.stopAgentBackgroundWork({ agentId: agent.id, itemId: "one", stopToken: "stale-token" }),
).rejects.toThrow();
await expect(
manager.stopAgentBackgroundWork({
agentId: agent.id,
itemId: "two",
stopToken: one.stopToken!,
}),
).rejects.toThrow();
await manager.stopAgentBackgroundWork({
agentId: agent.id,
itemId: "one",
stopToken: one.stopToken!,
});
expect(stopped).toEqual(["one"]);
activeSession?.pushEvent({
type: "background_work",
provider: "codex",
items: [{ id: "one", kind: "shell", description: null }],
});
activeSession?.pushEvent({
type: "background_work",
provider: "codex",
items: [{ id: "one", kind: "shell", description: null, canStop: true }],
});
await expect(
manager.stopAgentBackgroundWork({
agentId: agent.id,
itemId: "one",
stopToken: one.stopToken!,
}),
).rejects.toThrow();
expect(stopped).toEqual(["one"]);
await manager.closeAgent(agent.id);
});
@@ -98,6 +98,7 @@ import {
type ProviderSubagentDescriptor,
type ProviderSubagentStoreEvent,
} from "./provider-subagents/store.js";
import { AgentBackgroundWorkStore, type AgentBackgroundWorkItem } from "./background-work/index.js";
import { withTimeout } from "../../utils/promise-timeout.js";
import { extractAttention } from "../persistence-hooks.js";
@@ -171,6 +172,12 @@ async function assertUsableWorkingDirectory(cwd: string): Promise<void> {
}
}
interface StopAgentBackgroundWorkOptions {
agentId: string;
itemId: string;
stopToken: string;
}
async function resolveResumePurpose(
record: StoredAgentRecord | null,
requested: AgentResumePurpose | undefined,
@@ -255,6 +262,7 @@ export type AgentManagerEvent =
| { type: "agent_state"; agent: ManagedAgent }
| { type: "agent_removed"; agentId: string; workspaceId?: string; internal: boolean }
| { type: "provider_subagent"; event: ProviderSubagentStoreEvent }
| { type: "background_work"; agentId: string; items: AgentBackgroundWorkItem[] }
| { type: "timeline_replacement"; agentId: string; epoch: string }
| {
type: "agent_stream";
@@ -794,6 +802,7 @@ export class AgentManager {
private readonly agents = new Map<string, LiveManagedAgent>();
private readonly timelineStore: InMemoryAgentTimelineStore;
private readonly providerSubagents = new ProviderSubagentStore();
private readonly backgroundWork = new AgentBackgroundWorkStore();
private readonly agentsAwaitingInitialSnapshotPersist = new Set<string>();
private readonly sessionEventTails = new Map<string, Promise<void>>();
private readonly steerEventBarriers = new Map<string, SteerEventBarrier>();
@@ -1354,6 +1363,38 @@ export class AgentManager {
return this.providerSubagents.list(parentAgentId);
}
listAgentBackgroundWork(agentId: string): AgentBackgroundWorkItem[] {
this.requirePublicAgent(agentId);
return this.backgroundWork.list(agentId);
}
async stopAgentBackgroundWork({
agentId,
itemId,
stopToken,
}: StopAgentBackgroundWorkOptions): Promise<void> {
this.requirePublicAgent(agentId);
// Inventory events share the session queue; drain events already received before
// authorizing the token, including a reconnect's ownership revocation.
await this.sessionEventTails.get(agentId)?.catch(() => undefined);
const agent = this.requirePublicAgent(agentId);
if (!agent.session || !agent.session.stopBackgroundWork)
throw new Error("Background work cannot be stopped");
const item = this.backgroundWork.list(agentId).find((entry) => entry.id === itemId);
if (!item?.canStop || item.stopToken !== stopToken)
throw new Error("Background work is no longer available to stop");
await agent.session.stopBackgroundWork(itemId);
}
getAgentBackgroundWorkSummary(agentId: string): { count: number } | undefined {
return this.backgroundWork.summary(agentId);
}
private clearBackgroundWork(agentId: string): void {
const items = this.backgroundWork.clear(agentId);
if (items) this.dispatch({ type: "background_work", agentId, items });
}
listProviderSubagentActivity(): ProviderSubagentDescriptor[] {
const publicParentIds = new Set(
Array.from(this.agents.values())
@@ -1740,6 +1781,7 @@ export class AgentManager {
await this.closeReloadedSession(existing.session, agentId);
await this.drainSessionEvents(agentId);
this.cancelRunningProviderSubagents(agentId);
this.clearBackgroundWork(agentId);
closedExisting = this.prepareAgentForClosure(existing, "agent reloaded");
await this.persistSnapshot(closedExisting);
this.assertAcceptingAgentRegistrations();
@@ -1990,6 +2032,8 @@ export class AgentManager {
await agent.session.close();
await this.drainSessionEvents(agentId);
this.cancelRunningProviderSubagents(agentId);
// close() drops the session's subscribers before the provider could report an empty list.
this.clearBackgroundWork(agentId);
const closedAgent = this.prepareAgentForClosure(agent, "agent closed");
let persistError: unknown;
@@ -4182,6 +4226,7 @@ export class AgentManager {
for (const event of this.providerSubagents.deleteParent(agentId)) {
this.dispatch({ type: "provider_subagent", event });
}
this.clearBackgroundWork(agentId);
}
private emitClosedAgent(agent: ManagedAgentClosed, options?: { persist?: boolean }): void {
@@ -4291,6 +4336,14 @@ export class AgentManager {
this.dispatch({ type: "provider_subagent", event: update });
return;
}
if (event.type === "background_work") {
const items =
event.items === null
? this.backgroundWork.clear(agent.id)
: this.backgroundWork.apply(agent.id, event.items);
if (items) this.dispatch({ type: "background_work", agentId: agent.id, items });
return;
}
const turnId = getAgentStreamEventTurnId(event);
const matchingWaiters = this.runs.getMatchingWaiters(agent, turnId);
this.logger.trace(
@@ -5510,6 +5563,13 @@ export class AgentManager {
) {
continue;
}
if (
subscriber.agentId &&
event.type === "background_work" &&
subscriber.agentId !== event.agentId
) {
continue;
}
if (
subscriber.agentId &&
event.type === "agent_state" &&
@@ -5543,6 +5603,7 @@ export class AgentManager {
if (event.type === "agent_state") return event.agent.internal === true;
if (event.type === "agent_removed") return event.internal;
if (event.type === "agent_stream") return this.agents.get(event.agentId)?.internal === true;
if (event.type === "background_work") return this.agents.get(event.agentId)?.internal === true;
if (event.type !== "provider_subagent") return false;
const parentAgentId =
event.event.type === "upsert"
@@ -4,6 +4,7 @@ import { AGENT_LIFECYCLE_STATUSES } from "./agent-manager.js";
import {
buildStoredAgentPayload,
toAgentPayload,
toAgentListItemPayload,
toRecentProviderSessionDescriptorPayload,
toStoredAgentRecord,
type ManagedAgent,
@@ -539,3 +540,11 @@ describe("toRecentProviderSessionDescriptorPayload", () => {
});
});
});
it("omits unknown background work and preserves known summary in list items", () => {
const snapshot = toAgentPayload(createManagedAgent());
expect(toAgentListItemPayload(snapshot)).not.toHaveProperty("backgroundWork");
expect(toAgentListItemPayload({ ...snapshot, backgroundWork: { count: 2 } })).toMatchObject({
backgroundWork: { count: 2 },
});
});
@@ -287,6 +287,7 @@ export function toAgentListItemPayload(agent: AgentSnapshotPayload): AgentListIt
thinkingOptionId: agent.thinkingOptionId,
effectiveThinkingOptionId: agent.effectiveThinkingOptionId,
status: agent.status,
...(agent.backgroundWork ? { backgroundWork: agent.backgroundWork } : {}),
cwd: agent.cwd,
createdAt: agent.createdAt,
updatedAt: agent.updatedAt,
@@ -468,6 +468,12 @@ export type AgentStreamEvent =
type: "provider_subagent";
provider: AgentProvider;
event: import("./provider-subagents/store.js").ProviderSubagentInputEvent;
}
| {
/** Full replacement of the agent's live background work (shells, monitors…). */
type: "background_work";
provider: AgentProvider;
items: import("./background-work/index.js").AgentBackgroundWorkInput[] | null;
};
export function getAgentStreamEventTurnId(event: AgentStreamEvent): string | undefined {
@@ -705,6 +711,8 @@ export interface AgentSession {
* still uncertain.
*/
interrupt(): Promise<void>;
/** Stop one currently owned background item and wait for provider confirmation. */
stopBackgroundWork?(id: string): Promise<void>;
/** True only when releasing this runtime loses no background work or pending replies.
* An optional additional veto: missing methods use daemon baseline eligibility;
* implemented checks that reject or return false keep the runtime resident. */
@@ -0,0 +1,133 @@
import { describe, expect, test } from "vitest";
import { AgentBackgroundWorkStore } from "./index.js";
const T0 = "2026-10-06T10:00:00.000Z";
const T1 = "2026-10-06T10:05:00.000Z";
describe("AgentBackgroundWorkStore", () => {
test("stamps new items and keeps their start time across replacements", () => {
const store = new AgentBackgroundWorkStore();
expect(
store.apply("agent-a", [{ id: "bash-1", kind: "shell", description: "sleep 20" }], T0),
).toEqual([{ id: "bash-1", kind: "shell", description: "sleep 20", startedAt: T0 }]);
expect(
store.apply(
"agent-a",
[
{ id: "bash-1", kind: "shell", description: "sleep 20" },
{ id: "task-1", kind: "other", description: null },
],
T1,
),
).toEqual([
{ id: "bash-1", kind: "shell", description: "sleep 20", startedAt: T0 },
{ id: "task-1", kind: "other", description: null, startedAt: T1 },
]);
});
test("returns null when the replacement changes nothing", () => {
const store = new AgentBackgroundWorkStore();
store.apply("agent-a", [{ id: "bash-1", kind: "shell", description: "sleep 20" }], T0);
expect(
store.apply("agent-a", [{ id: "bash-1", kind: "shell", description: "sleep 20" }], T1),
).toBeNull();
expect(store.apply("agent-b", [], T1)).toEqual([]);
expect(store.apply("agent-b", [], T1)).toBeNull();
});
test("distinguishes an unknown inventory from an observed empty inventory", () => {
const store = new AgentBackgroundWorkStore();
expect(store.summary("agent-a")).toBeUndefined();
store.apply("agent-a", []);
expect(store.summary("agent-a")).toEqual({ count: 0 });
store.clear("agent-a");
expect(store.summary("agent-a")).toBeUndefined();
});
test("keeps the first occurrence of a duplicated id", () => {
const store = new AgentBackgroundWorkStore();
expect(
store.apply(
"agent-a",
[
{ id: "bash-1", kind: "shell", description: "first" },
{ id: "bash-1", kind: "shell", description: "second" },
],
T0,
),
).toEqual([{ id: "bash-1", kind: "shell", description: "first", startedAt: T0 }]);
});
test("scopes lists per agent and clears one agent", () => {
const store = new AgentBackgroundWorkStore();
store.apply("agent-a", [{ id: "bash-1", kind: "shell", description: null }], T0);
store.apply("agent-b", [{ id: "bash-2", kind: "shell", description: null }], T0);
expect(store.clear("agent-a")).toEqual([]);
expect(store.clear("agent-a")).toBeNull();
expect(store.list("agent-a")).toEqual([]);
expect(store.list("agent-b")).toHaveLength(1);
});
test("keeps only the wire fields of each input", () => {
const store = new AgentBackgroundWorkStore();
const input = { id: "bash-1", kind: "shell", description: null, pid: 4242 };
expect(store.apply("agent-a", [input], T0)).toEqual([
{ id: "bash-1", kind: "shell", description: null, startedAt: T0 },
]);
});
test("list returns a copy", () => {
const store = new AgentBackgroundWorkStore();
store.apply("agent-a", [{ id: "bash-1", kind: "shell", description: null }], T0);
store.list("agent-a").pop();
expect(store.list("agent-a")).toHaveLength(1);
});
});
test("stop tokens follow a live item and change when an id is reused", () => {
const store = new AgentBackgroundWorkStore();
const input = { id: "1", kind: "shell", description: null, canStop: true };
const first = store.apply("agent", [input], T0)![0];
expect(first.stopToken).toBeTruthy();
expect(store.apply("agent", [input], T1)).toBeNull();
store.apply("agent", []);
const second = store.apply("agent", [input], T0)![0];
expect(second.stopToken).not.toBe(first.stopToken);
});
test("revokes a stop token while inventory ownership is being re-established", () => {
const store = new AgentBackgroundWorkStore();
const item = { id: "1", kind: "shell", description: null, canStop: true };
const first = store.apply("agent", [item])![0];
const stale = store.apply("agent", [{ ...item, canStop: false }])![0];
expect(stale.stopToken).toBeUndefined();
expect(store.apply("agent", [item])![0].stopToken).not.toBe(first.stopToken);
});
test("publishes command changes without replacing a live item's identity", () => {
const store = new AgentBackgroundWorkStore();
const item = {
id: "1",
kind: "monitor",
description: "Watch build",
command: "tail -f build.log",
canStop: true,
};
const first = store.apply("agent", [item], T0)![0];
expect(first.command).toBe("tail -f build.log");
const changed = store.apply("agent", [{ ...item, command: "tail -f other.log" }], T1)![0];
expect(changed).toMatchObject({
command: "tail -f other.log",
startedAt: T0,
stopToken: first.stopToken,
});
});
@@ -0,0 +1,88 @@
import { randomUUID } from "node:crypto";
export interface AgentBackgroundWorkInput {
id: string;
/** Provider-neutral kind; consumers treat unknown kinds as other work. */
kind: string;
description: string | null;
command?: string;
canStop?: boolean;
}
export interface AgentBackgroundWorkItem extends AgentBackgroundWorkInput {
startedAt: string;
stopToken?: string;
}
/**
* Live background work per agent. Providers report the whole list on every change, so `apply`
* replaces it. `startedAt` is stamped the first time an id is seen and kept while it stays listed.
* Nothing is persisted; providers determine when their live inventory is no longer available.
*/
export class AgentBackgroundWorkStore {
private readonly items = new Map<string, AgentBackgroundWorkItem[]>();
apply(
agentId: string,
inputs: readonly AgentBackgroundWorkInput[],
timestamp: string = new Date().toISOString(),
): AgentBackgroundWorkItem[] | null {
const known = this.items.has(agentId);
const previous = this.items.get(agentId) ?? [];
const previousById = new Map(previous.map((item) => [item.id, item]));
const seen = new Set<string>();
const next: AgentBackgroundWorkItem[] = [];
for (const input of inputs) {
if (seen.has(input.id)) continue;
seen.add(input.id);
next.push({
id: input.id,
kind: input.kind,
description: input.description,
...(input.command ? { command: input.command } : {}),
...(input.canStop
? { canStop: true, stopToken: previousById.get(input.id)?.stopToken ?? randomUUID() }
: {}),
startedAt: previousById.get(input.id)?.startedAt ?? timestamp,
});
}
if (known && sameItems(previous, next)) return null;
this.items.set(agentId, next);
return [...next];
}
list(agentId: string): AgentBackgroundWorkItem[] {
return [...(this.items.get(agentId) ?? [])];
}
summary(agentId: string): { count: number } | undefined {
const items = this.items.get(agentId);
return items ? { count: items.length } : undefined;
}
clear(agentId: string): AgentBackgroundWorkItem[] | null {
if (!this.items.delete(agentId)) return null;
return [];
}
}
function sameItems(
left: readonly AgentBackgroundWorkItem[],
right: readonly AgentBackgroundWorkItem[],
): boolean {
return (
left.length === right.length &&
left.every((item, index) => {
const other = right[index];
return (
item.id === other.id &&
item.kind === other.kind &&
item.description === other.description &&
item.command === other.command &&
item.canStop === other.canStop &&
item.stopToken === other.stopToken &&
item.startedAt === other.startedAt
);
})
);
}
@@ -468,6 +468,7 @@ export function wrapSessionProvider(provider: AgentProvider, inner: AgentSession
respondToPermission: (requestId, response) => inner.respondToPermission(requestId, response),
describePersistence: () => mapPersistenceHandle(provider, inner.describePersistence()),
interrupt: () => inner.interrupt(),
stopBackgroundWork: inner.stopBackgroundWork?.bind(inner),
close: () => inner.close(),
canStopRuntime: inner.canStopRuntime?.bind(inner),
listCommands: inner.listCommands?.bind(inner),
@@ -712,19 +712,36 @@ describe("background Claude subagents", () => {
test("Claude runtime safety protects silent shells, monitors and pending completion notifications", () => {
const safety = new ClaudeRuntimeSafety();
const observe = (message: unknown) => safety.observe(message as SDKMessage);
observe({ type: "system", subtype: "init", claude_code_version: "2.1.292" });
expect(observe({ type: "system", subtype: "init", claude_code_version: "2.1.292" })).toEqual([]);
expect(safety.canStopRuntime()).toBe(true);
expect(
observe({
type: "system",
subtype: "background_tasks_changed",
tasks: [
{ task_id: "quiet-shell", task_type: "local_bash", description: "sleep 7200" },
{ task_id: "monitor", task_type: "local_bash", description: "watch build" },
{ task_id: "native-child", task_type: "local_agent", description: "research" },
],
}),
).toEqual([
{ id: "quiet-shell", kind: "shell", description: "sleep 7200", canStop: true },
{ id: "monitor", kind: "shell", description: "watch build", canStop: true },
]);
expect(safety.canStopRuntime()).toBe(false);
observe({
type: "system",
subtype: "background_tasks_changed",
tasks: [
{ task_id: "quiet-shell", task_type: "local_bash", description: "sleep 7200" },
{ task_id: "monitor", task_type: "local_bash", description: "watch build" },
{ task_id: "native-child", task_type: "local_agent", description: "research" },
],
subtype: "task_started",
task_id: "monitor",
tool_use_id: "tool-monitor",
});
expect(safety.canStopRuntime()).toBe(false);
observe({ type: "system", subtype: "background_tasks_changed", tasks: [] });
expect(
observe({
type: "assistant",
message: { content: [{ type: "tool_use", id: "tool-monitor", name: "Monitor" }] },
})?.find((item) => item.id === "monitor")?.kind,
).toBe("monitor");
expect(observe({ type: "system", subtype: "background_tasks_changed", tasks: [] })).toEqual([]);
expect(safety.canStopRuntime()).toBe(false);
for (const task_id of ["quiet-shell", "monitor"])
observe({ type: "system", subtype: "task_notification", task_id });
@@ -736,6 +753,57 @@ test("Claude runtime safety protects silent shells, monitors and pending complet
expect(safety.canStopRuntime()).toBe(false);
observe({ type: "system", subtype: "background_tasks_changed", tasks: [] });
expect(safety.canStopRuntime()).toBe(true);
safety.retire();
expect(
observe({
type: "system",
subtype: "background_tasks_changed",
tasks: [{ task_id: "late", task_type: "local_bash" }],
}),
).toBeUndefined();
safety.reset();
expect(
observe({
type: "system",
subtype: "background_tasks_changed",
tasks: [{ task_id: "new", task_type: "local_bash" }],
}),
).toEqual([{ id: "new", kind: "shell", description: null, canStop: true }]);
});
test("emits inventory only when relevant observations change it", () => {
const safety = new ClaudeRuntimeSafety();
const observe = (message: unknown) => safety.observe(message as SDKMessage);
expect(observe({ type: "system", subtype: "init", claude_code_version: "2.1.292" })).toEqual([]);
expect(
observe({
type: "system",
subtype: "background_tasks_changed",
tasks: [{ task_id: "m", task_type: "local_bash" }],
}),
).toEqual([{ id: "m", kind: "shell", description: null, canStop: true }]);
expect(
observe({
type: "system",
subtype: "background_tasks_changed",
tasks: [{ task_id: "m", task_type: "local_bash" }],
}),
).toBeUndefined();
expect(
observe({ type: "system", subtype: "task_started", task_id: "m", tool_use_id: "tool" }),
).toBeUndefined();
expect(
observe({
type: "assistant",
message: { content: [{ type: "tool_use", id: "tool", name: "Monitor" }] },
}),
).toEqual([{ id: "m", kind: "monitor", description: null, canStop: true }]);
expect(
observe({ type: "assistant", message: { content: [{ type: "text", text: "streamed" }] } }),
).toBeUndefined();
expect(
observe({ type: "system", subtype: "task_started", task_id: "m", tool_use_id: "other" }),
).toEqual([{ id: "m", kind: "shell", description: null, canStop: true }]);
});
test("Claude session runtime safety observes pumped background work and autonomous replies", async () => {
@@ -143,6 +143,13 @@ function getLatestCompletedBashCall(events: AgentStreamEvent[]): ToolCallTimelin
.find((item) => item.status === "completed" && item.name.toLowerCase() === "bash");
}
function hasBackgroundShell(events: AgentStreamEvent[]): boolean {
return events.some(
(event) =>
event.type === "background_work" && event.items?.some((item) => item.kind === "shell"),
);
}
function getInternalQuery(session: AgentSession): unknown {
return (session as AgentSession & { query?: unknown }).query ?? null;
}
@@ -419,6 +426,39 @@ describe("ClaudeAgentSession integration", () => {
}
}, 180_000);
test("lists a background shell as background work until it exits", async () => {
const handle = await createSession({ cwdPrefix: "claude-agent-background-work-" });
try {
const liveEventsPromise = collectSubscribedUntil(
handle.session,
(event, events) =>
event.type === "background_work" &&
event.items?.length === 0 &&
hasBackgroundShell(events),
90_000,
);
await collectUntilTerminal(
streamSession(
handle.session,
[
"Use the Bash tool with run_in_background.",
"Run exactly: sleep 8",
"Do not wait for the task result.",
"Reply immediately with exactly: SPAWNED",
].join(" "),
),
{ timeoutMs: 90_000 },
);
const liveEvents = await liveEventsPromise;
expect(hasBackgroundShell(liveEvents)).toBe(true);
expect(liveEvents.at(-1)).toMatchObject({ type: "background_work", items: [] });
} finally {
await cleanupSession(handle);
}
}, 180_000);
test("surfaces permission requests and resumes after approval", async () => {
const handle = await createSession({
cwdPrefix: "claude-agent-permission-",
@@ -118,6 +118,22 @@ const RUNNING_WORKFLOW_TURN_EVENTS = [
COMPLETED_TURN_EVENTS[2],
];
const BACKGROUND_SHELL_TURN_EVENTS = [
COMPLETED_TURN_EVENTS[0],
{
type: "system",
subtype: "background_tasks_changed",
tasks: [
{ task_id: "bash-1", task_type: "local_bash", description: "sleep 20" },
{ task_id: "wf-1", task_type: "local_workflow", description: "Run the spec workflow" },
],
uuid: "00000000-0000-4000-8000-000000000001",
session_id: "claude-runtime-exit-session",
},
COMPLETED_TURN_EVENTS[1],
COMPLETED_TURN_EVENTS[2],
];
const MISSING_RESUMED_CONVERSATION_RESULT = {
type: "result",
subtype: "error_during_execution",
@@ -272,6 +288,155 @@ describe("Claude runtime exit", () => {
}
});
test("reports background tasks except subagents as background work", async () => {
const queryFactory = vi.fn(() => createQueryMock(BACKGROUND_SHELL_TURN_EVENTS));
vi.spyOn(spawnUtils, "spawnProcess").mockReturnValue(createChildProcessStub());
const client = new ClaudeAgentClient({
logger: createTestLogger(),
queryFactory,
resolveBinary: async () => "/test/claude/bin",
});
const session = await client.createSession({ provider: "claude", cwd: process.cwd() });
try {
const events: AgentStreamEvent[] = [];
session.subscribe((event) => events.push(event));
await session.run("start a background shell");
expect(events).toContainEqual(
expect.objectContaining({
type: "background_work",
provider: "claude",
items: [{ id: "bash-1", kind: "shell", description: "sleep 20", canStop: true }],
}),
);
} finally {
await session.close();
}
});
test("empties background work when the process dies while idle", async () => {
let capturedOptions: Options | undefined;
const queryFactory = vi.fn(({ options }: ClaudeQueryInput) => {
capturedOptions = options;
return createQueryMock(BACKGROUND_SHELL_TURN_EVENTS);
});
const child = createChildProcessStub();
vi.spyOn(spawnUtils, "spawnProcess").mockReturnValue(child);
const client = new ClaudeAgentClient({
logger: createTestLogger(),
queryFactory,
resolveBinary: async () => "/test/claude/bin",
});
const session = await client.createSession({ provider: "claude", cwd: process.cwd() });
try {
await session.run("start a background shell");
capturedOptions?.spawnClaudeCodeProcess?.(SPAWN_OPTIONS);
const events: AgentStreamEvent[] = [];
session.subscribe((event) => events.push(event));
child.emit("exit", 1, null);
expect(events).toContainEqual(
expect.objectContaining({ type: "background_work", provider: "claude", items: null }),
);
} finally {
await session.close();
}
});
test("empties background work when a query restart retires the process", async () => {
let capturedOptions: Options | undefined;
const child = createChildProcessStub();
const queryFactory = vi.fn(({ options }: ClaudeQueryInput) => {
capturedOptions = options;
return createQueryMock(BACKGROUND_SHELL_TURN_EVENTS, {
tail: new Promise<never>(() => undefined),
onReturn: () => child.emit("exit", 0, null),
});
});
vi.spyOn(spawnUtils, "spawnProcess").mockReturnValue(child);
const client = new ClaudeAgentClient({
logger: createTestLogger(),
queryFactory,
resolveBinary: async () => "/test/claude/bin",
});
const session = await client.createSession({ provider: "claude", cwd: process.cwd() });
try {
await session.run("start a background shell");
capturedOptions?.spawnClaudeCodeProcess?.(SPAWN_OPTIONS);
const events: AgentStreamEvent[] = [];
session.subscribe((event) => events.push(event));
await session.setThinkingOption(null);
await session.listCommands();
expect(events).toContainEqual(
expect.objectContaining({ type: "background_work", provider: "claude", items: null }),
);
} finally {
await session.close();
}
});
test("ignores background work a retired query reports after its restart", async () => {
let capturedOptions: Options | undefined;
let deliverLateEvent: ((event: unknown) => void) | undefined;
let retiredQuery: Query | undefined;
const child = createChildProcessStub();
const queryFactory = vi.fn(({ options }: ClaudeQueryInput) => {
capturedOptions = options;
if (retiredQuery) {
return createQueryMock([], { tail: new Promise<never>(() => undefined) });
}
retiredQuery = createQueryMock(BACKGROUND_SHELL_TURN_EVENTS, {
tail: new Promise<unknown>((resolve) => {
deliverLateEvent = resolve;
}),
onReturn: () => child.emit("exit", 0, null),
});
return retiredQuery;
});
vi.spyOn(spawnUtils, "spawnProcess").mockReturnValue(child);
const client = new ClaudeAgentClient({
logger: createTestLogger(),
queryFactory,
resolveBinary: async () => "/test/claude/bin",
});
const session = await client.createSession({ provider: "claude", cwd: process.cwd() });
try {
await session.run("start a background shell");
capturedOptions?.spawnClaudeCodeProcess?.(SPAWN_OPTIONS);
const events: AgentStreamEvent[] = [];
session.subscribe((event) => events.push(event));
await session.setThinkingOption(null);
await session.listCommands();
expect(queryFactory).toHaveBeenCalledTimes(2);
// The retired query still had this queued when the restart emptied the list.
deliverLateEvent?.(BACKGROUND_SHELL_TURN_EVENTS[1]);
// One next() per scripted event and one for the late event; the pump stops
// reading a query it no longer owns instead of routing that event.
await vi.waitFor(() =>
expect(retiredQuery?.next).toHaveBeenCalledTimes(BACKGROUND_SHELL_TURN_EVENTS.length + 1),
);
// Let a routed late event reach subscribers before checking the list.
await new Promise<void>((resolve) => setImmediate(resolve));
const backgroundWork = events.filter((event) => event.type === "background_work");
expect(backgroundWork.at(-1)).toEqual(
expect.objectContaining({ type: "background_work", provider: "claude", items: null }),
);
} finally {
await session.close();
}
});
test("tree-kills the retired process when the resumed conversation is gone", async () => {
let capturedOptions: Options | undefined;
let deliverMissingConversation: ((event: unknown) => void) | undefined;
@@ -348,4 +513,77 @@ describe("Claude runtime exit", () => {
await session.close();
}
});
test("waits for a stopped task notification after the stop request is accepted", async () => {
let deliver!: (message: unknown) => void;
const tail = new Promise<unknown>((resolve) => {
deliver = resolve;
});
const query = createQueryMock(BACKGROUND_SHELL_TURN_EVENTS, { tail });
query.stopTask = async () => {};
const client = new ClaudeAgentClient({
logger: createTestLogger(),
queryFactory: () => query,
resolveBinary: async () => "/test/claude/bin",
});
const session = await client.createSession({ provider: "claude", cwd: process.cwd() });
try {
await session.run("start a background shell");
let settled = false;
const pending = session.stopBackgroundWork!("bash-1").then(() => {
settled = true;
return undefined;
});
await Promise.resolve();
expect(settled).toBe(false);
deliver({
type: "system",
subtype: "task_notification",
task_id: "bash-1",
status: "stopped",
output_file: "",
summary: "Stopped",
uuid: "00000000-0000-4000-8000-000000000002",
session_id: "claude-runtime-exit-session",
});
await pending;
expect(settled).toBe(true);
await expect(session.stopBackgroundWork!("wf-1")).rejects.toThrow();
} finally {
await session.close();
}
});
test.each(["completed", "failed"])(
"reports the task's %s outcome when it races a stop",
async (status) => {
let deliver!: (message: unknown) => void;
const tail = new Promise<unknown>((resolve) => {
deliver = resolve;
});
const query = createQueryMock(BACKGROUND_SHELL_TURN_EVENTS, { tail });
query.stopTask = async () => {};
const client = new ClaudeAgentClient({
logger: createTestLogger(),
queryFactory: () => query,
resolveBinary: async () => "/test/claude/bin",
});
const session = await client.createSession({ provider: "claude", cwd: process.cwd() });
try {
await session.run("start a background shell");
const result = session.stopBackgroundWork!("bash-1");
deliver({
type: "system",
subtype: "task_notification",
task_id: "bash-1",
status,
output_file: "",
summary: "Finished",
uuid: "00000000-0000-4000-8000-000000000003",
session_id: "claude-runtime-exit-session",
});
await expect(result).rejects.toThrow(status);
} finally {
await session.close();
}
},
);
});
@@ -2908,8 +2908,13 @@ describe("ClaudeAgentSession context window usage", () => {
await session.startTurn("turn", { clientMessageId: "client-message-1" });
await new Promise((resolve) => setTimeout(resolve, 0));
expect(events.slice(0, 2).map((event) => event.type)).toEqual(["turn_started", "timeline"]);
expect(events[1]).toMatchObject({
const started = events.findIndex((event) => event.type === "turn_started");
const userMessage = events.findIndex(
(event) => event.type === "timeline" && event.item.type === "user_message",
);
expect(started).toBeGreaterThanOrEqual(0);
expect(userMessage).toBeGreaterThan(started);
expect(events[userMessage]).toMatchObject({
type: "timeline",
item: { type: "user_message", clientMessageId: "client-message-1" },
});
@@ -2294,6 +2294,7 @@ class ClaudeAgentSession implements AgentSession {
private readonly resolveBinary: () => Promise<string>;
private query: Query | null = null;
private readonly runtimeSafety = new ClaudeRuntimeSafety();
private readonly backgroundStops = new Map<string, (outcome: string) => void>();
private processingSdkMessages = 0;
private readonly harnessEnvironment: Record<string, string>;
private readonly usageSessionKey = randomUUID();
@@ -2669,6 +2670,33 @@ class ClaudeAgentSession implements AgentSession {
};
}
async stopBackgroundWork(id: string): Promise<void> {
const query = this.query;
if (!query || !this.runtimeSafety.canStop(id))
throw new Error("Background work is no longer available");
if (this.backgroundStops.has(id)) throw new Error("Stop is already pending");
let timer: ReturnType<typeof setTimeout> | undefined;
let finish!: (outcome: string) => void;
const outcome = new Promise<string>((resolve) => {
finish = resolve;
this.backgroundStops.set(id, resolve);
});
try {
const result = await Promise.race([
Promise.all([query.stopTask(id), outcome]).then(([, value]) => value),
outcome.then((value) => (value === "stopped" ? new Promise<string>(() => {}) : value)),
new Promise<string>((resolve) => {
timer = setTimeout(() => resolve("timeout"), 15000);
}),
]);
if (result !== "stopped") throw new Error(`Background work stop: ${result}`);
if (this.query !== query) throw new Error("Background work connection changed");
} finally {
clearTimeout(timer);
if (this.backgroundStops.get(id) === finish) this.backgroundStops.delete(id);
}
}
async interrupt(): Promise<void> {
if (this.cancelCurrentTurn) {
this.cancelCurrentTurn();
@@ -3478,6 +3506,7 @@ class ClaudeAgentSession implements AgentSession {
const oldQuery = this.query;
if (!oldQuery) return;
const oldInput = this.input;
for (const finish of this.backgroundStops.values()) finish("connection lost");
// Null out query/input BEFORE awaiting the old iterator's return so the
// old pump sees this.query !== activeQuery and skips failActiveTurns.
this.query = null;
@@ -3532,6 +3561,7 @@ class ClaudeAgentSession implements AgentSession {
this.logger.debug({ options: summarizeClaudeOptionsForLog(options) }, "claude query");
this.input = input;
this.runtimeSafety.reset();
this.dispatchEvents([{ type: "background_work", provider: "claude", items: null }]);
// A fresh Claude process has no turn of its own in flight.
this.mainTurnInFlight = false;
this.query = claudeQuery(
@@ -4127,11 +4157,15 @@ class ClaudeAgentSession implements AgentSession {
}
private failRunningRuntimeTasks(): void {
this.dispatchEvents(
foldSubagentObservations(this.taskProtocolSource.failRunningTasks()).map(
for (const finish of this.backgroundStops.values()) finish("connection lost");
this.runtimeSafety.retire();
this.dispatchEvents([
...foldSubagentObservations(this.taskProtocolSource.failRunningTasks()).map(
(event): AgentStreamEvent => ({ type: "provider_subagent", provider: "claude", event }),
),
);
// The SDK resets its background task set when the CLI process (re)starts.
{ type: "background_work", provider: "claude", items: null },
]);
}
private startQueryPump(): void {
@@ -4199,12 +4233,19 @@ class ClaudeAgentSession implements AgentSession {
if (!this.ownsQuery(activeQuery)) return true;
this.processingSdkMessages++;
try {
this.runtimeSafety.observe(message);
const backgroundWork = this.runtimeSafety.observe(message);
if (message.type === "system" && message.subtype === "task_notification")
this.backgroundStops.get(message.task_id)?.(message.status);
logRawMessage(message);
consecutiveInterruptAbortRecoveries = 0;
if (await this.handleMissingResumedConversation(message, activeQuery)) return true;
if (!this.ownsQuery(activeQuery)) return true;
await this.routeSdkMessageFromPump(message);
if (!this.ownsQuery(activeQuery)) return true;
if (backgroundWork && this.runtimeSafety.isCurrentInventory(backgroundWork))
this.dispatchEvents([
{ type: "background_work", provider: "claude", items: backgroundWork },
]);
return false;
} finally {
this.processingSdkMessages--;
@@ -4841,6 +4882,13 @@ class ClaudeAgentSession implements AgentSession {
this.appendTaskNotificationEvents(message, events);
return;
}
if (message.subtype === "background_tasks_changed") {
this.logger.debug(
{ tasks: message.tasks.map(({ task_id, task_type }) => ({ task_id, task_type })) },
"claude.background_tasks_changed",
);
return;
}
if (message.subtype === "task_progress") {
return;
}
@@ -0,0 +1,21 @@
import { describe, expect, test } from "vitest";
import { toBackgroundWorkInputs } from "./background-work.js";
describe("toBackgroundWorkInputs", () => {
test("maps shells, drops subagents and workflows", () => {
expect(
toBackgroundWorkInputs([
{ task_id: "bash-1", task_type: "local_bash", description: "sleep 20" },
{ task_id: "agent-1", task_type: "local_agent", description: "Explore" },
{ task_id: "wf-1", task_type: "local_workflow", description: "Run the spec workflow" },
{ task_id: "x-1", task_type: "remote_something", description: " " },
{ task_id: "bash-2", task_type: "local_bash" },
]),
).toEqual([
{ id: "bash-1", kind: "shell", description: "sleep 20", canStop: true },
{ id: "x-1", kind: "other", description: null, canStop: false },
{ id: "bash-2", kind: "shell", description: null, canStop: true },
]);
});
});
@@ -0,0 +1,36 @@
import type { AgentBackgroundWorkInput } from "../../background-work/index.js";
import { CLAUDE_SUBAGENT_TASK_TYPE, CLAUDE_WORKFLOW_TASK_TYPE } from "./subagents/live-source.js";
export interface ClaudeBackgroundTask {
task_id: string;
task_type: string;
/** The user's `claude` binary can differ from the SDK typings, so this is read defensively. */
description?: unknown;
}
/** Subagents and workflows already live in the subagents track. */
const SUBAGENT_TASK_TYPES = new Set([CLAUDE_SUBAGENT_TASK_TYPE, CLAUDE_WORKFLOW_TASK_TYPE]);
/** Maps `background_tasks_changed.tasks` to provider-agnostic background work. */
export function toBackgroundWorkInputs(
tasks: readonly ClaudeBackgroundTask[],
): AgentBackgroundWorkInput[] {
return tasks
.filter((task) => !SUBAGENT_TASK_TYPES.has(task.task_type))
.map((task) => ({
id: task.task_id,
kind: toKind(task.task_type),
canStop: task.task_type === "local_bash",
description: readDescription(task.description),
}));
}
/** The inventory owner refines shell entries when launch metadata identifies a monitor. */
function toKind(taskType: string): string {
if (taskType === "local_bash") return "shell";
return "other";
}
function readDescription(value: unknown): string | null {
return typeof value === "string" && value.trim().length > 0 ? value.trim() : null;
}
@@ -0,0 +1,48 @@
import type { SDKMessage } from "@anthropic-ai/claude-agent-sdk";
import { expect, test } from "vitest";
import { ClaudeRuntimeSafety } from "./runtime-safety.js";
const message = (value: unknown) => value as SDKMessage;
test("enriches only live tasks with their correlated launch metadata", () => {
const runtime = new ClaudeRuntimeSafety();
runtime.observe(
message({
type: "assistant",
message: {
content: [
{
type: "tool_use",
id: "tool-1",
name: "Monitor",
input: { description: "Watch build", command: "tail -f build.log" },
},
{ type: "tool_use", id: "old", name: "Bash", input: { command: "old command" } },
],
},
}),
);
runtime.observe(
message({ type: "system", subtype: "task_started", task_id: "task-1", tool_use_id: "tool-1" }),
);
expect(
runtime.observe(
message({
type: "system",
subtype: "background_tasks_changed",
tasks: [{ task_id: "task-1", task_type: "local_bash", description: "Watch build" }],
}),
),
).toEqual([
{
id: "task-1",
kind: "monitor",
description: "Watch build",
command: "tail -f build.log",
canStop: true,
},
]);
expect(
runtime.observe(message({ type: "system", subtype: "background_tasks_changed", tasks: [] })),
).toEqual([]);
});
@@ -1,42 +1,160 @@
import type { SDKMessage } from "@anthropic-ai/claude-agent-sdk";
import type { AgentBackgroundWorkInput } from "../../background-work/index.js";
import { parseClaudeCodeVersion } from "./model-manifest.js";
import { toBackgroundWorkInputs, type ClaudeBackgroundTask } from "./background-work.js";
/** Process-local background ownership. Completion delivery can follow an empty roster. */
export class ClaudeRuntimeSafety {
private supported = false;
private liveTasks = new Set<string>();
private inventoryState: "unknown" | "supported" | "retired" = "unknown";
private liveTasks: readonly ClaudeBackgroundTask[] = [];
private published: AgentBackgroundWorkInput[] | undefined;
private readonly awaitingNotifications = new Set<string>();
private readonly monitorTools = new Set<string>();
private readonly toolInputs = new Map<string, { description?: string; command?: string }>();
private readonly taskTools = new Map<string, string>();
reset(): void {
this.supported = false;
this.liveTasks.clear();
this.inventoryState = "unknown";
this.liveTasks = [];
this.published = undefined;
this.awaitingNotifications.clear();
this.monitorTools.clear();
this.toolInputs.clear();
this.taskTools.clear();
}
observe(message: SDKMessage): void {
if (message.type !== "system") return;
retire(): void {
this.reset();
this.inventoryState = "retired";
}
/** Returns a full replacement only when the visible inventory changes. */
observe(message: SDKMessage): AgentBackgroundWorkInput[] | undefined {
if (this.inventoryState === "retired") return undefined;
const changed = this.observeTools(message);
const systemChanged = message.type === "system" ? this.observeSystem(message) : false;
if ((!changed && !systemChanged) || this.inventoryState !== "supported") return undefined;
const items = this.buildInventory();
if (this.published && sameInventory(this.published, items)) return undefined;
this.published = items;
return items;
}
private observeTools(message: SDKMessage): boolean {
if (message.type === "assistant") {
let relevant = false;
for (const block of message.message.content) {
if (block.type === "tool_use" && (block.name === "Monitor" || block.name === "Bash")) {
if (block.name === "Monitor") relevant = this.observeMonitor(block.id) || relevant;
relevant = this.observeInput(block.id, block.input) || relevant;
}
}
return relevant;
}
if (message.type === "stream_event" && message.event.type === "content_block_start") {
const block = message.event.content_block;
if (block.type === "tool_use" && block.name === "Monitor")
return this.observeMonitor(block.id);
}
return false;
}
private observeInput(id: string, value: unknown): boolean {
const input = value as Record<string, unknown> | null;
const description = typeof input?.description === "string" ? input.description.trim() : "";
const command = typeof input?.command === "string" ? input.command.trim() : "";
const previous = this.toolInputs.get(id);
if (description === (previous?.description ?? "") && command === (previous?.command ?? ""))
return false;
this.toolInputs.set(id, {
...(description ? { description } : {}),
...(command ? { command } : {}),
});
return this.liveTasks.some((task) => this.taskTools.get(task.task_id) === id);
}
private observeMonitor(id: string): boolean {
if (this.monitorTools.has(id)) return false;
this.monitorTools.add(id);
return this.liveTasks.some((task) => this.taskTools.get(task.task_id) === id);
}
private observeSystem(message: Extract<SDKMessage, { type: "system" }>): boolean {
if (message.subtype === "task_started" && message.tool_use_id) {
const previous = this.taskTools.get(message.task_id);
this.taskTools.set(message.task_id, message.tool_use_id);
return (
previous !== message.tool_use_id &&
(this.monitorTools.has(message.tool_use_id) ||
this.toolInputs.has(message.tool_use_id) ||
(previous !== undefined &&
(this.monitorTools.has(previous) || this.toolInputs.has(previous)))) &&
this.liveTasks.some((task) => task.task_id === message.task_id)
);
}
if (message.subtype === "init") {
// 2.1.292 is a known roster-capable release. Older binaries can demonstrate
// support by sending a roster; a fresh process starts with an empty roster.
const version = parseClaudeCodeVersion(message.claude_code_version ?? "");
if (
version &&
(version[0] > 2 ||
(version[0] === 2 && (version[1] > 1 || (version[1] === 1 && version[2] >= 292))))
)
this.supported = true;
) {
const changed = this.inventoryState !== "supported";
this.inventoryState = "supported";
return changed;
}
}
if (message.subtype === "background_tasks_changed") {
this.supported = true;
this.liveTasks = new Set(message.tasks.map((task) => task.task_id));
for (const id of this.liveTasks) this.awaitingNotifications.add(id);
}
if (message.subtype === "task_notification") {
this.awaitingNotifications.delete(message.task_id);
this.inventoryState = "supported";
this.liveTasks = message.tasks;
for (const task of this.liveTasks) this.awaitingNotifications.add(task.task_id);
return true;
}
if (message.subtype === "task_notification") this.awaitingNotifications.delete(message.task_id);
return false;
}
canStopRuntime(): boolean {
return this.supported && this.liveTasks.size === 0 && this.awaitingNotifications.size === 0;
return (
this.inventoryState === "supported" &&
this.liveTasks.length === 0 &&
this.awaitingNotifications.size === 0
);
}
canStop(id: string): boolean {
return this.buildInventory().some((item) => item.id === id && item.canStop);
}
isCurrentInventory(items: AgentBackgroundWorkInput[]): boolean {
return this.inventoryState === "supported" && this.published === items;
}
private buildInventory(): AgentBackgroundWorkInput[] {
return toBackgroundWorkInputs(this.liveTasks).map((item) => {
const tool = this.taskTools.get(item.id) ?? "";
if (this.monitorTools.has(tool)) item.kind = "monitor";
const input = this.toolInputs.get(tool);
if (input?.description) item.description = input.description;
if (input?.command) item.command = input.command;
return item;
});
}
}
function sameInventory(
left: readonly AgentBackgroundWorkInput[],
right: readonly AgentBackgroundWorkInput[],
): boolean {
return (
left.length === right.length &&
left.every(
(item, index) =>
item.id === right[index].id &&
item.kind === right[index].kind &&
item.description === right[index].description &&
item.command === right[index].command &&
item.canStop === right[index].canStop,
)
);
}
@@ -49,9 +49,9 @@ interface TaskStartedMessage {
}
/** Task-tool subagents. Backgrounded shell commands announce as `local_bash`. */
const CLAUDE_SUBAGENT_TASK_TYPE = "local_agent";
export const CLAUDE_SUBAGENT_TASK_TYPE = "local_agent";
/** Workflow executions use the same announced task lifecycle as Task-tool subagents. */
const CLAUDE_WORKFLOW_TASK_TYPE = "local_workflow";
export const CLAUDE_WORKFLOW_TASK_TYPE = "local_workflow";
/**
* Not every announced task belongs in the subagents track. Verified on the wire:
@@ -702,6 +702,20 @@ function markdownImageSource(markdown: string): string {
return source.startsWith("file://") ? fileURLToPath(source) : source;
}
function latestBackgroundWork(events: AgentStreamEvent[]) {
return events.findLast((event) => event.type === "background_work");
}
function hasBackgroundListRequest(appServer: FakeCodexAppServer, threadId: string) {
return appServer
.requests()
.some(
(request) =>
request.method === "thread/backgroundTerminals/list" &&
(request.params as { threadId: string }).threadId === threadId,
);
}
function emitCodexUserMessage(
appServer: FakeCodexAppServer,
input: { id: string; text: string; threadId?: string; turnId?: string },
@@ -2172,13 +2186,16 @@ describe("Codex app-server provider", () => {
expect.objectContaining({ type: "user_message", messageId: "user-1", text: "Hello" }),
expect.objectContaining({ type: "assistant_message", text: "Hi there." }),
]);
expect(codex.threadMethods(0)).toEqual([
// Background terminal observation is independent of conversation history loading.
const historyMethods = (index: number) =>
codex.threadMethods(index).filter((method) => method !== "thread/backgroundTerminals/list");
expect(historyMethods(0)).toEqual([
"thread/loaded/list",
"thread/resume",
"thread/read",
"thread/read",
]);
expect(codex.threadMethods(1)).toEqual(["thread/loaded/list", "thread/resume"]);
expect(historyMethods(1)).toEqual(["thread/loaded/list", "thread/resume"]);
expect(codex.manager.getTimeline(codex.agentId)).toEqual(projected);
} finally {
await codex.close();
@@ -2548,6 +2565,77 @@ describe("Codex app-server provider", () => {
await session.close();
});
test("keeps a background terminal controllable on its original thread after rewind", async () => {
let live = true;
const appServer = createFakeCodexAppServer({
initialize: () => ({ userAgent: "paseo/0.159.0 (Ubuntu 26.4.0; x86_64) (paseo; 0)" }),
"thread/read": () => ({ thread: { id: "thread-1", historyMode: "legacy", turns: [] } }),
"thread/backgroundTerminals/list": (params) => {
const threadId = (params as { threadId: string }).threadId;
if (threadId === "thread-1" && live)
return { data: [{ processId: "12", command: "sleep 180" }], nextCursor: null };
if (threadId === "forked-thread")
return { data: [{ processId: "12", command: "sleep 99" }], nextCursor: null };
return { data: [], nextCursor: null };
},
"thread/backgroundTerminals/terminate": (params) => {
expect(params).toMatchObject({ threadId: "thread-1", processId: "12" });
live = false;
return { terminated: true };
},
});
const session = new CodexAppServerAgentSession(
createConfig({ cwd: "/workspace/project" }),
null,
createTestLogger(),
async () => appServer.child,
);
const events: AgentStreamEvent[] = [];
session.subscribe((event) => events.push(event));
try {
await session.startTurn("first");
emitCodexUserMessage(appServer, { id: "first", text: "first", turnId: "turn-first" });
appServer.completeTurn();
await vi.waitFor(() =>
expect(latestBackgroundWork(events)).toMatchObject({
items: [{ command: "sleep 180" }],
}),
);
await new Promise<void>((resolve) => setImmediate(resolve));
await session.startTurn("second");
emitCodexUserMessage(appServer, { id: "second", text: "second", turnId: "turn-second" });
appServer.completeTurn();
await new Promise<void>((resolve) => setImmediate(resolve));
await new Promise<void>((resolve) => setImmediate(resolve));
await session.revertConversation({ messageId: "second" });
await session.startTurn("continue");
appServer.startsTerminalCommand({
threadId: "forked-thread",
turnId: "turn-next",
itemId: "new",
processId: "12",
command: "echo hello",
});
await vi.waitFor(() =>
expect(hasBackgroundListRequest(appServer, "forked-thread")).toBe(true),
);
await vi.waitFor(() =>
expect(latestBackgroundWork(events)).toMatchObject({
items: [{ command: "sleep 180" }, { command: "sleep 99" }],
}),
);
const items = latestBackgroundWork(events)?.items;
expect(items).toHaveLength(2);
expect(items![0].id).not.toBe(items![1].id);
await session.stopBackgroundWork(items![0].id);
expect(latestBackgroundWork(events)).toMatchObject({
items: [{ command: "sleep 99" }],
});
} finally {
await session.close();
}
});
test("rewinds a paginated conversation through the public session capability", async () => {
const appServer = createFakeCodexAppServer({
...fakeCodexThreadHistory({
@@ -7816,7 +7904,7 @@ describe("Codex interactive resume startup", () => {
unsubscribe();
const laterEvents: AgentStreamEvent[] = [];
session.subscribe((event) => laterEvents.push(event));
expect(laterEvents).toEqual([]);
expect(laterEvents.filter((event) => event.type !== "background_work")).toEqual([]);
} finally {
await session.close();
appServer.assertNoErrors();
@@ -1,3 +1,4 @@
import { CodexBackgroundInventory } from "./codex/background-inventory.js";
import { validateProviderOptions } from "../provider-options.js";
import {
getAgentStreamEventTurnId,
@@ -3494,6 +3495,7 @@ export class CodexAppServerAgentSession implements AgentSession {
cancelRequested: boolean;
} | null = null;
private client: CodexAppServerClient | null = null;
private readonly background = new CodexBackgroundInventory((event) => this.emitEvent(event));
private threadRollbackAvailable = true;
private readonly subscribers = new Set<(event: AgentStreamEvent) => void>();
private startupEvents: AgentStreamEvent[] | null = null;
@@ -3717,6 +3719,7 @@ export class CodexAppServerAgentSession implements AgentSession {
throw this.createClosedError();
}
this.connectionState = "connected";
this.background.start({ client, thread: this.currentThreadId });
} catch (error) {
try {
if (this.client === client) {
@@ -3768,6 +3771,7 @@ export class CodexAppServerAgentSession implements AgentSession {
private handleUnexpectedTermination(error: Error): void {
this.connectionState = "disconnected";
this.background.stop();
const hasActiveRootTurn = this.activeForegroundTurnId !== null || this.currentTurnId !== null;
this.clearPendingPermissions({ preservePlanApprovals: !hasActiveRootTurn });
if (hasActiveRootTurn) {
@@ -4634,6 +4638,8 @@ export class CodexAppServerAgentSession implements AgentSession {
const startupEvents = this.startupEvents;
this.startupEvents = null;
for (const event of startupEvents ?? []) this.notifySubscribers(event);
const background = this.background.snapshot();
if (background) callback(background);
return () => {
this.subscribers.delete(callback);
};
@@ -5125,6 +5131,10 @@ export class CodexAppServerAgentSession implements AgentSession {
}
}
stopBackgroundWork(id: string): Promise<void> {
return this.background.stopItem(id);
}
async canStopRuntime(): Promise<boolean> {
if (
!this.client ||
@@ -5286,6 +5296,7 @@ export class CodexAppServerAgentSession implements AgentSession {
async close(): Promise<void> {
this.closed = true;
this.background.stop();
this.clearPendingPermissions();
this.pendingToolCallDetails.clear();
this.pendingSubAgentNotificationsByThreadId.clear();
@@ -5316,6 +5327,7 @@ export class CodexAppServerAgentSession implements AgentSession {
}
private async disposeClient(): Promise<void> {
this.background.stop();
const client = this.client;
this.connectionState = "disconnected";
this.currentTurnId = null;
@@ -5714,6 +5726,14 @@ export class CodexAppServerAgentSession implements AgentSession {
await this.dispatchSubAgentNotification(parsed, route.callId);
return;
}
if (
parsed.kind === "turn_completed" ||
parsed.kind === "exec_command_started" ||
parsed.kind === "exec_command_completed" ||
((parsed.kind === "item_started" || parsed.kind === "item_completed") &&
normalizeCodexThreadItemType(parsed.item.type) === "commandExecution")
)
this.background.changed(this.currentThreadId);
await this.dispatchParsedNotification(parsed);
}
@@ -6353,6 +6373,7 @@ export class CodexAppServerAgentSession implements AgentSession {
parsed: Extract<ParsedCodexNotification, { kind: "thread_started" }>,
): void {
this.currentThreadId = parsed.threadId;
this.background.changed(this.currentThreadId);
this.emitEvent({
type: "thread_started",
provider: CODEX_PROVIDER,
@@ -45,6 +45,7 @@ function createHarness(
} = {},
) {
const handlers: Record<string, Handler> = {
"thread/backgroundTerminals/list": () => ({ data: [], nextCursor: null }),
"turn/interrupt": () => {
const completeTurn = () =>
appServer.completeTurn({ threadId: THREAD, status: "interrupted" });
@@ -169,7 +170,11 @@ describe("an interrupted Codex turn's commands end before the next turn", () =>
expect(
harness.appServer
.requests()
.filter((message) => String(message.method).startsWith("thread/backgroundTerminals/"))
.filter(
(message) =>
String(message.method).startsWith("thread/backgroundTerminals/") &&
message.method !== "thread/backgroundTerminals/list",
)
.map((message) => message.method),
).toEqual([TERMINATE, TERMINATE]);
harness.appServer.assertNoErrors();
@@ -372,3 +377,328 @@ describe("an interrupted Codex turn's commands end before the next turn", () =>
expect(harness.terminated()).toEqual(["201", "202"]);
}, 15_000);
});
test("publishes every page of the current thread's live background terminals", async () => {
const appServer = createFakeCodexAppServer({
"thread/backgroundTerminals/list": (params) => {
const cursor = paramsOf(params).cursor;
return cursor
? { data: [{ itemId: "item-2", processId: "2", command: "sleep 2" }], nextCursor: null }
: { data: [{ itemId: "item-1", processId: "1", command: "sleep 1" }], nextCursor: "next" };
},
});
const session = new CodexAppServerAgentSession(
{ provider: "codex", cwd: "/workspace/project", modeId: "auto", model: "gpt-5.4" },
{ sessionId: THREAD },
createTestLogger(),
async () => appServer.child,
);
sessions.push(session);
const events: import("../agent-sdk-types.js").AgentStreamEvent[] = [];
session.subscribe((event) => events.push(event));
await session.connect();
await vi.waitFor(() =>
expect(events.findLast((event) => event.type === "background_work")).toMatchObject({
items: [
{ kind: "shell", description: null, command: "sleep 1" },
{ kind: "shell", description: null, command: "sleep 2" },
],
}),
);
});
test("discards a terminal list overtaken by a command event", async () => {
const first = deferred<unknown>();
let lists = 0;
const appServer = createFakeCodexAppServer({
"thread/backgroundTerminals/list": () =>
++lists === 1 ? first.promise : { data: [], nextCursor: null },
});
const session = new CodexAppServerAgentSession(
{ provider: "codex", cwd: "/workspace/project", modeId: "auto", model: "gpt-5.4" },
{ sessionId: THREAD },
createTestLogger(),
async () => appServer.child,
);
sessions.push(session);
const events: import("../agent-sdk-types.js").AgentStreamEvent[] = [];
session.subscribe((event) => events.push(event));
await session.connect();
await vi.waitFor(() => expect(lists).toBe(1));
const shown = waitForTimelineToolCall(session, "item-new");
appServer.startsTerminalCommand({
threadId: THREAD,
turnId: "turn",
itemId: "item-new",
processId: "new",
command: "sleep 1",
});
await shown;
first.resolve({
data: [{ itemId: "old", processId: "old", command: "sleep 2" }],
nextCursor: null,
});
await vi.waitFor(() =>
expect(events.findLast((event) => event.type === "background_work")).toMatchObject({
items: [],
}),
);
expect(events.filter((event) => event.type === "background_work")).not.toContainEqual(
expect.objectContaining({ items: [expect.objectContaining({ command: "sleep 2" })] }),
);
});
test("does not publish a terminal response from a lost runtime", async () => {
const first = deferred<unknown>();
const appServer = createFakeCodexAppServer({
"thread/backgroundTerminals/list": () => first.promise,
});
const session = new CodexAppServerAgentSession(
{ provider: "codex", cwd: "/workspace/project", modeId: "auto", model: "gpt-5.4" },
{ sessionId: THREAD },
createTestLogger(),
async () => appServer.child,
);
sessions.push(session);
const events: import("../agent-sdk-types.js").AgentStreamEvent[] = [];
session.subscribe((event) => events.push(event));
await session.connect();
await appServer.waitForRequest("thread/backgroundTerminals/list");
appServer.disconnect();
first.resolve({
data: [{ itemId: "old", processId: "old", command: "sleep 2" }],
nextCursor: null,
});
await new Promise((resolve) => setTimeout(resolve, 50));
expect(events.filter((event) => event.type === "background_work")).not.toContainEqual(
expect.objectContaining({ items: [expect.objectContaining({ command: "sleep 2" })] }),
);
});
test("accepts a slow terminal snapshot after the polling interval", async () => {
const pending = deferred<unknown>();
const appServer = createFakeCodexAppServer({
"thread/backgroundTerminals/list": () => pending.promise,
});
const session = new CodexAppServerAgentSession(
{ provider: "codex", cwd: "/workspace/project", model: "gpt-5.4" },
{ sessionId: THREAD },
createTestLogger(),
async () => appServer.child,
);
sessions.push(session);
const events: import("../agent-sdk-types.js").AgentStreamEvent[] = [];
session.subscribe((event) => events.push(event));
await session.connect();
await appServer.waitForRequest("thread/backgroundTerminals/list");
await new Promise((resolve) => setTimeout(resolve, 5200));
expect(
appServer.requests().filter((message) => message.method === "thread/backgroundTerminals/list"),
).toHaveLength(1);
pending.resolve({ data: [{ processId: "slow", command: "sleep 120" }], nextCursor: null });
await vi.waitFor(() =>
expect(events.findLast((event) => event.type === "background_work")).toMatchObject({
items: [{ command: "sleep 120" }],
}),
);
expect(
appServer.requests().filter((message) => message.method === "thread/backgroundTerminals/list"),
).toHaveLength(1);
}, 12000);
test("does not retry an unsupported inventory on later command events", async () => {
let calls = 0;
const appServer = createFakeCodexAppServer({
"thread/backgroundTerminals/list": () => {
calls++;
return { __jsonRpcError: { code: -32601, message: "Method not found" } };
},
});
const session = new CodexAppServerAgentSession(
{ provider: "codex", cwd: "/workspace/project", model: "gpt-5.4" },
{ sessionId: THREAD },
createTestLogger(),
async () => appServer.child,
);
sessions.push(session);
await session.connect();
await vi.waitFor(() => expect(calls).toBe(1));
const shown = waitForTimelineToolCall(session, "unsupported-item");
appServer.startsTerminalCommand({
threadId: THREAD,
turnId: "turn",
itemId: "unsupported-item",
processId: "1",
command: "sleep 1",
});
await shown;
await new Promise((resolve) => setTimeout(resolve, 100));
expect(calls).toBe(1);
});
test("retries transient inventory failures and cancels retries on stop", async () => {
vi.useFakeTimers();
try {
const { CodexBackgroundInventory } = await import("./codex/background-inventory.js");
let calls = 0;
let fail = true;
const inventory = new CodexBackgroundInventory(() => {});
inventory.start({
client: {
request: async () => {
calls++;
if (fail) throw new Error("temporary");
return { data: [{ processId: "live", command: "sleep 120" }], nextCursor: null };
},
},
thread: THREAD,
});
await vi.advanceTimersByTimeAsync(0);
expect(inventory.snapshot()).toBeNull();
inventory.changed(THREAD);
await vi.advanceTimersByTimeAsync(0);
expect(calls).toBe(1);
fail = false;
await vi.advanceTimersByTimeAsync(5000);
expect(inventory.snapshot()?.items).toMatchObject([{ command: "sleep 120" }]);
fail = true;
await vi.advanceTimersByTimeAsync(5000);
expect(inventory.snapshot()?.items).toMatchObject([{ command: "sleep 120" }]);
inventory.stop();
const stoppedCalls = calls;
await vi.advanceTimersByTimeAsync(15000);
expect(calls).toBe(stoppedCalls);
} finally {
vi.useRealTimers();
}
});
test("does not retry a failure from a stopped inventory generation", async () => {
vi.useFakeTimers();
try {
const { CodexBackgroundInventory } = await import("./codex/background-inventory.js");
let reject!: (error: Error) => void;
let calls = 0;
const response = new Promise<unknown>((_, fail) => {
reject = fail;
});
const inventory = new CodexBackgroundInventory(() => {});
inventory.start({
client: {
request: () => {
calls++;
return response;
},
},
thread: THREAD,
});
inventory.stop();
reject(new Error("late failure"));
await vi.advanceTimersByTimeAsync(15000);
expect(calls).toBe(1);
expect(inventory.snapshot()).toBeNull();
} finally {
vi.useRealTimers();
}
});
test("stops only the selected live background terminal and confirms the roster", async () => {
const live = new Set(["1", "2"]);
const appServer = createFakeCodexAppServer({
"thread/backgroundTerminals/list": () => ({
data: [...live].map((processId) => ({ processId, command: `sleep ${processId}` })),
nextCursor: null,
}),
"thread/backgroundTerminals/terminate": (params) => ({
terminated: live.delete(String(paramsOf(params).processId)),
}),
});
const session = new CodexAppServerAgentSession(
{ provider: "codex", cwd: "/workspace/project", model: "gpt-5.4" },
{ sessionId: THREAD },
createTestLogger(),
async () => appServer.child,
);
sessions.push(session);
const events: import("../agent-sdk-types.js").AgentStreamEvent[] = [];
session.subscribe((event) => events.push(event));
await session.connect();
await vi.waitFor(() =>
expect(events.findLast((event) => event.type === "background_work")).toMatchObject({
items: [{ command: "sleep 1" }, { command: "sleep 2" }],
}),
);
const item = events
.findLast((event) => event.type === "background_work")
?.items?.find((entry) => entry.command === "sleep 1");
expect(item).toBeDefined();
await session.stopBackgroundWork(item!.id);
expect(events.findLast((event) => event.type === "background_work")).toMatchObject({
items: [{ command: "sleep 2" }],
});
await expect(session.stopBackgroundWork(item!.id)).rejects.toThrow();
});
test("reports a rejected background stop and keeps the item available", async () => {
const appServer = createFakeCodexAppServer({
"thread/backgroundTerminals/list": () => ({
data: [{ processId: "1", command: "sleep 1" }],
nextCursor: null,
}),
"thread/backgroundTerminals/terminate": () => ({ terminated: false }),
});
const session = new CodexAppServerAgentSession(
{ provider: "codex", cwd: "/workspace/project", model: "gpt-5.4" },
{ sessionId: THREAD },
createTestLogger(),
async () => appServer.child,
);
sessions.push(session);
const events: import("../agent-sdk-types.js").AgentStreamEvent[] = [];
session.subscribe((event) => events.push(event));
await session.connect();
await vi.waitFor(() =>
expect(events.findLast((event) => event.type === "background_work")).toMatchObject({
items: [{ command: "sleep 1" }],
}),
);
const item = events.findLast((event) => event.type === "background_work")?.items?.[0];
expect(item).toBeDefined();
await expect(session.stopBackgroundWork(item!.id)).rejects.toThrow("not terminated");
expect(events.findLast((event) => event.type === "background_work")).toMatchObject({
items: [{ command: "sleep 1" }],
});
});
test("does not confirm a stop after the provider connection is lost", async () => {
const response = deferred<unknown>();
const appServer = createFakeCodexAppServer({
"thread/backgroundTerminals/list": () => ({
data: [{ processId: "1", command: "sleep 1" }],
nextCursor: null,
}),
"thread/backgroundTerminals/terminate": () => response.promise,
});
const session = new CodexAppServerAgentSession(
{ provider: "codex", cwd: "/workspace/project", model: "gpt-5.4" },
{ sessionId: THREAD },
createTestLogger(),
async () => appServer.child,
);
sessions.push(session);
const events: import("../agent-sdk-types.js").AgentStreamEvent[] = [];
session.subscribe((event) => events.push(event));
await session.connect();
await vi.waitFor(() =>
expect(events.findLast((event) => event.type === "background_work")).toMatchObject({
items: [{ command: "sleep 1" }],
}),
);
const item = events.findLast((event) => event.type === "background_work")?.items?.[0];
expect(item).toBeDefined();
const pending = session.stopBackgroundWork(item!.id);
await appServer.waitForRequest("thread/backgroundTerminals/terminate");
appServer.disconnect();
response.resolve({ terminated: true });
await expect(pending).rejects.toThrow();
});
@@ -0,0 +1,187 @@
import { z } from "zod";
import type { AgentStreamEvent } from "../../agent-sdk-types.js";
import { CodexAppServerRpcError } from "./app-server-transport.js";
type InventoryEvent = Extract<AgentStreamEvent, { type: "background_work" }>;
interface Client {
request(method: string, params?: unknown, timeoutMs?: number): Promise<unknown>;
}
export class CodexBackgroundInventory {
private generation = 0;
private client: Client | null = null;
private thread: string | null = null;
private readonly threads = new Set<string>();
private origins = new Map<string, { threadId: string; processId: string }>();
private pending = false;
private dirty = false;
private unsupported = false;
private retrying = false;
private timer: ReturnType<typeof setTimeout> | null = null;
private current: InventoryEvent | null = null;
private readonly listeners = new Set<() => void>();
async stopItem(id: string): Promise<void> {
const client = this.client;
const origin = this.origins.get(id);
const generation = this.generation;
if (!client || !origin || !this.current?.items?.some((item) => item.id === id))
throw new Error("Background terminal is no longer available");
const result = z
.object({ terminated: z.boolean() })
.parse(
await client.request(
"thread/backgroundTerminals/terminate",
{ threadId: origin.threadId, processId: origin.processId },
10000,
),
);
if (generation !== this.generation) throw new Error("Background terminal connection changed");
if (!result.terminated) throw new Error("Background terminal was not terminated");
await this.waitUntilAbsent(id, generation);
}
private waitUntilAbsent(id: string, generation: number): Promise<void> {
return new Promise((resolve, reject) => {
const timer = setTimeout(
() => finish(new Error("Background terminal confirmation timed out")),
15000,
);
const finish = (error?: Error) => {
clearTimeout(timer);
this.listeners.delete(check);
if (error) reject(error);
else resolve();
};
const check = () => {
if (generation !== this.generation)
finish(new Error("Background terminal connection changed"));
else if (this.current?.items && !this.current.items.some((item) => item.id === id))
finish();
};
this.listeners.add(check);
this.changed(this.thread);
});
}
private notify() {
for (const listener of this.listeners) listener();
}
constructor(private readonly emit: (event: InventoryEvent) => void) {}
snapshot() {
return this.current;
}
start({ client, thread }: { client: Client; thread: string | null }) {
this.stop();
this.client = client;
this.thread = thread;
this.unsupported = false;
this.changed(thread);
}
stop() {
this.generation++;
this.notify();
this.client = null;
this.thread = null;
this.threads.clear();
this.origins.clear();
this.pending = false;
this.dirty = false;
this.retrying = false;
if (this.timer) clearTimeout(this.timer);
this.timer = null;
if (this.current) this.emit({ type: "background_work", provider: "codex", items: null });
this.current = null;
}
changed(thread: string | null) {
this.thread = thread;
if (thread) this.threads.add(thread);
if (!this.client || !thread || this.unsupported) return;
this.dirty = true;
if (!this.retrying) this.refresh();
}
private refresh() {
if (this.pending || !this.client || !this.thread || this.unsupported) return;
if (this.timer) clearTimeout(this.timer);
this.timer = null;
this.pending = true;
this.dirty = false;
const generation = this.generation;
const client = this.client;
const threads = [...this.threads];
void (async () => {
let failed = false;
try {
const { items, origins, activeThreads } = await readInventories(client, threads);
if (generation !== this.generation || this.dirty) return;
this.origins = origins;
this.threads.clear();
for (const threadId of activeThreads) this.threads.add(threadId);
if (this.thread) this.threads.add(this.thread);
this.current = { type: "background_work", provider: "codex", items };
this.emit(this.current);
this.notify();
} catch (error) {
failed = true;
if (
generation === this.generation &&
error instanceof CodexAppServerRpcError &&
error.code === -32601
)
this.unsupported = true;
} finally {
if (generation === this.generation) {
this.pending = false;
if (this.dirty && !failed && !this.unsupported) this.refresh();
else if (!this.unsupported && (failed || this.current?.items?.length)) {
this.retrying = failed;
this.timer = setTimeout(() => {
this.timer = null;
this.retrying = false;
this.refresh();
}, 5000);
this.timer.unref();
}
}
}
})();
}
}
async function readInventories(client: Client, threads: string[]) {
const items: NonNullable<InventoryEvent["items"]> = [];
const origins = new Map<string, { threadId: string; processId: string }>();
const activeThreads = new Set<string>();
for (const threadId of threads) {
let cursor: string | null = null;
do {
const page = z
.object({
data: z.array(z.object({ processId: z.string(), command: z.string() })),
nextCursor: z.string().nullable(),
})
.parse(
await client.request(
"thread/backgroundTerminals/list",
{ threadId, ...(cursor ? { cursor } : {}) },
10_000,
),
);
for (const row of page.data) {
const id = JSON.stringify([threadId, row.processId]);
origins.set(id, { threadId, processId: row.processId });
activeThreads.add(threadId);
items.push({
id,
kind: "shell",
canStop: true,
description: null,
...(row.command ? { command: row.command } : {}),
});
}
cursor = page.nextCursor;
} while (cursor);
}
return { items, origins, activeThreads };
}
@@ -175,6 +175,7 @@ export function createFakeCodexAppServer(
"skills/list": () => ({ data: [] }),
"thread/start": () => ({ thread: { id: "thread-1" } }),
"thread/loaded/list": () => ({ data: [] }),
"thread/backgroundTerminals/list": () => ({ data: [], nextCursor: null }),
"thread/resume": () => ({}),
"turn/start": () => ({}),
"thread/fork": (params) => ({
@@ -91,6 +91,7 @@ export class V2Harness {
readonly api: V2Api = {
server: { info: unexpected },
shell: {
remove: unexpected,
list: async () => ({ location: { directory: this.info.location.directory }, data: [] }),
},
plugin: {
@@ -1201,3 +1201,773 @@ test("OpenCode v2 runtime safety follows descendant locations before releasing t
await session.close();
}
});
test("reconstructs only promoted session-owned running shells and reconciles exits", async () => {
const harness = new V2Harness();
harness.history.push({
id: "assistant",
type: "assistant",
agent: "build",
model: { providerID: "test", modelID: "test" },
time: { created: 1 },
content: [
{
type: "tool",
id: "tool",
name: "shell",
time: { created: 1 },
state: {
status: "completed",
input: { description: "Watch tests", command: "sleep ours" },
content: [{ type: "text", text: "" }],
metadata: { status: "running", shellID: "ours" },
},
},
],
});
harness.api.message.list = async (input) =>
input.cursor
? { data: [...harness.history], cursor: {} }
: { data: [], cursor: { next: "second-page" } };
let live = ["ours", "other"];
harness.api.shell.list = async () => ({
location: harness.info.location,
data: live.map((id) => ({
id,
status: "running" as const,
command: `sleep ${id}`,
cwd: "/tmp/project",
shell: "sh",
file: "sh",
metadata: { sessionID: id === "ours" ? harness.info.id : "other" },
time: { started: 1 },
})),
});
const client = new OpenCodeV2AgentClient({
logger: createTestLogger(),
runtime: harness.runtime,
});
const session = await client.createSession({ provider: "opencode", cwd: "/tmp/project" });
const events: AgentStreamEvent[] = [];
session.subscribe((event) => events.push(event));
try {
await expect
.poll(() => events.findLast((event) => event.type === "background_work"))
.toMatchObject({
items: [{ id: "ours", kind: "shell", description: "Watch tests", command: "sleep ours" }],
});
live = [];
harness.push({ id: "reconnect", created: 2, type: "server.connected", data: {} });
await expect
.poll(() => events.findLast((event) => event.type === "background_work"))
.toMatchObject({ items: [] });
} finally {
await session.close();
}
});
test("an exit received during shell listing cannot restore an old row", async () => {
const harness = new V2Harness();
harness.history.push({
id: "assistant",
type: "assistant",
agent: "build",
model: { providerID: "test", modelID: "test" },
time: { created: 1 },
content: [
{
type: "tool",
id: "tool",
name: "shell",
time: { created: 1 },
state: {
status: "completed",
input: {},
content: [{ type: "text", text: "" }],
metadata: { status: "running", shellID: "shell" },
},
},
],
});
const shell = {
id: "shell",
status: "running" as const,
command: "sleep 120",
cwd: "/tmp/project",
shell: "sh",
file: "sh",
metadata: { sessionID: harness.info.id },
time: { started: 1 },
};
let listCount = 0;
let release!: () => void;
const gate = new Promise<void>((resolve) => {
release = resolve;
});
harness.api.shell.list = async () => {
listCount++;
if (listCount === 2) {
await gate;
return { location: harness.info.location, data: [shell] };
}
return { location: harness.info.location, data: [] };
};
const client = new OpenCodeV2AgentClient({
logger: createTestLogger(),
runtime: harness.runtime,
});
const session = await client.createSession({ provider: "opencode", cwd: "/tmp/project" });
const events: AgentStreamEvent[] = [];
session.subscribe((event) => events.push(event));
try {
harness.push({ id: "created", created: 2, type: "shell.created", data: { info: shell } });
await expect.poll(() => listCount).toBe(2);
harness.push({
id: "exited",
created: 3,
type: "shell.exited",
data: { id: "shell", status: "exited" },
});
await new Promise((resolve) => setTimeout(resolve, 100));
release();
await expect.poll(() => listCount).toBeGreaterThan(2);
await expect
.poll(() => events.findLast((event) => event.type === "background_work"))
.toMatchObject({ items: [] });
expect(events.filter((event) => event.type === "background_work")).not.toContainEqual(
expect.objectContaining({ items: [expect.objectContaining({ id: "shell" })] }),
);
} finally {
release();
await session.close();
}
});
test("does not publish a shell list from a lost server generation", async () => {
const harness = new V2Harness();
let exit!: (error: Error) => void;
Object.assign(harness.connection, {
exited: new Promise<Error>((resolve) => {
exit = resolve;
}),
});
harness.history.push({
id: "assistant",
type: "assistant",
agent: "build",
model: { providerID: "test", modelID: "test" },
time: { created: 1 },
content: [
{
type: "tool",
id: "tool",
name: "shell",
time: { created: 1 },
state: {
status: "completed",
input: {},
content: [{ type: "text", text: "" }],
metadata: { status: "running", shellID: "old" },
},
},
],
});
const shell = {
id: "old",
status: "running" as const,
command: "sleep 120",
cwd: "/tmp/project",
shell: "sh",
file: "sh",
metadata: { sessionID: harness.info.id },
time: { started: 1 },
};
let reads = 0;
let release!: () => void;
const gate = new Promise<void>((resolve) => {
release = resolve;
});
harness.api.shell.list = async () => {
reads++;
if (reads === 2) {
await gate;
return { location: harness.info.location, data: [shell] };
}
return { location: harness.info.location, data: [] };
};
const client = new OpenCodeV2AgentClient({
logger: createTestLogger(),
runtime: harness.runtime,
});
const session = await client.createSession({ provider: "opencode", cwd: "/tmp/project" });
const events: AgentStreamEvent[] = [];
session.subscribe((event) => events.push(event));
try {
harness.push({ id: "created", created: 2, type: "shell.created", data: { info: shell } });
await expect.poll(() => reads).toBe(2);
exit(new Error("server exited"));
await expect
.poll(() => events.findLast((event) => event.type === "background_work"))
.toMatchObject({ items: null });
release();
await new Promise((resolve) => setTimeout(resolve, 100));
expect(events.filter((event) => event.type === "background_work")).not.toContainEqual(
expect.objectContaining({ items: [expect.objectContaining({ id: "old" })] }),
);
} finally {
release();
await session.close();
}
});
test("polls a live shell without rereading conversation history", async () => {
const harness = new V2Harness();
harness.history.push({
id: "assistant",
type: "assistant",
agent: "build",
model: { providerID: "test", modelID: "test" },
time: { created: 1 },
content: [
{
type: "tool",
id: "tool",
name: "shell",
time: { created: 1 },
state: {
status: "completed",
input: {},
content: [{ type: "text", text: "" }],
metadata: { status: "running", shellID: "live" },
},
},
],
});
let historyReads = 0;
let shellReads = 0;
const readHistory = harness.api.message.list;
harness.api.message.list = async (...args) => {
historyReads++;
return readHistory(...args);
};
harness.api.shell.list = async () => {
shellReads++;
return {
location: harness.info.location,
data: [
{
id: "live",
status: "running",
command: "sleep 120",
cwd: "/tmp/project",
shell: "sh",
file: "sh",
metadata: { sessionID: harness.info.id },
time: { started: 1 },
},
],
};
};
const client = new OpenCodeV2AgentClient({
logger: createTestLogger(),
runtime: harness.runtime,
});
const session = await client.createSession({ provider: "opencode", cwd: "/tmp/project" });
const events: AgentStreamEvent[] = [];
session.subscribe((event) => events.push(event));
try {
await expect
.poll(() => events.findLast((event) => event.type === "background_work"))
.toMatchObject({ items: [{ id: "live" }] });
const initialHistoryReads = historyReads;
const initialShellReads = shellReads;
await expect.poll(() => shellReads, { timeout: 7000 }).toBeGreaterThan(initialShellReads);
expect(historyReads).toBe(initialHistoryReads);
} finally {
await session.close();
}
}, 12000);
test("buffers tool promotion and shell events during background history bootstrap", async () => {
const harness = new V2Harness();
let historyReads = 0;
let release!: () => void;
const gate = new Promise<void>((resolve) => {
release = resolve;
});
harness.api.message.list = async () => {
historyReads++;
if (historyReads === 3) await gate;
return { data: [], cursor: {} };
};
const shell = {
id: "promoted",
status: "running" as const,
command: "sleep 120",
cwd: "/tmp/project",
shell: "sh",
file: "sh",
metadata: { sessionID: harness.info.id },
time: { started: 1 },
};
harness.api.shell.list = async () => ({ location: harness.info.location, data: [shell] });
const client = new OpenCodeV2AgentClient({
logger: createTestLogger(),
runtime: harness.runtime,
});
const session = await client.createSession({ provider: "opencode", cwd: "/tmp/project" });
const events: AgentStreamEvent[] = [];
session.subscribe((event) => events.push(event));
try {
await expect.poll(() => historyReads).toBeGreaterThanOrEqual(3);
harness.push({
id: "input",
created: 2,
type: "session.tool.input.started",
durable: { aggregateID: harness.info.id, seq: 1, version: 1 },
data: { sessionID: harness.info.id, assistantMessageID: "msg", id: "tool", name: "shell" },
});
harness.push({
id: "input-ended",
created: 3,
type: "session.tool.input.ended",
durable: { aggregateID: harness.info.id, seq: 2, version: 1 },
data: {
sessionID: harness.info.id,
assistantMessageID: "msg",
id: "tool",
text: JSON.stringify({ description: "Watch tests", command: "sleep 120" }),
},
});
harness.push({ id: "created", created: 3, type: "shell.created", data: { info: shell } });
harness.push({
id: "success",
created: 4,
type: "session.tool.success",
durable: { aggregateID: harness.info.id, seq: 3, version: 2 },
data: {
sessionID: harness.info.id,
assistantMessageID: "msg",
id: "tool",
executed: true,
content: [{ type: "text", text: "" }],
metadata: { status: "running", shellID: "promoted" },
},
});
await new Promise((resolve) => setTimeout(resolve, 100));
expect(events.filter((event) => event.type === "background_work")).toEqual([]);
release();
await expect
.poll(() => events.findLast((event) => event.type === "background_work"))
.toMatchObject({
items: [{ id: "promoted", description: "Watch tests", command: "sleep 120" }],
});
harness.push({
id: "next-input",
created: 5,
type: "session.tool.input.started",
durable: { aggregateID: harness.info.id, seq: 4, version: 1 },
data: { sessionID: harness.info.id, assistantMessageID: "next", id: "tool", name: "shell" },
});
harness.push({
id: "next-success",
created: 6,
type: "session.tool.success",
durable: { aggregateID: harness.info.id, seq: 5, version: 2 },
data: {
sessionID: harness.info.id,
assistantMessageID: "next",
id: "tool",
executed: true,
content: [{ type: "text", text: "" }],
metadata: { status: "running", shellID: "second" },
},
});
harness.api.shell.list = async () => ({
location: harness.info.location,
data: [shell, { ...shell, id: "second" }],
});
await expect
.poll(() => events.findLast((event) => event.type === "background_work"))
.toMatchObject({ items: [{ id: "promoted" }, { id: "second" }] });
} finally {
release();
await session.close();
}
});
test("retries failed background bootstrap and roster without losing stale work", async () => {
const { vi } = await import("vitest");
vi.useFakeTimers();
try {
const { OpenCodeBackgroundInventory } = await import("./background-inventory.js");
const harness = new V2Harness();
harness.history.push({
id: "assistant",
type: "assistant",
agent: "build",
model: { providerID: "test", modelID: "test" },
time: { created: 1 },
content: [
{
type: "tool",
id: "tool",
name: "shell",
time: { created: 1 },
state: {
status: "completed",
input: {},
content: [{ type: "text", text: "" }],
metadata: { status: "running", shellID: "live" },
},
},
],
});
let historyCalls = 0;
let shellCalls = 0;
let historyFails = true;
let shellFails = true;
harness.api.message.list = async () => {
historyCalls++;
if (historyFails) throw new Error("history unavailable");
return { data: harness.history, cursor: {} };
};
harness.api.shell.list = async () => {
shellCalls++;
if (shellFails) throw new Error("roster unavailable");
return {
location: harness.info.location,
data: [
{
id: "live",
status: "running",
command: "sleep 120",
cwd: "/tmp/project",
shell: "sh",
file: "sh",
metadata: { sessionID: harness.info.id },
time: { started: 1 },
},
],
};
};
const inventory = new OpenCodeBackgroundInventory({
id: harness.info.id,
cwd: "/tmp/project",
emit: () => {},
report: () => {},
});
inventory.start(harness.api);
await vi.advanceTimersByTimeAsync(0);
expect(inventory.snapshot()).toBeNull();
historyFails = false;
await vi.advanceTimersByTimeAsync(5000);
expect(historyCalls).toBe(2);
expect(shellCalls).toBe(1);
expect(inventory.snapshot()).toBeNull();
inventory.observe({ id: "deleted", created: 1, type: "shell.deleted", data: { id: "other" } });
await vi.advanceTimersByTimeAsync(0);
expect(shellCalls).toBe(1);
shellFails = false;
await vi.advanceTimersByTimeAsync(5000);
expect(historyCalls).toBe(2);
expect(inventory.snapshot()?.items).toMatchObject([{ id: "live", canStop: true }]);
historyFails = true;
inventory.start(harness.api);
await vi.advanceTimersByTimeAsync(0);
expect(inventory.snapshot()?.items).toEqual([
{ id: "live", kind: "shell", description: null, command: "sleep 120" },
]);
await expect(inventory.stopItem("live")).rejects.toThrow("no longer available");
historyFails = false;
await vi.advanceTimersByTimeAsync(5000);
expect(historyCalls).toBe(4);
expect(inventory.snapshot()?.items).toMatchObject([{ id: "live", canStop: true }]);
shellFails = true;
await vi.advanceTimersByTimeAsync(5000);
expect(inventory.snapshot()?.items).toMatchObject([{ id: "live" }]);
inventory.disconnect();
const stoppedCalls = shellCalls;
await vi.advanceTimersByTimeAsync(15000);
expect(shellCalls).toBe(stoppedCalls);
} finally {
vi.useRealTimers();
}
});
test("does not retry a failed bootstrap from a disconnected generation", async () => {
const { vi } = await import("vitest");
vi.useFakeTimers();
try {
const { OpenCodeBackgroundInventory } = await import("./background-inventory.js");
const harness = new V2Harness();
let reject!: (error: Error) => void;
let calls = 0;
const response = new Promise<Awaited<ReturnType<typeof harness.api.message.list>>>(
(_, fail) => {
reject = fail;
},
);
harness.api.message.list = () => {
calls++;
return response;
};
const inventory = new OpenCodeBackgroundInventory({
id: harness.info.id,
cwd: "/tmp/project",
emit: () => {},
report: () => {},
});
inventory.start(harness.api);
inventory.disconnect();
reject(new Error("late failure"));
await vi.advanceTimersByTimeAsync(15000);
expect(calls).toBe(1);
expect(inventory.snapshot()).toBeNull();
} finally {
vi.useRealTimers();
}
});
test("stops a selected promoted shell and retains its sibling", async () => {
const harness = new V2Harness();
harness.history.push({
id: "assistant",
type: "assistant",
agent: "build",
model: { providerID: "test", modelID: "test" },
time: { created: 1 },
content: ["one", "two"].map((id) => ({
type: "tool" as const,
id,
name: "shell",
time: { created: 1 },
state: {
status: "completed" as const,
input: {},
content: [{ type: "text" as const, text: "" }],
metadata: { status: "running", shellID: id },
},
})),
});
let live = ["one", "two"];
harness.api.shell.list = async () => ({
location: harness.info.location,
data: live.map((id) => ({
id,
status: "running" as const,
command: `sleep ${id}`,
cwd: "/tmp/project",
shell: "sh",
file: "sh",
metadata: { sessionID: harness.info.id },
time: { started: 1 },
})),
});
harness.api.shell.remove = async ({ id }) => {
live = live.filter((entry) => entry !== id);
return { data: true };
};
const client = new OpenCodeV2AgentClient({
logger: createTestLogger(),
runtime: harness.runtime,
});
const session = await client.createSession({ provider: "opencode", cwd: "/tmp/project" });
const events: AgentStreamEvent[] = [];
session.subscribe((event) => events.push(event));
try {
await expect
.poll(() => events.findLast((event) => event.type === "background_work"))
.toMatchObject({ items: [{ id: "one" }, { id: "two" }] });
await session.stopBackgroundWork!("one");
expect(events.findLast((event) => event.type === "background_work")).toMatchObject({
items: [{ id: "two" }],
});
await expect(session.stopBackgroundWork!("one")).rejects.toThrow();
} finally {
await session.close();
}
});
test("closing during reconnect does not restart the background inventory", async () => {
const harness = new V2Harness();
const client = new OpenCodeV2AgentClient({
logger: createTestLogger(),
runtime: harness.runtime,
});
const session = await client.createSession({ provider: "opencode", cwd: "/tmp/project" });
const events: AgentStreamEvent[] = [];
session.subscribe((event) => events.push(event));
await expect
.poll(() => events.findLast((event) => event.type === "background_work"))
.toMatchObject({ items: [] });
let unblock!: () => void;
let entered = false;
const blocked = new Promise<void>((resolve) => {
unblock = resolve;
});
harness.connection.bindSession = async () => {
entered = true;
await blocked;
};
let reads = 0;
const original = harness.api.message.list;
harness.api.message.list = (...args) => {
reads++;
return original(...args);
};
harness.push({ id: "reconnect-close", created: 2, type: "server.connected", data: {} });
await expect.poll(() => entered).toBe(true);
const closing = session.close();
unblock();
await closing;
expect(reads).toBe(0);
});
test("keeps a shell available when its stop request times out", async () => {
const { vi } = await import("vitest");
vi.useFakeTimers();
try {
const { OpenCodeBackgroundInventory } = await import("./background-inventory.js");
const harness = new V2Harness();
harness.history.push({
id: "assistant",
type: "assistant",
agent: "build",
model: { providerID: "test", modelID: "test" },
time: { created: 1 },
content: [
{
type: "tool",
id: "tool",
name: "shell",
time: { created: 1 },
state: {
status: "completed",
input: {},
content: [{ type: "text", text: "" }],
metadata: { status: "running", shellID: "live" },
},
},
],
});
harness.api.shell.list = async () => ({
location: harness.info.location,
data: [
{
id: "live",
status: "running",
command: "sleep 120",
cwd: "/tmp/project",
shell: "sh",
file: "sh",
metadata: { sessionID: harness.info.id },
time: { started: 1 },
},
],
});
harness.api.shell.remove = () => new Promise(() => {});
const inventory = new OpenCodeBackgroundInventory({
id: harness.info.id,
cwd: "/tmp/project",
emit: () => {},
report: () => {},
});
inventory.start(harness.api);
await vi.advanceTimersByTimeAsync(0);
expect(inventory.snapshot()?.items).toMatchObject([{ id: "live", canStop: true }]);
const result = expect(inventory.stopItem("live")).rejects.toThrow("timed out");
await vi.advanceTimersByTimeAsync(10000);
await result;
expect(inventory.snapshot()?.items).toMatchObject([{ id: "live" }]);
inventory.disconnect();
} finally {
vi.useRealTimers();
}
});
test("keeps a live background shell visible and stoppable after its launch is rewound", async () => {
const harness = new V2Harness();
harness.history.push({
id: "assistant",
type: "assistant",
agent: "build",
model: { providerID: "test", modelID: "test" },
time: { created: 1 },
content: [
{
type: "tool",
id: "tool",
name: "shell",
time: { created: 1 },
state: {
status: "completed",
input: {},
content: [{ type: "text", text: "" }],
metadata: { status: "running", shellID: "live" },
},
},
],
});
let live = true;
harness.api.shell.list = async () => ({
location: harness.info.location,
data: live
? [
{
id: "live",
status: "running",
command: "sleep 180",
cwd: "/tmp/project",
shell: "sh",
file: "sh",
metadata: { sessionID: harness.info.id },
time: { started: 1 },
},
]
: [],
});
harness.api.shell.remove = async () => {
live = false;
return { data: true };
};
const client = new OpenCodeV2AgentClient({
logger: createTestLogger(),
runtime: harness.runtime,
});
const session = await client.createSession({ provider: "opencode", cwd: "/tmp/project" });
const events: AgentStreamEvent[] = [];
session.subscribe((event) => events.push(event));
try {
await expect
.poll(() => events.findLast((event) => event.type === "background_work"))
.toMatchObject({ items: [{ id: "live", canStop: true }] });
harness.history.length = 0;
harness.push({
id: "revert",
created: 2,
type: "session.revert.committed",
durable: { aggregateID: harness.info.id, seq: 1, version: 1 },
data: { sessionID: harness.info.id, to: "earlier" },
});
await expect
.poll(() => events.filter((event) => event.type === "background_work").length)
.toBeGreaterThanOrEqual(3);
expect(events.findLast((event) => event.type === "background_work")).toMatchObject({
items: [{ id: "live", canStop: true }],
});
harness.push({ id: "reconnect-after-rewind", created: 3, type: "server.connected", data: {} });
await expect
.poll(() => events.filter((event) => event.type === "background_work").length)
.toBeGreaterThanOrEqual(5);
expect(events.findLast((event) => event.type === "background_work")).toMatchObject({
items: [{ id: "live", canStop: true }],
});
await session.stopBackgroundWork!("live");
expect(events.findLast((event) => event.type === "background_work")).toMatchObject({
items: [],
});
} finally {
await session.close();
}
});
@@ -2,7 +2,7 @@ import type { OpenCodeClient } from "@opencode/client";
export interface V2Api {
server: OpenCodeClient["server"];
shell: Pick<OpenCodeClient["shell"], "list">;
shell: Pick<OpenCodeClient["shell"], "list" | "remove">;
session: Pick<
OpenCodeClient["session"],
| "create"
@@ -0,0 +1,308 @@
import type { OpenCodeEvent, SessionMessageInfo, ShellInfo } from "@opencode/client";
import type { AgentStreamEvent } from "../../../agent-sdk-types.js";
import type { V2Api } from "./api.js";
import { messages } from "./history.js";
type InventoryEvent = Extract<AgentStreamEvent, { type: "background_work" }>;
type RelevantEvent = Extract<
OpenCodeEvent,
{
type:
| "session.tool.input.started"
| "session.tool.input.ended"
| "session.tool.success"
| "shell.created"
| "shell.exited"
| "shell.deleted";
}
>;
export class OpenCodeBackgroundInventory {
private generation = 0;
private client: V2Api | null = null;
private loading = false;
private pending = false;
private dirty = false;
private retrying = false;
private events: RelevantEvent[] = [];
private names = new Map<string, string>();
private inputs = new Map<string, unknown>();
private descriptions = new Map<string, string>();
private promoted = new Set<string>();
private timer: ReturnType<typeof setTimeout> | null = null;
private current: InventoryEvent | null = null;
private readonly listeners = new Set<() => void>();
async stopItem(id: string): Promise<void> {
const client = this.client;
const generation = this.generation;
if (!client || !this.current?.items?.some((item) => item.id === id && item.canStop))
throw new Error("Background shell is no longer available");
let timer: ReturnType<typeof setTimeout> | undefined;
let disconnected!: () => void;
const lost = new Promise<never>((_, reject) => {
disconnected = () => {
if (generation !== this.generation)
reject(new Error("Background shell connection changed"));
};
this.listeners.add(disconnected);
timer = setTimeout(() => reject(new Error("Background shell stop timed out")), 10000);
});
try {
await Promise.race([client.shell.remove({ id, location: { directory: this.cwd } }), lost]);
} finally {
clearTimeout(timer);
this.listeners.delete(disconnected);
}
if (generation !== this.generation) throw new Error("Background shell connection changed");
await new Promise<void>((resolve, reject) => {
const confirmationTimer = setTimeout(
() => finish(new Error("Background shell confirmation timed out")),
15000,
);
const finish = (error?: Error) => {
clearTimeout(confirmationTimer);
this.listeners.delete(check);
if (error) reject(error);
else resolve();
};
const check = () => {
if (generation !== this.generation)
finish(new Error("Background shell connection changed"));
else if (this.current?.items && !this.current.items.some((item) => item.id === id))
finish();
};
this.listeners.add(check);
this.dirty = true;
this.refresh();
});
}
private notify() {
for (const listener of this.listeners) listener();
}
constructor(options: {
id: string;
cwd: string;
emit: (event: InventoryEvent) => void;
report: (error: unknown) => void;
}) {
this.id = options.id;
this.cwd = options.cwd;
this.emit = options.emit;
this.report = options.report;
}
private readonly id: string;
private readonly cwd: string;
private readonly emit: (event: InventoryEvent) => void;
private readonly report: (error: unknown) => void;
snapshot() {
return this.current;
}
start(client: V2Api) {
this.disconnect();
this.client = client;
this.loading = true;
this.events = [];
const generation = this.generation;
void this.bootstrap(client, generation);
}
disconnect(lost = false) {
this.generation++;
this.notify();
this.client = null;
this.loading = false;
this.pending = false;
this.dirty = false;
this.retrying = false;
this.events = [];
if (this.timer) clearTimeout(this.timer);
this.timer = null;
if (lost) {
this.promoted.clear();
this.descriptions.clear();
this.current = null;
this.emit({ type: "background_work", provider: "opencode", items: null });
} else if (this.current?.items?.some((item) => item.canStop)) {
// Keep stale rows visible, but revoke their stop tokens until this generation's
// roster confirms ownership again.
const items = this.current.items.map((item) => ({
id: item.id,
kind: item.kind,
description: item.description,
...(item.command ? { command: item.command } : {}),
}));
this.current = { type: "background_work", provider: "opencode", items };
this.emit(this.current);
}
}
private async bootstrap(client: V2Api, generation: number) {
try {
const history = await messages(client, this.id);
if (generation !== this.generation) return;
this.names = new Map();
this.inputs = new Map();
// Rewinds can remove launch messages without stopping the shell. Keep known
// promotions until the current provider roster confirms they are gone.
this.readHistory(history);
for (const event of this.events) this.apply(event);
this.events = [];
this.loading = false;
this.dirty = true;
this.refresh();
} catch (error) {
if (generation === this.generation) {
this.report(error);
this.schedule(() => void this.bootstrap(client, generation), true);
}
}
}
private readHistory(history: readonly SessionMessageInfo[]) {
for (const message of history) {
if (message.type !== "assistant") continue;
for (const part of message.content) {
if (part.type !== "tool") continue;
if (part.state.status === "completed") {
if (part.name === "shell") this.promote(part.state.metadata, part.state.input);
} else if (part.state.status !== "error") {
const key = `${message.id}:${part.id}`;
this.names.set(key, part.name);
if (part.name === "shell") this.inputs.set(key, part.state.input);
}
}
}
}
private promote(metadata: { [key: string]: unknown } | undefined, input?: unknown) {
if (metadata?.status === "running" && typeof metadata.shellID === "string") {
this.promoted.add(metadata.shellID);
if (
input &&
typeof input === "object" &&
"description" in input &&
typeof input.description === "string" &&
input.description.trim()
)
this.descriptions.set(metadata.shellID, input.description.trim());
}
}
observe(event: OpenCodeEvent) {
if (event.type === "session.revert.committed" && event.data.sessionID === this.id) {
if (this.client) this.start(this.client);
return;
}
if (
event.type !== "shell.created" &&
event.type !== "shell.exited" &&
event.type !== "shell.deleted" &&
event.type !== "session.tool.input.started" &&
event.type !== "session.tool.input.ended" &&
event.type !== "session.tool.success"
)
return;
if ("sessionID" in event.data && event.data.sessionID !== this.id) return;
if (!this.client) return;
if (this.loading) {
this.events.push(event);
return;
}
this.apply(event);
if (event.type !== "session.tool.input.started" && event.type !== "session.tool.input.ended") {
this.dirty = true;
if (!this.retrying) this.refresh();
}
}
private apply(event: RelevantEvent) {
if (event.type === "session.tool.input.started")
this.names.set(`${event.data.assistantMessageID}:${event.data.id}`, event.data.name);
if (
event.type === "session.tool.input.ended" &&
this.names.get(`${event.data.assistantMessageID}:${event.data.id}`) === "shell"
) {
try {
this.inputs.set(
`${event.data.assistantMessageID}:${event.data.id}`,
JSON.parse(event.data.text),
);
} catch {
/* A malformed tool input supplies no presentation metadata. */
}
}
if (
event.type === "session.tool.success" &&
this.names.get(`${event.data.assistantMessageID}:${event.data.id}`) === "shell"
)
this.promote(
event.data.metadata,
this.inputs.get(`${event.data.assistantMessageID}:${event.data.id}`),
);
if (event.type === "session.tool.success") {
this.names.delete(`${event.data.assistantMessageID}:${event.data.id}`);
this.inputs.delete(`${event.data.assistantMessageID}:${event.data.id}`);
}
}
private refresh() {
if (!this.client || this.loading || this.pending) return;
if (this.timer) clearTimeout(this.timer);
this.timer = null;
this.pending = true;
this.dirty = false;
const generation = this.generation;
const client = this.client;
let failed = false;
void client.shell
.list({ location: { directory: this.cwd } })
.then((result) => {
if (generation !== this.generation || this.dirty) return undefined;
this.publish(result.data);
return undefined;
})
.catch((error: unknown) => {
if (generation === this.generation) {
failed = true;
this.report(error);
}
})
.finally(() => {
if (generation !== this.generation) return;
this.pending = false;
if (this.dirty && !failed) this.refresh();
else if (failed || this.current?.items?.length) this.schedule(() => this.refresh(), failed);
});
}
private schedule(callback: () => void, retry = false) {
this.retrying = retry;
this.timer = setTimeout(() => {
this.timer = null;
this.retrying = false;
callback();
}, 5000);
this.timer.unref();
}
private publish(shells: ShellInfo[]) {
const live = new Set(
shells
.filter((shell) => shell.status === "running" && shell.metadata.sessionID === this.id)
.map((shell) => shell.id),
);
this.promoted = new Set([...this.promoted].filter((id) => live.has(id)));
for (const id of this.descriptions.keys()) if (!live.has(id)) this.descriptions.delete(id);
this.current = {
type: "background_work",
provider: "opencode",
items: shells
.filter(
(shell) =>
shell.status === "running" &&
shell.metadata.sessionID === this.id &&
this.promoted.has(shell.id),
)
.map((shell) => ({
id: shell.id,
kind: "shell",
description: this.descriptions.get(shell.id) ?? null,
command: shell.command || undefined,
canStop: true,
})),
};
this.emit(this.current);
this.notify();
}
}
@@ -1,3 +1,4 @@
import { OpenCodeBackgroundInventory } from "./background-inventory.js";
import { SessionChildren } from "./children.js";
import { SessionTurns } from "./turns.js";
import { V2Timeline } from "./timeline.js";
@@ -57,6 +58,7 @@ export class OpenCodeV2Session implements AgentSession {
private refreshTimer: ReturnType<typeof setTimeout> | null = null;
private closed = false;
private history: SessionMessageInfo[] = [];
private readonly background: OpenCodeBackgroundInventory;
private modes: AgentMode[] = [];
constructor(
private connection: V2Connection,
@@ -70,6 +72,16 @@ export class OpenCodeV2Session implements AgentSession {
private readonly acquire?: () => Promise<V2Connection>,
private readonly moved?: (connection: V2Connection) => void,
) {
this.background = new OpenCodeBackgroundInventory({
id: info.id,
cwd: config.cwd,
emit: (event) => this.emit(event),
report: (error) =>
this.logger.warn(
{ error: toDiagnosticErrorMessage(error) },
"OpenCode background inventory failed",
),
});
this.permissions = new SessionPermissions(
() => this.client,
info.id,
@@ -143,6 +155,7 @@ export class OpenCodeV2Session implements AgentSession {
void connection.exited.then((error) => {
if (this.closed || this.connection !== connection) return undefined;
this.exited = true;
this.background.disconnect(true);
this.streamAbort.abort(error);
this.turns.fail(error);
return undefined;
@@ -242,6 +255,8 @@ export class OpenCodeV2Session implements AgentSession {
}
subscribe(callback: (event: AgentStreamEvent) => void) {
this.listeners.add(callback);
const background = this.background.snapshot();
if (background) callback(background);
// Reopening can adopt an execution during initialization, before its caller
// can subscribe. Deliver that live turn so it remains visible and interruptible.
if (this.turns.id)
@@ -406,6 +421,9 @@ export class OpenCodeV2Session implements AgentSession {
steerActiveTurn(prompt: AgentPromptInput, options: SteerActiveTurnOptions): Promise<SteerResult> {
return this.turns.steerActiveTurn(prompt, options);
}
stopBackgroundWork(id: string) {
return this.background.stopItem(id);
}
interrupt() {
return this.turns.interrupt();
}
@@ -515,9 +533,12 @@ export class OpenCodeV2Session implements AgentSession {
if (event.type === "server.connected") {
this.timeline.resetStreams();
await this.reconcileConnection();
signal.throwIfAborted();
this.background.start(this.client);
connected = true;
ready();
}
this.background.observe(event);
await this.children.observe(event);
this.observeOwnEvent(event);
}
@@ -533,6 +554,7 @@ export class OpenCodeV2Session implements AgentSession {
"OpenCode event stream interrupted; reconciling on reconnect",
);
}
this.background.disconnect();
await delay(100, undefined, { signal }).catch(() => undefined);
}
}
@@ -573,6 +595,7 @@ export class OpenCodeV2Session implements AgentSession {
this.turns.close();
this.streamAbort.abort();
if (this.refreshTimer) clearTimeout(this.refreshTimer);
this.background.disconnect();
await this.stream;
await this.reconnecting?.catch(() => undefined);
await this.sync.catch(() => undefined);
@@ -19,6 +19,8 @@ const INBOUND_PERMISSION = {
"agent.detach.request": "workspace.write",
"agent.fork_context.request": "workspace.read",
"agent.provider_subagents.list.request": "workspace.read",
"agent.background_work.list.request": "workspace.read",
"agent.background_work.stop.request": ["workspace.write", "hub.execute"],
"agent.provider_subagents.timeline.get.request": "workspace.read",
"agent.rewind.request": "workspace.write",
"agent.skills.get_status.request": "daemon.read",
@@ -251,6 +253,9 @@ const OUTBOUND_PERMISSION = {
"agent.provider_subagents.list.response": "workspace.read",
"agent.provider_subagents.timeline.get.response": "workspace.read",
"agent.provider_subagents.update": "workspace.read",
"agent.background_work.list.response": "workspace.read",
"agent.background_work.stop.response": ["workspace.write", "hub.execute"],
"agent.background_work.update": "workspace.read",
"agent.rewind.response": "workspace.write",
"agent.skills.get_status.response": "daemon.read",
"agent.skills.import_legacy_selection.response": "daemon.manage",
+249
View File
@@ -411,6 +411,7 @@ function createSessionForTest(options: SessionForTestOptions = {}): Session {
agentManager: asAgentManager({
listAgents: vi.fn(() => []),
listProviderSubagentActivity: vi.fn(() => []),
getAgentBackgroundWorkSummary: () => undefined,
subscribe: vi.fn(() => () => {}),
...options.agentManager,
}),
@@ -6105,3 +6106,251 @@ test("a failing direct source observes fire-and-forget file replies without unha
rmSync(root, { recursive: true, force: true });
}
});
test("answers an agent background work list request", async () => {
const messages: SessionOutboundMessage[] = [];
const session = createSessionForTest({
messages,
agentManager: {
waitForAgentClose: vi.fn().mockResolvedValue(undefined),
getAgent: vi.fn(() => ({ id: "agent-a" })),
getAgentBackgroundWorkSummary: () => ({ count: 1 }),
listAgentBackgroundWork: vi.fn(() => [
{
id: "bash-1",
kind: "shell",
description: "sleep 20",
startedAt: "2026-10-06T10:00:00.000Z",
},
]),
},
});
await session.handleMessage({
type: "agent.background_work.list.request",
agentId: "agent-a",
requestId: "list-1",
});
expect(messages).toContainEqual({
type: "agent.background_work.list.response",
payload: {
requestId: "list-1",
agentId: "agent-a",
items: [
{
id: "bash-1",
kind: "shell",
description: "sleep 20",
startedAt: "2026-10-06T10:00:00.000Z",
},
],
error: null,
},
});
});
test("reports an unavailable inventory for a stored agent without loading it", async () => {
const messages: SessionOutboundMessage[] = [];
const resumeAgentFromPersistence = vi.fn();
const createAgent = vi.fn();
const listAgentBackgroundWork = vi.fn(() => []);
const session = createSessionForTest({
messages,
agentManager: {
waitForAgentClose: vi.fn().mockResolvedValue(undefined),
getAgent: () => undefined,
getRegisteredProviderIds: vi.fn(() => ["claude"]),
resumeAgentFromPersistence,
createAgent,
listAgentBackgroundWork,
},
agentStorage: {
get: vi.fn().mockResolvedValue(
createStoredAgentRecord({
id: "agent-a",
cwd: "/tmp/repo",
provider: "claude",
persistence: { provider: "claude", sessionId: "session-a" },
}),
),
},
});
await session.handleMessage({
type: "agent.background_work.list.request",
agentId: "agent-a",
requestId: "list-stored",
});
expect(messages).toContainEqual({
type: "agent.background_work.list.response",
payload: {
requestId: "list-stored",
agentId: "agent-a",
items: [],
error: "Background work inventory is unavailable",
},
});
expect(resumeAgentFromPersistence).not.toHaveBeenCalled();
expect(createAgent).not.toHaveBeenCalled();
expect(listAgentBackgroundWork).not.toHaveBeenCalled();
});
test.each([
{
name: "an archived agent",
record: createStoredAgentRecord({
id: "agent-a",
cwd: "/tmp/repo",
archivedAt: "2026-10-06T10:00:00.000Z",
}),
error: "Agent is archived: agent-a",
},
{ name: "an unknown agent", record: undefined, error: "Unknown agent 'agent-a'" },
{
name: "an internal agent",
record: createStoredAgentRecord({ id: "agent-a", cwd: "/tmp/repo", internal: true }),
error: "Unknown agent 'agent-a'",
},
])("rejects a background work list request for $name that is not loaded", async (scenario) => {
const messages: SessionOutboundMessage[] = [];
const session = createSessionForTest({
messages,
agentManager: {
waitForAgentClose: vi.fn().mockResolvedValue(undefined),
getAgent: () => undefined,
getRegisteredProviderIds: vi.fn(() => ["codex"]),
resumeAgentFromPersistence: vi.fn(),
createAgent: vi.fn(),
},
agentStorage: { get: vi.fn().mockResolvedValue(scenario.record) },
});
await session.handleMessage({
type: "agent.background_work.list.request",
agentId: "agent-a",
requestId: "list-rejected",
});
expect(messages).toContainEqual({
type: "agent.background_work.list.response",
payload: { requestId: "list-rejected", agentId: "agent-a", items: [], error: scenario.error },
});
});
test("never pushes background work to an old app", async () => {
const messages: SessionOutboundMessage[] = [];
const targetedMessages: Array<{ source: object; message: SessionOutboundMessage }> = [];
const agentEventListeners: Array<(event: AgentManagerEvent) => void> = [];
const session = createSessionForTest({
messages,
targetedMessages,
agentManager: {
getAgent: vi.fn(() => undefined),
subscribe: vi.fn((listener: (event: AgentManagerEvent) => void) => {
agentEventListeners.push(listener);
return () => {};
}),
},
});
// provider_subagents makes subagent updates flow to this legacy socket, so the
// manager listener is installed for it.
const legacySocket = {};
session.updateClientCapabilities({ provider_subagents: true }, legacySocket);
const listener = agentEventListeners[0];
if (!listener) throw new Error("Agent event listener was not installed");
listener({
type: "background_work",
agentId: "agent-a",
items: [
{ id: "bash-1", kind: "shell", description: null, startedAt: "2026-10-06T10:00:00.000Z" },
],
});
const pushed = [...messages, ...targetedMessages.map((entry) => entry.message)];
expect(pushed.some((message) => message.type === "agent.background_work.update")).toBe(false);
});
test("pushes background work only to clients that subscribed to it", async () => {
const messages: SessionOutboundMessage[] = [];
const agentEventListeners: Array<(event: AgentManagerEvent) => void> = [];
const session = createSessionForTest({
messages,
agentManager: {
getAgent: vi.fn(() => undefined),
subscribe: vi.fn((listener: (event: AgentManagerEvent) => void) => {
agentEventListeners.push(listener);
return () => {};
}),
},
});
const forward = (event: AgentManagerEvent) => {
for (const listener of agentEventListeners) listener(event);
};
const event: AgentManagerEvent = {
type: "background_work",
agentId: "agent-a",
items: [
{ id: "bash-1", kind: "shell", description: null, startedAt: "2026-10-06T10:00:00.000Z" },
],
};
// Subscribe to an unrelated event so the manager listener is installed, then prove
// a client that did not ask for background work never receives it.
await session.handleMessage({
type: "session.events.set_subscription.request",
requestId: "events-1",
events: ["agent.provider_subagents.update"],
});
if (agentEventListeners.length === 0) throw new Error("Agent event listener was not installed");
forward(event);
expect(messages.some((message) => message.type === "agent.background_work.update")).toBe(false);
await session.handleMessage({
type: "session.events.set_subscription.request",
requestId: "events-2",
events: ["agent.background_work.update"],
});
messages.length = 0;
forward(event);
expect(messages).toContainEqual(
expect.objectContaining({
type: "agent.background_work.update",
payload: expect.objectContaining({ agentId: "agent-a", items: event.items }),
}),
);
});
test("does not publish internal agents' background work to subscribed clients", async () => {
const messages: SessionOutboundMessage[] = [];
let listener: ((event: AgentManagerEvent) => void) | undefined;
const session = createSessionForTest({
messages,
agentManager: {
getAgent: () => ({ id: "internal-agent", internal: true }) as never,
subscribe: (callback) => {
listener = callback;
return () => {};
},
},
});
await session.handleMessage({
type: "session.events.set_subscription.request",
requestId: "subscribe-internal-background",
events: ["agent.background_work.update"],
});
messages.length = 0;
if (!listener) throw new Error("Agent event listener was not installed");
listener({
type: "background_work",
agentId: "internal-agent",
items: [
{ id: "work", kind: "shell", description: "private", startedAt: "2026-10-06T10:00:00.000Z" },
],
});
expect(messages.some((message) => message.type === "agent.background_work.update")).toBe(false);
});
+108 -9
View File
@@ -22,6 +22,7 @@ import { isAbsolute } from "node:path";
import { CreationService, type CreatedAgent } from "./creation/index.js";
import type { CreationSnapshot, AgentCreateRequest } from "@getpaseo/protocol/messages";
import type { MessageReceipts } from "./message-receipts/index.js";
import type { AgentBackgroundWorkItem } from "./agent/background-work/index.js";
import equal from "fast-deep-equal";
import { WorkspaceSetupObservation } from "./workspace-setup-observation.js";
import { SessionDelivery, type OwnedSubscription } from "./session/owned-subscriptions/index.js";
@@ -1771,7 +1772,8 @@ export class Session {
this.wantsEvent("agent_attention_required") ||
this.wantsEvent("agent_permission_request") ||
this.wantsEvent("agent_permission_resolved") ||
this.wantsEvent("agent.provider_subagents.update");
this.wantsEvent("agent.provider_subagents.update") ||
this.wantsEvent("agent.background_work.update");
if (agents && !this.unsubscribeAgentEvents) this.subscribeToAgentEvents();
if (!agents) {
this.unsubscribeAgentEvents?.();
@@ -2089,14 +2091,12 @@ export class Session {
}
if (event.type === "provider_subagent") {
// The subagents track of an internal parent stays hidden.
const parentAgentId =
event.event.type === "upsert"
? event.event.subagent.parentAgentId
: event.event.parentAgentId;
if (this.agentManager.getAgent(parentAgentId)?.internal) return;
this.emitProviderSubagentWorkspaceUpdate(event.event);
this.forwardProviderSubagentUpdate(event.event);
this.forwardProviderSubagent(event.event);
return;
}
if (event.type === "background_work") {
this.forwardBackgroundWork(event);
return;
}
@@ -2171,6 +2171,26 @@ export class Session {
);
}
private forwardProviderSubagent(event: ProviderSubagentManagerEvent): void {
const parentAgentId =
event.type === "upsert" ? event.subagent.parentAgentId : event.parentAgentId;
if (this.agentManager.getAgent(parentAgentId)?.internal) return;
this.emitProviderSubagentWorkspaceUpdate(event);
this.forwardProviderSubagentUpdate(event);
}
private forwardBackgroundWork(
event: Extract<AgentManagerEvent, { type: "background_work" }>,
): void {
const agent = this.agentManager.getAgent(event.agentId);
if (agent?.internal) return;
this.emit({
type: "agent.background_work.update",
payload: { agentId: event.agentId, items: event.items },
});
if (agent) void this.agentUpdates.forwardLiveAgent(agent);
}
private emitProviderSubagentWorkspaceUpdate(event: ProviderSubagentManagerEvent): void {
if (event.type === "timeline") {
return;
@@ -2203,6 +2223,7 @@ export class Session {
}
private async enrichAgentPayload(payload: AgentSnapshotPayload): Promise<AgentSnapshotPayload> {
payload.backgroundWork = this.agentManager.getAgentBackgroundWorkSummary(payload.id);
const storedRecord = await this.agentStorage.get(payload.id);
payload.title = storedRecord?.title ?? null;
payload.archivedAt = storedRecord?.archivedAt ?? null;
@@ -2967,6 +2988,10 @@ export class Session {
return this.handleAgentTimelineListPromptsRequest(msg, source);
case "agent.provider_subagents.list.request":
return this.handleProviderSubagentListRequest(msg);
case "agent.background_work.list.request":
return this.handleAgentBackgroundWorkListRequest(msg);
case "agent.background_work.stop.request":
return this.handleAgentBackgroundWorkStopRequest(msg);
case "agent.provider_subagents.timeline.get.request":
return this.handleProviderSubagentTimelineRequest(msg, source);
case "session.events.set_subscription.request": {
@@ -5858,6 +5883,7 @@ export class Session {
pendingPermissionCount: agent.pendingPermissions?.length ?? 0,
requiresAttention: agent.requiresAttention,
attentionReason: agent.attentionReason ?? null,
backgroundWork: agent.backgroundWork,
});
case "created_at":
return Date.parse(agent.createdAt);
@@ -8468,6 +8494,75 @@ export class Session {
}
}
private async handleAgentBackgroundWorkStopRequest(
msg: Extract<SessionInboundMessage, { type: "agent.background_work.stop.request" }>,
): Promise<void> {
let error: string | null = null;
try {
await this.agentManager.stopAgentBackgroundWork({
agentId: msg.agentId,
itemId: msg.itemId,
stopToken: msg.stopToken,
});
} catch (cause) {
error = cause instanceof Error ? cause.message : String(cause);
}
this.emit({
type: "agent.background_work.stop.response",
payload: {
requestId: msg.requestId,
agentId: msg.agentId,
itemId: msg.itemId,
error,
},
});
}
private async handleAgentBackgroundWorkListRequest(
msg: Extract<SessionInboundMessage, { type: "agent.background_work.list.request" }>,
): Promise<void> {
try {
this.emit({
type: "agent.background_work.list.response",
payload: {
requestId: msg.requestId,
agentId: msg.agentId,
items: await this.readAgentBackgroundWork(msg.agentId),
error: null,
},
});
} catch (error) {
this.emit({
type: "agent.background_work.list.response",
payload: {
requestId: msg.requestId,
agentId: msg.agentId,
items: [],
error: error instanceof Error ? error.message : String(error),
},
});
}
}
// Background work lives only in memory, so an agent that is not loaded has none.
// Answering from storage keeps a read from resuming a dormant agent.
private async readAgentBackgroundWork(agentId: string): Promise<AgentBackgroundWorkItem[]> {
if (this.agentManager.getAgent(agentId)) {
if (!this.agentManager.getAgentBackgroundWorkSummary(agentId)) {
throw new Error("Background work inventory is unavailable");
}
return this.agentManager.listAgentBackgroundWork(agentId);
}
const record = await this.agentStorage.get(agentId);
if (!record || record.internal) {
throw new Error(`Unknown agent '${agentId}'`);
}
if (record.archivedAt) {
throw new Error(`Agent is archived: ${agentId}`);
}
throw new Error("Background work inventory is unavailable");
}
private async handleProviderSubagentTimelineRequest(
msg: Extract<SessionInboundMessage, { type: "agent.provider_subagents.timeline.get.request" }>,
source?: object,
@@ -9189,6 +9284,7 @@ function sessionEventCategory(message: SessionOutboundMessage): SessionEventSubs
case "script_status_update":
case "workspace_setup_progress":
case "agent.provider_subagents.update":
case "agent.background_work.update":
case "terminal_attention_required":
case "activity_log":
case "hub.execution.agent.update":
@@ -9227,6 +9323,9 @@ function legacyWantsEvent(
return !capabilities.has(CLIENT_CAPS.explicitEventSubscriptions);
case "agent.provider_subagents.update":
return capabilities.has(CLIENT_CAPS.providerSubagents);
// Old apps parse outbound messages strictly and do not know this type.
case "agent.background_work.update":
return false;
default:
return true;
}
@@ -37,6 +37,7 @@ export function asAgentManager(stub: {
return createStub<SessionOptions["agentManager"]>({
listProviderSubagentActivity: () => [],
getRetiredInternalAgent: () => null,
getAgentBackgroundWorkSummary: () => undefined,
...stub,
});
}
@@ -2149,6 +2149,10 @@ export class VoiceAssistantWebSocketServer {
agentForkContextCursor: true,
// COMPAT(providerSubagents): added in v0.1.107, remove gate after 2027-01-12.
providerSubagents: true,
// COMPAT(agentBackgroundWork): added after v0.11.0-beta.5, remove gate after 2027-04-06.
agentBackgroundWork: true,
// COMPAT(agentBackgroundWorkStop): added after v0.15.0, remove after 2027-05-03.
agentBackgroundWorkStop: true,
// COMPAT(projectedSubagentTimeline): added after v0.8.0, remove gates after 2027-03-14; retain advertisement.
projectedSubagentTimeline: true,
wholeTurnTimeline: true,
@@ -114,6 +114,10 @@ class InMemoryAgentManager {
};
}
getAgentBackgroundWorkSummary() {
return undefined;
}
fetchTimeline(_agentId: string, options?: AgentTimelineFetchOptions) {
return this.timeline.fetch("agent-1", options);
}
@@ -273,6 +273,7 @@ interface AgentState {
pendingPermissionCount?: number;
requiresAttention?: boolean;
attentionReason?: AgentSnapshotPayload["attentionReason"];
backgroundWork?: AgentSnapshotPayload["backgroundWork"];
background?: boolean;
}
@@ -293,6 +294,7 @@ function createAgent(
updatedAt: NOW,
lastUserMessageAt: input.lastUserMessageAt ?? null,
status: input.status,
backgroundWork: input.backgroundWork,
capabilities: {
supportsStreaming: true,
supportsSessionPersistence: true,
@@ -324,6 +326,17 @@ function createAgent(
}
describe("WorkspaceDirectory", () => {
test("treats background work as running without changing foreground status", async () => {
const workspace = new WorkspaceStatus();
workspace.hasRootAgent({
id: "root-agent",
status: "idle",
backgroundWork: { count: 1 },
requiresAttention: true,
attentionReason: "finished",
});
await expect(workspace.workspaceStatus()).resolves.toBe("running");
});
test("uses root agent activity, not delegated child activity, for workspace status", async () => {
const workspace = new WorkspaceStatus();
@@ -392,7 +392,7 @@ export class WorkspaceDirectory {
continue;
}
const isWorkspaceRoot = workspaceAgent.id === agent.id;
if (!isWorkspaceRoot && agent.status !== "running") {
if (!isWorkspaceRoot && agent.status !== "running" && !agent.backgroundWork?.count) {
continue;
}
const bucket = isWorkspaceRoot
@@ -401,6 +401,7 @@ export class WorkspaceDirectory {
pendingPermissionCount: agent.pendingPermissions?.length ?? 0,
requiresAttention: agent.requiresAttention,
attentionReason: agent.attentionReason ?? null,
backgroundWork: agent.backgroundWork,
})
: "running";
@@ -550,6 +551,7 @@ export class WorkspaceDirectory {
pendingPermissionCount: agent.pendingPermissions?.length ?? 0,
requiresAttention: agent.requiresAttention,
attentionReason: agent.attentionReason ?? null,
backgroundWork: agent.backgroundWork,
});
return derived === winningBucket;
})
@@ -93,6 +93,10 @@ function createLogger(): pino.Logger {
} as unknown as pino.Logger;
}
function decodedTerminalFrames(frames: TerminalStreamFrame[]): string {
return frames.map((frame) => new TextDecoder().decode(frame.payload)).join("");
}
describe("terminal-session-controller restore", () => {
test("delivers output produced while restore is in flight after the restore frame", async () => {
let terminalListener: ((message: ServerMessage) => void) | null = null;
@@ -172,7 +176,10 @@ describe("terminal-session-controller restore", () => {
terminalListener?.({ type: "output", data: "restore-after\n", revision: 2 });
snapshot.resolve(terminalReplay("restore-before", 1, "\x1b[?1h\x1b[?2004h"));
await snapshot.promise;
await new Promise((resolve) => setTimeout(resolve, 20));
await vi.waitFor(() => {
const output = decodedTerminalFrames(binaryFrames.slice(1));
expect(output).toBe("\x1b[?1h\x1b[?2004hrestore-after\n");
});
expect(outboundMessages).toContainEqual({
type: "subscribe_terminal_response",
@@ -183,14 +190,12 @@ describe("terminal-session-controller restore", () => {
requestId: "req-1",
},
});
expect(binaryFrames.map((frame) => frame.opcode)).toEqual([
TerminalStreamOpcode.Restore,
TerminalStreamOpcode.Output,
]);
expect(binaryFrames[0]?.opcode).toBe(TerminalStreamOpcode.Restore);
const outputFrames = binaryFrames.slice(1);
expect(outputFrames.length).toBeGreaterThan(0);
expect(outputFrames.every((frame) => frame.opcode === TerminalStreamOpcode.Output)).toBe(true);
expect(new TextDecoder().decode(binaryFrames[0]?.payload)).toContain("restore-before");
expect(new TextDecoder().decode(binaryFrames[1]?.payload)).toBe(
"\x1b[?1h\x1b[?2004hrestore-after\n",
);
expect(decodedTerminalFrames(outputFrames)).toBe("\x1b[?1h\x1b[?2004hrestore-after\n");
});
});