feat(pi): improve subagent monitoring #41
@@ -106,7 +106,7 @@ export default function subagents(pi: ExtensionAPI) {
|
||||
pi.registerTool({
|
||||
name: "subagent_list",
|
||||
label: "List subagents",
|
||||
description: "List active and recent subagents for this parent session",
|
||||
description: "List active and terminal subagents for this parent session until terminal entries are cleared",
|
||||
parameters: Type.Object({}),
|
||||
async execute(_toolCallId, _params, _signal, _onUpdate, ctx) {
|
||||
return textResult(getSupervisor(ctx).list());
|
||||
@@ -167,6 +167,20 @@ export default function subagents(pi: ExtensionAPI) {
|
||||
},
|
||||
});
|
||||
|
||||
pi.registerTool({
|
||||
name: "subagent_clear",
|
||||
label: "Clear terminal subagents",
|
||||
description: "Remove terminal subagents from the current-session visible work set. Omitting ids clears all terminal children",
|
||||
parameters: Type.Object({
|
||||
ids: Type.Optional(Type.Array(Type.String({ description: "Subagent id returned by subagent_spawn or subagent_batch" }))),
|
||||
}),
|
||||
async execute(_toolCallId, params, _signal, _onUpdate, ctx) {
|
||||
const input = params as { ids?: unknown };
|
||||
const ids = Array.isArray(input.ids) ? input.ids.map(String) : undefined;
|
||||
return textResult({ cleared: getSupervisor(ctx).clearTerminal(ids) });
|
||||
},
|
||||
});
|
||||
|
||||
pi.registerCommand("subagent-spawn", {
|
||||
description: "Start an ad hoc independent subagent",
|
||||
handler: async (args, ctx) => {
|
||||
@@ -194,6 +208,14 @@ export default function subagents(pi: ExtensionAPI) {
|
||||
},
|
||||
});
|
||||
|
||||
pi.registerCommand("subagent-clear", {
|
||||
description: "Clear terminal subagent records. Pass ids to clear selected terminal records only",
|
||||
handler: async (args, ctx) => {
|
||||
const ids = args.trim().split(/\s+/u).filter(Boolean);
|
||||
ctx.ui.notify(JSON.stringify({ cleared: getSupervisor(ctx).clearTerminal(ids.length > 0 ? ids : undefined) }, null, 2), "info");
|
||||
},
|
||||
});
|
||||
|
||||
pi.registerCommand("subagent-status", {
|
||||
description: "Show a subagent status by id",
|
||||
handler: async (args, ctx) => {
|
||||
|
||||
@@ -236,58 +236,42 @@ test("maxConcurrent preserves queued records", async () => {
|
||||
assert.equal(runner.starts.length, 2);
|
||||
});
|
||||
|
||||
test("recent terminal statuses expire from list by ttl", async () => {
|
||||
test("terminal records stay listed past ttl and remain retrievable until cleared", async () => {
|
||||
const runner = new FakeRunner();
|
||||
const supervisor = new Supervisor(runner, "/tmp", { recentTerminalTtlMs: 5 });
|
||||
const accepted = await spawnStarted(supervisor);
|
||||
|
||||
runner.starts[0].events.completed("done", "agent_settled");
|
||||
assert.equal(supervisor.list().some((status) => status.id === accepted.id), true);
|
||||
const completed = await spawnStarted(supervisor, "one");
|
||||
const failed = await spawnStarted(supervisor, "two");
|
||||
|
||||
runner.starts[0].events.completed("one done", "agent_settled");
|
||||
runner.starts[1].events.failed("two failed");
|
||||
await sleep(10);
|
||||
|
||||
assert.equal(supervisor.list().some((status) => status.id === accepted.id), false);
|
||||
assert.equal(supervisor.result(accepted.id).result, "done");
|
||||
const listedIds = supervisor.list().map((status) => status.id);
|
||||
assert.ok(listedIds.includes(completed.id));
|
||||
assert.ok(listedIds.includes(failed.id));
|
||||
assert.equal(supervisor.result(completed.id).result, "one done");
|
||||
assert.equal(supervisor.result(failed.id).error, "two failed");
|
||||
|
||||
(supervisor as Supervisor & { clearTerminal(): void }).clearTerminal();
|
||||
|
||||
const afterClearIds = supervisor.list().map((status) => status.id);
|
||||
assert.equal(afterClearIds.includes(completed.id), false);
|
||||
assert.equal(afterClearIds.includes(failed.id), false);
|
||||
assert.throws(() => supervisor.status(completed.id), /unknown subagent id/);
|
||||
assert.throws(() => supervisor.result(failed.id), /unknown subagent id/);
|
||||
});
|
||||
|
||||
test("recent terminal ttl does not hide active statuses", async () => {
|
||||
test("zero recent terminal ttl does not hide terminal statuses", async () => {
|
||||
const runner = new FakeRunner();
|
||||
const supervisor = new Supervisor(runner, "/tmp", { recentTerminalTtlMs: 0 });
|
||||
const accepted = await spawnStarted(supervisor);
|
||||
|
||||
runner.starts[0].events.completed("done", "agent_settled");
|
||||
|
||||
assert.equal(supervisor.list().some((status) => status.id === accepted.id), true);
|
||||
});
|
||||
|
||||
test("zero recent terminal ttl hides terminal statuses immediately", async () => {
|
||||
const runner = new FakeRunner();
|
||||
const supervisor = new Supervisor(runner, "/tmp", { recentTerminalTtlMs: 0 });
|
||||
const accepted = await spawnStarted(supervisor);
|
||||
|
||||
runner.starts[0].events.completed("done", "agent_settled");
|
||||
|
||||
assert.equal(supervisor.list().some((status) => status.id === accepted.id), false);
|
||||
assert.equal(supervisor.result(accepted.id).result, "done");
|
||||
});
|
||||
|
||||
test("recent terminal ttl emits a change when an entry expires", async () => {
|
||||
const runner = new FakeRunner();
|
||||
let changes = 0;
|
||||
const supervisor = new Supervisor(runner, "/tmp", {
|
||||
recentTerminalTtlMs: 5,
|
||||
onChange: () => {
|
||||
changes += 1;
|
||||
},
|
||||
});
|
||||
await spawnStarted(supervisor);
|
||||
const beforeComplete = changes;
|
||||
|
||||
runner.starts[0].events.completed("done", "agent_settled");
|
||||
await sleep(15);
|
||||
|
||||
assert.ok(changes > beforeComplete + 1);
|
||||
assert.equal(supervisor.list().length, 0);
|
||||
});
|
||||
|
||||
test("wait blocks until multiple subagents are terminal", async () => {
|
||||
const runner = new FakeRunner();
|
||||
const supervisor = new Supervisor(runner, "/tmp");
|
||||
|
||||
@@ -48,7 +48,6 @@ export class Supervisor {
|
||||
private readonly children = new Map<string, RunningChild>();
|
||||
private readonly queue: RunningChild[] = [];
|
||||
private readonly waiters = new Set<() => void>();
|
||||
private recentTerminalTimer?: ReturnType<typeof setTimeout>;
|
||||
|
||||
constructor(
|
||||
private readonly runner: ChildRunner,
|
||||
@@ -78,9 +77,7 @@ export class Supervisor {
|
||||
const active = statuses.filter((status) => !isTerminal(status.state));
|
||||
const terminal = statuses
|
||||
.filter((status) => isTerminal(status.state))
|
||||
.filter((status) => this.isRecentTerminal(status))
|
||||
.sort((a, b) => Date.parse(b.completedAt ?? b.startedAt) - Date.parse(a.completedAt ?? a.startedAt))
|
||||
.slice(0, this.options.recentTerminalLimit ?? 10);
|
||||
.sort((a, b) => Date.parse(b.completedAt ?? b.startedAt) - Date.parse(a.completedAt ?? a.startedAt));
|
||||
return [...active, ...terminal];
|
||||
}
|
||||
|
||||
@@ -92,6 +89,20 @@ export class Supervisor {
|
||||
return cloneResult(this.require(id).record);
|
||||
}
|
||||
|
||||
clearTerminal(ids?: string[]): SubagentStatus[] {
|
||||
const selectedIds = ids ? [...new Set(ids.map((id) => id.trim()).filter(Boolean))] : undefined;
|
||||
if (selectedIds) for (const id of selectedIds) this.require(id);
|
||||
const cleared: SubagentStatus[] = [];
|
||||
for (const [id, child] of this.children) {
|
||||
if (selectedIds && !selectedIds.includes(id)) continue;
|
||||
if (!isTerminal(child.record.status.state)) continue;
|
||||
cleared.push(cloneStatus(child.record.status));
|
||||
this.children.delete(id);
|
||||
}
|
||||
if (cleared.length > 0) this.emitChange();
|
||||
return cleared;
|
||||
}
|
||||
|
||||
async wait(
|
||||
ids: string[],
|
||||
options: { timeoutMs?: number; signal?: AbortSignal; mode?: SubagentWaitMode } = {},
|
||||
@@ -137,7 +148,6 @@ export class Supervisor {
|
||||
}
|
||||
|
||||
async shutdown(): Promise<void> {
|
||||
this.clearRecentTerminalTimer();
|
||||
await Promise.allSettled(
|
||||
[...this.children.values()].map(async (child) => {
|
||||
if (!isTerminal(child.record.status.state)) {
|
||||
@@ -146,7 +156,6 @@ export class Supervisor {
|
||||
}
|
||||
}),
|
||||
);
|
||||
this.clearRecentTerminalTimer();
|
||||
}
|
||||
|
||||
private createChild(request: SpawnRequest): SpawnAccepted {
|
||||
@@ -345,39 +354,6 @@ export class Supervisor {
|
||||
private emitChange() {
|
||||
this.options.onChange?.(this.list());
|
||||
for (const waiter of this.waiters) waiter();
|
||||
this.scheduleRecentTerminalExpiry();
|
||||
}
|
||||
|
||||
private scheduleRecentTerminalExpiry() {
|
||||
this.clearRecentTerminalTimer();
|
||||
const ttl = this.options.recentTerminalTtlMs;
|
||||
if (ttl === undefined || ttl <= 0) return;
|
||||
const now = Date.now();
|
||||
const nextExpiryMs = [...this.children.values()]
|
||||
.map((child) => child.record.status)
|
||||
.filter((status) => isTerminal(status.state))
|
||||
.map((status) => Date.parse(status.completedAt ?? status.startedAt))
|
||||
.filter((completed) => Number.isFinite(completed))
|
||||
.map((completed) => completed + ttl - now)
|
||||
.filter((remaining) => remaining > 0)
|
||||
.sort((a, b) => a - b)[0];
|
||||
if (nextExpiryMs === undefined) return;
|
||||
this.recentTerminalTimer = setTimeout(() => this.emitChange(), nextExpiryMs + 1);
|
||||
}
|
||||
|
||||
private clearRecentTerminalTimer() {
|
||||
if (!this.recentTerminalTimer) return;
|
||||
clearTimeout(this.recentTerminalTimer);
|
||||
this.recentTerminalTimer = undefined;
|
||||
}
|
||||
|
||||
private isRecentTerminal(status: SubagentStatus): boolean {
|
||||
const ttl = this.options.recentTerminalTtlMs;
|
||||
if (ttl === undefined) return true;
|
||||
if (ttl <= 0) return false;
|
||||
const completed = Date.parse(status.completedAt ?? status.startedAt);
|
||||
if (!Number.isFinite(completed)) return true;
|
||||
return Date.now() - completed <= ttl;
|
||||
}
|
||||
|
||||
private waitReady(ids: string[], mode: SubagentWaitMode): boolean {
|
||||
|
||||
@@ -6,7 +6,7 @@ export function renderSummary(statuses: SubagentStatus[]): string[] {
|
||||
const queued = statuses.filter((status) => status.state === "queued").length;
|
||||
const terminal = statuses.filter((status) => isTerminalState(status.state)).length;
|
||||
if (running === 0 && queued === 0 && terminal === 0) return [];
|
||||
return [`subagents: ${running} running · ${queued} queued · ${terminal} recent`];
|
||||
return [`subagents: ${running} running · ${queued} queued · ${terminal} terminal`];
|
||||
}
|
||||
|
||||
export function renderInspector(statuses: SubagentStatus[]): string[] {
|
||||
|
||||
Reference in New Issue
Block a user