Compare commits
6 Commits
e143495d6c
...
8913c9958f
| Author | SHA1 | Date | |
|---|---|---|---|
| 8913c9958f | |||
| 30121697ee | |||
| e7a700f010 | |||
| 418d9c986d | |||
| 4e6baed5f0 | |||
| fdf0f89b37 |
141
modules/agents/pi/extensions/subagents/agents.ts
Normal file
141
modules/agents/pi/extensions/subagents/agents.ts
Normal file
@@ -0,0 +1,141 @@
|
|||||||
|
import { existsSync, readdirSync, readFileSync } from "node:fs";
|
||||||
|
import { homedir } from "node:os";
|
||||||
|
import { basename, join } from "node:path";
|
||||||
|
import type { ContextMode } from "./types.ts";
|
||||||
|
import type { Diagnostics } from "./config.ts";
|
||||||
|
|
||||||
|
export interface AgentDefinition {
|
||||||
|
name: string;
|
||||||
|
description: string;
|
||||||
|
body: string;
|
||||||
|
context?: ContextMode;
|
||||||
|
model?: string;
|
||||||
|
thinking?: string;
|
||||||
|
tools?: string;
|
||||||
|
allowedContexts?: ContextMode[];
|
||||||
|
hidden?: boolean;
|
||||||
|
source: string;
|
||||||
|
}
|
||||||
|
|
||||||
|
export function loadAgents(cwd: string, projectTrusted: boolean, diagnostics: Diagnostics, agentDir = defaultAgentDir()): Map<string, AgentDefinition> {
|
||||||
|
const user = loadTier(join(agentDir, "agents"), "user", diagnostics);
|
||||||
|
const project = projectTrusted ? loadTier(join(cwd, ".pi", "agents"), "project", diagnostics) : new Map<string, AgentDefinition>();
|
||||||
|
return new Map([...user, ...project]);
|
||||||
|
}
|
||||||
|
|
||||||
|
function loadTier(dir: string, tier: string, diagnostics: Diagnostics): Map<string, AgentDefinition> {
|
||||||
|
const agents = new Map<string, AgentDefinition>();
|
||||||
|
if (!existsSync(dir)) return agents;
|
||||||
|
for (const entry of readdirSync(dir, { withFileTypes: true })) {
|
||||||
|
if (!entry.isFile() || !entry.name.endsWith(".md")) continue;
|
||||||
|
const path = join(dir, entry.name);
|
||||||
|
const parsed = parseAgent(path, diagnostics);
|
||||||
|
if (!parsed) continue;
|
||||||
|
if (agents.has(parsed.name)) {
|
||||||
|
diagnostics.warnings.push(`Duplicate ${tier} agent '${parsed.name}' ignored at ${path}`);
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
const stem = basename(entry.name, ".md");
|
||||||
|
if (stem !== parsed.name) diagnostics.warnings.push(`${tier} agent file '${entry.name}' name '${parsed.name}' does not match filename`);
|
||||||
|
agents.set(parsed.name, parsed);
|
||||||
|
}
|
||||||
|
return agents;
|
||||||
|
}
|
||||||
|
|
||||||
|
export function parseAgent(path: string, diagnostics: Diagnostics): AgentDefinition | undefined {
|
||||||
|
try {
|
||||||
|
const text = readFileSync(path, "utf8");
|
||||||
|
const match = /^---\n([\s\S]*?)\n---\n?([\s\S]*)$/u.exec(text);
|
||||||
|
if (!match) {
|
||||||
|
diagnostics.warnings.push(`Agent ${path} missing YAML frontmatter`);
|
||||||
|
return undefined;
|
||||||
|
}
|
||||||
|
const frontmatter = parseFrontmatter(match[1]);
|
||||||
|
const name = stringField(frontmatter, "name");
|
||||||
|
const description = stringField(frontmatter, "description");
|
||||||
|
if (!name || !/^[a-z0-9-]+$/.test(name)) {
|
||||||
|
diagnostics.warnings.push(`Agent ${path} has invalid name`);
|
||||||
|
return undefined;
|
||||||
|
}
|
||||||
|
if (!description) {
|
||||||
|
diagnostics.warnings.push(`Agent ${path} has invalid description`);
|
||||||
|
return undefined;
|
||||||
|
}
|
||||||
|
const context = contextField(frontmatter.context);
|
||||||
|
const allowedContexts = contextsField(frontmatter.allowedContexts);
|
||||||
|
if (frontmatter.context !== undefined && !context) diagnostics.warnings.push(`Agent ${path} has invalid context`);
|
||||||
|
if (frontmatter.allowedContexts !== undefined && !allowedContexts) diagnostics.warnings.push(`Agent ${path} has invalid allowedContexts`);
|
||||||
|
if (context && allowedContexts && !allowedContexts.includes(context)) diagnostics.warnings.push(`Agent ${path} context is outside allowedContexts`);
|
||||||
|
return {
|
||||||
|
name,
|
||||||
|
description,
|
||||||
|
body: match[2].trim(),
|
||||||
|
context,
|
||||||
|
model: stringField(frontmatter, "model"),
|
||||||
|
thinking: stringField(frontmatter, "thinking"),
|
||||||
|
tools: stringField(frontmatter, "tools"),
|
||||||
|
allowedContexts,
|
||||||
|
hidden: booleanField(frontmatter, "hidden"),
|
||||||
|
source: path,
|
||||||
|
};
|
||||||
|
} catch (error) {
|
||||||
|
diagnostics.warnings.push(`Failed to load agent ${path}: ${error instanceof Error ? error.message : String(error)}`);
|
||||||
|
return undefined;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
function parseFrontmatter(text: string): Record<string, unknown> {
|
||||||
|
const result: Record<string, unknown> = {};
|
||||||
|
const lines = text.split(/\r?\n/u);
|
||||||
|
for (let i = 0; i < lines.length; i += 1) {
|
||||||
|
const line = lines[i];
|
||||||
|
if (!line.trim() || line.trimStart().startsWith("#")) continue;
|
||||||
|
const scalar = /^(\w+):\s*(.*?)\s*$/u.exec(line);
|
||||||
|
if (!scalar) continue;
|
||||||
|
const [, key, raw] = scalar;
|
||||||
|
if (raw !== "") {
|
||||||
|
result[key] = parseScalar(raw);
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
const values: string[] = [];
|
||||||
|
while (i + 1 < lines.length) {
|
||||||
|
const item = /^\s+-\s*(.*?)\s*$/u.exec(lines[i + 1]);
|
||||||
|
if (!item) break;
|
||||||
|
values.push(String(parseScalar(item[1])));
|
||||||
|
i += 1;
|
||||||
|
}
|
||||||
|
result[key] = values;
|
||||||
|
}
|
||||||
|
return result;
|
||||||
|
}
|
||||||
|
|
||||||
|
function parseScalar(raw: string): string | boolean {
|
||||||
|
const unquoted = raw.replace(/^['"]|['"]$/gu, "");
|
||||||
|
if (unquoted === "true") return true;
|
||||||
|
if (unquoted === "false") return false;
|
||||||
|
return unquoted;
|
||||||
|
}
|
||||||
|
|
||||||
|
function stringField(record: Record<string, unknown>, key: string): string | undefined {
|
||||||
|
const value = record[key];
|
||||||
|
return typeof value === "string" && value.trim() ? value.trim() : undefined;
|
||||||
|
}
|
||||||
|
|
||||||
|
function booleanField(record: Record<string, unknown>, key: string): boolean | undefined {
|
||||||
|
const value = record[key];
|
||||||
|
return typeof value === "boolean" ? value : undefined;
|
||||||
|
}
|
||||||
|
|
||||||
|
function contextField(value: unknown): ContextMode | undefined {
|
||||||
|
return value === "independent" || value === "fork" ? value : undefined;
|
||||||
|
}
|
||||||
|
|
||||||
|
function contextsField(value: unknown): ContextMode[] | undefined {
|
||||||
|
if (!Array.isArray(value)) return undefined;
|
||||||
|
const contexts = value.map(contextField);
|
||||||
|
return contexts.every(Boolean) ? (contexts as ContextMode[]) : undefined;
|
||||||
|
}
|
||||||
|
|
||||||
|
function defaultAgentDir(): string {
|
||||||
|
return process.env.PI_CODING_AGENT_DIR ?? join(homedir(), ".pi", "agent");
|
||||||
|
}
|
||||||
138
modules/agents/pi/extensions/subagents/config.test.ts
Normal file
138
modules/agents/pi/extensions/subagents/config.test.ts
Normal file
@@ -0,0 +1,138 @@
|
|||||||
|
import assert from "node:assert/strict";
|
||||||
|
import { mkdtempSync, mkdirSync, writeFileSync } from "node:fs";
|
||||||
|
import { tmpdir } from "node:os";
|
||||||
|
import { join } from "node:path";
|
||||||
|
import test from "node:test";
|
||||||
|
import { loadAgents } from "./agents.ts";
|
||||||
|
import { BUILT_IN_TOOL_PROFILES, loadConfig, resolveSpawn, type Diagnostics } from "./config.ts";
|
||||||
|
|
||||||
|
function fixture() {
|
||||||
|
const root = mkdtempSync(join(tmpdir(), "subagents-config-"));
|
||||||
|
const agentDir = join(root, "agent");
|
||||||
|
const cwd = join(root, "project");
|
||||||
|
mkdirSync(agentDir, { recursive: true });
|
||||||
|
mkdirSync(cwd, { recursive: true });
|
||||||
|
return { root, agentDir, cwd };
|
||||||
|
}
|
||||||
|
|
||||||
|
function diagnostics(): Diagnostics {
|
||||||
|
return { warnings: [] };
|
||||||
|
}
|
||||||
|
|
||||||
|
test("missing config files and agent directories are normal", () => {
|
||||||
|
const { cwd, agentDir } = fixture();
|
||||||
|
const diag = diagnostics();
|
||||||
|
|
||||||
|
const config = loadConfig(cwd, true, diag, agentDir);
|
||||||
|
const agents = loadAgents(cwd, true, diag, agentDir);
|
||||||
|
|
||||||
|
assert.equal(config.defaultContext, "independent");
|
||||||
|
assert.equal(config.defaultTools, "read-only");
|
||||||
|
assert.equal(config.recentTerminalTtlMs, 300000);
|
||||||
|
assert.equal(agents.size, 0);
|
||||||
|
assert.deepEqual(diag.warnings, []);
|
||||||
|
});
|
||||||
|
|
||||||
|
test("global and trusted project config merge in order", () => {
|
||||||
|
const { cwd, agentDir } = fixture();
|
||||||
|
mkdirSync(join(cwd, ".pi"), { recursive: true });
|
||||||
|
writeFileSync(join(agentDir, "subagents.json"), JSON.stringify({ defaultTools: "global-profile", recentTerminalTtlMs: 1000, toolProfiles: { "global-profile": { activeTools: ["read"] } } }));
|
||||||
|
writeFileSync(join(cwd, ".pi", "subagents.json"), JSON.stringify({ defaultTools: "project-profile", recentTerminalTtlMs: 2000, toolProfiles: { "project-profile": { activeTools: ["ls"] } } }));
|
||||||
|
|
||||||
|
const config = loadConfig(cwd, true, diagnostics(), agentDir);
|
||||||
|
|
||||||
|
assert.equal(config.defaultTools, "project-profile");
|
||||||
|
assert.equal(config.recentTerminalTtlMs, 2000);
|
||||||
|
assert.deepEqual(config.toolProfiles["global-profile"].activeTools, ["read"]);
|
||||||
|
assert.deepEqual(config.toolProfiles["project-profile"].activeTools, ["ls"]);
|
||||||
|
});
|
||||||
|
|
||||||
|
test("recent terminal ttl preserves zero and rejects invalid values", () => {
|
||||||
|
const { cwd, agentDir } = fixture();
|
||||||
|
writeFileSync(join(agentDir, "subagents.json"), JSON.stringify({ recentTerminalTtlMs: 0 }));
|
||||||
|
const zeroDiag = diagnostics();
|
||||||
|
|
||||||
|
const zeroConfig = loadConfig(cwd, true, zeroDiag, agentDir);
|
||||||
|
|
||||||
|
assert.equal(zeroConfig.recentTerminalTtlMs, 0);
|
||||||
|
assert.deepEqual(zeroDiag.warnings, []);
|
||||||
|
|
||||||
|
writeFileSync(join(agentDir, "subagents.json"), JSON.stringify({ recentTerminalTtlMs: -1 }));
|
||||||
|
const invalidDiag = diagnostics();
|
||||||
|
|
||||||
|
const invalidConfig = loadConfig(cwd, true, invalidDiag, agentDir);
|
||||||
|
|
||||||
|
assert.equal(invalidConfig.recentTerminalTtlMs, 300000);
|
||||||
|
assert.ok(invalidDiag.warnings.some((warning) => warning.includes("Invalid global recentTerminalTtlMs ignored")));
|
||||||
|
});
|
||||||
|
|
||||||
|
test("project config is ignored when project is untrusted", () => {
|
||||||
|
const { cwd, agentDir } = fixture();
|
||||||
|
mkdirSync(join(cwd, ".pi"), { recursive: true });
|
||||||
|
writeFileSync(join(cwd, ".pi", "subagents.json"), JSON.stringify({ defaultTools: "project-profile", toolProfiles: { "project-profile": { activeTools: ["ls"] } } }));
|
||||||
|
|
||||||
|
const config = loadConfig(cwd, false, diagnostics(), agentDir);
|
||||||
|
|
||||||
|
assert.equal(config.defaultTools, "read-only");
|
||||||
|
assert.equal(config.toolProfiles["project-profile"], undefined);
|
||||||
|
});
|
||||||
|
|
||||||
|
test("agents load with project precedence over user", () => {
|
||||||
|
const { cwd, agentDir } = fixture();
|
||||||
|
mkdirSync(join(agentDir, "agents"), { recursive: true });
|
||||||
|
mkdirSync(join(cwd, ".pi", "agents"), { recursive: true });
|
||||||
|
writeFileSync(join(agentDir, "agents", "review.md"), "---\nname: review\ndescription: User review\ntools: read-only\n---\nuser body\n");
|
||||||
|
writeFileSync(join(cwd, ".pi", "agents", "review.md"), "---\nname: review\ndescription: Project review\ntools: full-tools\n---\nproject body\n");
|
||||||
|
|
||||||
|
const agents = loadAgents(cwd, true, diagnostics(), agentDir);
|
||||||
|
|
||||||
|
assert.equal(agents.get("review")?.description, "Project review");
|
||||||
|
assert.equal(agents.get("review")?.body, "project body");
|
||||||
|
});
|
||||||
|
|
||||||
|
test("duplicate same-tier definitions and invalid frontmatter produce diagnostics", () => {
|
||||||
|
const { cwd, agentDir } = fixture();
|
||||||
|
const dir = join(agentDir, "agents");
|
||||||
|
mkdirSync(dir, { recursive: true });
|
||||||
|
writeFileSync(join(dir, "one.md"), "---\nname: same\ndescription: One\n---\none\n");
|
||||||
|
writeFileSync(join(dir, "two.md"), "---\nname: same\ndescription: Two\n---\ntwo\n");
|
||||||
|
writeFileSync(join(dir, "bad.md"), "---\nname: Bad Name\n---\nbad\n");
|
||||||
|
const diag = diagnostics();
|
||||||
|
|
||||||
|
const agents = loadAgents(cwd, true, diag, agentDir);
|
||||||
|
|
||||||
|
assert.equal(agents.size, 1);
|
||||||
|
assert.ok(diag.warnings.some((warning) => warning.includes("Duplicate user agent 'same'")));
|
||||||
|
assert.ok(diag.warnings.some((warning) => warning.includes("invalid name")));
|
||||||
|
});
|
||||||
|
|
||||||
|
test("named spawn resolves overrides, frontmatter, config, and defaults", () => {
|
||||||
|
const { cwd, agentDir } = fixture();
|
||||||
|
mkdirSync(join(agentDir, "agents"), { recursive: true });
|
||||||
|
writeFileSync(join(agentDir, "subagents.json"), JSON.stringify({ defaultTools: "local-review", toolProfiles: { "local-review": { activeTools: ["read"] } } }));
|
||||||
|
writeFileSync(join(agentDir, "agents", "review.md"), "---\nname: review\ndescription: Review\ncontext: independent\nmodel: inherit\nthinking: high\ntools: local-review\n---\nagent body\n");
|
||||||
|
const diag = diagnostics();
|
||||||
|
const config = loadConfig(cwd, true, diag, agentDir);
|
||||||
|
const agents = loadAgents(cwd, true, diag, agentDir);
|
||||||
|
|
||||||
|
const resolved = resolveSpawn({ agent: "review", prompt: "check this", thinking: "low" }, config, agents);
|
||||||
|
|
||||||
|
assert.equal(resolved.prompt, "check this");
|
||||||
|
assert.equal(resolved.context, "independent");
|
||||||
|
assert.equal(resolved.model, "inherit");
|
||||||
|
assert.equal(resolved.thinking, "low");
|
||||||
|
assert.equal(resolved.tools, "local-review");
|
||||||
|
assert.deepEqual(resolved.toolProfile.activeTools, ["read"]);
|
||||||
|
assert.equal(resolved.agentBody, "agent body");
|
||||||
|
});
|
||||||
|
|
||||||
|
test("built-in tool profile names cannot be overridden", () => {
|
||||||
|
const { cwd, agentDir } = fixture();
|
||||||
|
writeFileSync(join(agentDir, "subagents.json"), JSON.stringify({ toolProfiles: { "read-only": { activeTools: ["bash"] } } }));
|
||||||
|
const diag = diagnostics();
|
||||||
|
|
||||||
|
const config = loadConfig(cwd, true, diag, agentDir);
|
||||||
|
|
||||||
|
assert.deepEqual(config.toolProfiles["read-only"], BUILT_IN_TOOL_PROFILES["read-only"]);
|
||||||
|
assert.ok(diag.warnings.some((warning) => warning.includes("Ignoring global override for built-in tool profile 'read-only'")));
|
||||||
|
});
|
||||||
182
modules/agents/pi/extensions/subagents/config.ts
Normal file
182
modules/agents/pi/extensions/subagents/config.ts
Normal file
@@ -0,0 +1,182 @@
|
|||||||
|
import { existsSync, readFileSync } from "node:fs";
|
||||||
|
import { homedir } from "node:os";
|
||||||
|
import { join } from "node:path";
|
||||||
|
import type { ContextMode, SpawnRequest, ToolProfile } from "./types.ts";
|
||||||
|
import type { AgentDefinition } from "./agents.ts";
|
||||||
|
|
||||||
|
export interface Diagnostics {
|
||||||
|
warnings: string[];
|
||||||
|
}
|
||||||
|
|
||||||
|
export interface SubagentsConfig {
|
||||||
|
defaultContext: ContextMode;
|
||||||
|
defaultTools: string;
|
||||||
|
maxConcurrent: number;
|
||||||
|
recentTerminalTtlMs: number;
|
||||||
|
ui: {
|
||||||
|
enabled: boolean;
|
||||||
|
defaultExpanded: boolean;
|
||||||
|
};
|
||||||
|
toolProfiles: Record<string, ToolProfile>;
|
||||||
|
}
|
||||||
|
|
||||||
|
export interface ResolvedSpawnRequest extends SpawnRequest {
|
||||||
|
prompt: string;
|
||||||
|
context: ContextMode;
|
||||||
|
tools: string;
|
||||||
|
toolProfile: ToolProfile;
|
||||||
|
agentBody?: string;
|
||||||
|
}
|
||||||
|
|
||||||
|
export const BUILT_IN_TOOL_PROFILES: Record<string, ToolProfile> = {
|
||||||
|
none: { activeTools: [] },
|
||||||
|
"read-only": { activeTools: ["read", "grep", "find", "ls"] },
|
||||||
|
"read-only-with-safe-bash": { activeTools: ["read", "grep", "find", "ls", "bash"] },
|
||||||
|
"full-tools": { activeTools: null },
|
||||||
|
};
|
||||||
|
|
||||||
|
const DEFAULT_CONFIG: SubagentsConfig = {
|
||||||
|
defaultContext: "independent",
|
||||||
|
defaultTools: "read-only",
|
||||||
|
maxConcurrent: 3,
|
||||||
|
recentTerminalTtlMs: 5 * 60 * 1000,
|
||||||
|
ui: { enabled: true, defaultExpanded: false },
|
||||||
|
toolProfiles: { ...BUILT_IN_TOOL_PROFILES },
|
||||||
|
};
|
||||||
|
|
||||||
|
export function loadConfig(cwd: string, projectTrusted: boolean, diagnostics: Diagnostics, agentDir = defaultAgentDir()): SubagentsConfig {
|
||||||
|
let config = cloneConfig(DEFAULT_CONFIG);
|
||||||
|
config = mergeConfig(config, readConfig(join(agentDir, "subagents.json"), diagnostics, "global"), diagnostics, "global");
|
||||||
|
if (projectTrusted) {
|
||||||
|
config = mergeConfig(config, readConfig(join(cwd, ".pi", "subagents.json"), diagnostics, "project"), diagnostics, "project");
|
||||||
|
}
|
||||||
|
if (!config.toolProfiles[config.defaultTools]) {
|
||||||
|
diagnostics.warnings.push(`Unknown defaultTools profile '${config.defaultTools}', using read-only`);
|
||||||
|
config.defaultTools = "read-only";
|
||||||
|
}
|
||||||
|
return config;
|
||||||
|
}
|
||||||
|
|
||||||
|
export function resolveSpawn(request: SpawnRequest, config: SubagentsConfig, agents: Map<string, AgentDefinition>): ResolvedSpawnRequest {
|
||||||
|
const prompt = typeof request.prompt === "string" ? request.prompt.trim() : "";
|
||||||
|
if (!prompt) throw new Error("prompt is required");
|
||||||
|
const agent = request.agent ? agents.get(request.agent) : undefined;
|
||||||
|
if (request.agent && !agent) throw new Error(`unknown subagent agent: ${request.agent}`);
|
||||||
|
|
||||||
|
const context = request.context ?? agent?.context ?? config.defaultContext;
|
||||||
|
if (context !== "independent" && context !== "fork") throw new Error(`unsupported context: ${context}`);
|
||||||
|
if (agent?.allowedContexts && !agent.allowedContexts.includes(context)) {
|
||||||
|
throw new Error(`agent '${agent.name}' does not allow ${context} context`);
|
||||||
|
}
|
||||||
|
|
||||||
|
const tools = request.tools ?? agent?.tools ?? config.defaultTools;
|
||||||
|
const toolProfile = config.toolProfiles[tools];
|
||||||
|
if (!toolProfile) throw new Error(`unknown tool profile: ${tools}`);
|
||||||
|
|
||||||
|
return {
|
||||||
|
...request,
|
||||||
|
prompt,
|
||||||
|
agent: agent?.name ?? request.agent,
|
||||||
|
context,
|
||||||
|
model: request.model ?? agent?.model,
|
||||||
|
thinking: request.thinking ?? agent?.thinking,
|
||||||
|
tools,
|
||||||
|
toolProfile,
|
||||||
|
agentBody: agent?.body,
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
function readConfig(path: string, diagnostics: Diagnostics, label: string): Partial<SubagentsConfig> | undefined {
|
||||||
|
if (!existsSync(path)) return undefined;
|
||||||
|
try {
|
||||||
|
const parsed = JSON.parse(readFileSync(path, "utf8"));
|
||||||
|
return normalizeConfig(parsed, diagnostics, label);
|
||||||
|
} catch (error) {
|
||||||
|
diagnostics.warnings.push(`Invalid ${label} subagents.json: ${error instanceof Error ? error.message : String(error)}`);
|
||||||
|
return undefined;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
function normalizeConfig(raw: unknown, diagnostics: Diagnostics, label: string): Partial<SubagentsConfig> | undefined {
|
||||||
|
if (!raw || typeof raw !== "object" || Array.isArray(raw)) {
|
||||||
|
diagnostics.warnings.push(`Invalid ${label} subagents.json: root must be an object`);
|
||||||
|
return undefined;
|
||||||
|
}
|
||||||
|
const input = raw as Record<string, unknown>;
|
||||||
|
const config: Partial<SubagentsConfig> = {};
|
||||||
|
if (input.defaultContext === "independent" || input.defaultContext === "fork") config.defaultContext = input.defaultContext;
|
||||||
|
else if (input.defaultContext !== undefined) diagnostics.warnings.push(`Invalid ${label} defaultContext ignored`);
|
||||||
|
if (typeof input.defaultTools === "string") config.defaultTools = input.defaultTools;
|
||||||
|
else if (input.defaultTools !== undefined) diagnostics.warnings.push(`Invalid ${label} defaultTools ignored`);
|
||||||
|
if (typeof input.maxConcurrent === "number" && Number.isInteger(input.maxConcurrent) && input.maxConcurrent > 0) config.maxConcurrent = input.maxConcurrent;
|
||||||
|
else if (input.maxConcurrent !== undefined) diagnostics.warnings.push(`Invalid ${label} maxConcurrent ignored`);
|
||||||
|
if (typeof input.recentTerminalTtlMs === "number" && Number.isInteger(input.recentTerminalTtlMs) && input.recentTerminalTtlMs >= 0) {
|
||||||
|
config.recentTerminalTtlMs = input.recentTerminalTtlMs;
|
||||||
|
} else if (input.recentTerminalTtlMs !== undefined) diagnostics.warnings.push(`Invalid ${label} recentTerminalTtlMs ignored`);
|
||||||
|
if (input.ui !== undefined) config.ui = normalizeUi(input.ui, diagnostics, label);
|
||||||
|
if (input.toolProfiles !== undefined) config.toolProfiles = normalizeProfiles(input.toolProfiles, diagnostics, label);
|
||||||
|
return config;
|
||||||
|
}
|
||||||
|
|
||||||
|
function normalizeUi(raw: unknown, diagnostics: Diagnostics, label: string): SubagentsConfig["ui"] | undefined {
|
||||||
|
if (!raw || typeof raw !== "object" || Array.isArray(raw)) {
|
||||||
|
diagnostics.warnings.push(`Invalid ${label} ui ignored`);
|
||||||
|
return undefined;
|
||||||
|
}
|
||||||
|
const input = raw as Record<string, unknown>;
|
||||||
|
return {
|
||||||
|
enabled: typeof input.enabled === "boolean" ? input.enabled : DEFAULT_CONFIG.ui.enabled,
|
||||||
|
defaultExpanded: typeof input.defaultExpanded === "boolean" ? input.defaultExpanded : DEFAULT_CONFIG.ui.defaultExpanded,
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
function normalizeProfiles(raw: unknown, diagnostics: Diagnostics, label: string): Record<string, ToolProfile> {
|
||||||
|
const profiles: Record<string, ToolProfile> = {};
|
||||||
|
if (!raw || typeof raw !== "object" || Array.isArray(raw)) {
|
||||||
|
diagnostics.warnings.push(`Invalid ${label} toolProfiles ignored`);
|
||||||
|
return profiles;
|
||||||
|
}
|
||||||
|
for (const [name, value] of Object.entries(raw as Record<string, unknown>)) {
|
||||||
|
if (name in BUILT_IN_TOOL_PROFILES) {
|
||||||
|
diagnostics.warnings.push(`Ignoring ${label} override for built-in tool profile '${name}'`);
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
const profile = normalizeProfile(value);
|
||||||
|
if (!profile) {
|
||||||
|
diagnostics.warnings.push(`Invalid ${label} tool profile '${name}' ignored`);
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
profiles[name] = profile;
|
||||||
|
}
|
||||||
|
return profiles;
|
||||||
|
}
|
||||||
|
|
||||||
|
function normalizeProfile(raw: unknown): ToolProfile | undefined {
|
||||||
|
if (!raw || typeof raw !== "object" || Array.isArray(raw)) return undefined;
|
||||||
|
const activeTools = (raw as { activeTools?: unknown }).activeTools;
|
||||||
|
if (!Array.isArray(activeTools) || !activeTools.every((tool) => typeof tool === "string")) return undefined;
|
||||||
|
return { activeTools };
|
||||||
|
}
|
||||||
|
|
||||||
|
function mergeConfig(base: SubagentsConfig, override: Partial<SubagentsConfig> | undefined, diagnostics: Diagnostics, label: string): SubagentsConfig {
|
||||||
|
if (!override) return base;
|
||||||
|
const merged = cloneConfig(base);
|
||||||
|
if (override.defaultContext) merged.defaultContext = override.defaultContext;
|
||||||
|
if (override.defaultTools) merged.defaultTools = override.defaultTools;
|
||||||
|
if (override.maxConcurrent) merged.maxConcurrent = override.maxConcurrent;
|
||||||
|
if (override.recentTerminalTtlMs !== undefined) merged.recentTerminalTtlMs = override.recentTerminalTtlMs;
|
||||||
|
if (override.ui) merged.ui = { ...merged.ui, ...override.ui };
|
||||||
|
if (override.toolProfiles) merged.toolProfiles = { ...merged.toolProfiles, ...override.toolProfiles };
|
||||||
|
for (const key of Object.keys(merged.toolProfiles)) {
|
||||||
|
if (key in BUILT_IN_TOOL_PROFILES) merged.toolProfiles[key] = BUILT_IN_TOOL_PROFILES[key];
|
||||||
|
}
|
||||||
|
return merged;
|
||||||
|
}
|
||||||
|
|
||||||
|
function cloneConfig(config: SubagentsConfig): SubagentsConfig {
|
||||||
|
return { ...config, ui: { ...config.ui }, toolProfiles: { ...config.toolProfiles } };
|
||||||
|
}
|
||||||
|
|
||||||
|
function defaultAgentDir(): string {
|
||||||
|
return process.env.PI_CODING_AGENT_DIR ?? join(homedir(), ".pi", "agent");
|
||||||
|
}
|
||||||
296
modules/agents/pi/extensions/subagents/index.ts
Normal file
296
modules/agents/pi/extensions/subagents/index.ts
Normal file
@@ -0,0 +1,296 @@
|
|||||||
|
import type { ExtensionAPI, ExtensionContext } from "@earendil-works/pi-coding-agent";
|
||||||
|
import { Type } from "typebox";
|
||||||
|
import { loadAgents } from "./agents.ts";
|
||||||
|
import { loadConfig, resolveSpawn, type Diagnostics } from "./config.ts";
|
||||||
|
import { SubprocessRpcRunner } from "./runner.ts";
|
||||||
|
import { Supervisor } from "./supervisor.ts";
|
||||||
|
import type { SpawnRequest, SubagentStatus } from "./types.ts";
|
||||||
|
import { widget } from "./ui.ts";
|
||||||
|
|
||||||
|
let supervisor: Supervisor | undefined;
|
||||||
|
let lastDiagnostics: Diagnostics = { warnings: [] };
|
||||||
|
let lastStatuses: SubagentStatus[] = [];
|
||||||
|
let uiExpanded = false;
|
||||||
|
|
||||||
|
export default function subagents(pi: ExtensionAPI) {
|
||||||
|
const getSupervisor = (ctx: ExtensionContext): Supervisor => {
|
||||||
|
if (supervisor) return supervisor;
|
||||||
|
const diagnostics: Diagnostics = { warnings: [] };
|
||||||
|
const cwd = cwdOf(ctx);
|
||||||
|
const config = loadConfig(cwd, isProjectTrusted(ctx), diagnostics);
|
||||||
|
lastDiagnostics = diagnostics;
|
||||||
|
uiExpanded = config.ui.defaultExpanded;
|
||||||
|
supervisor = new Supervisor(new SubprocessRpcRunner(), cwd, {
|
||||||
|
maxConcurrent: config.maxConcurrent,
|
||||||
|
recentTerminalTtlMs: config.recentTerminalTtlMs,
|
||||||
|
onMilestone: (status, event) => pi.appendEntry("subagent_milestone", { event, status }),
|
||||||
|
onChange: (statuses) => {
|
||||||
|
lastStatuses = statuses;
|
||||||
|
updateUi(ctx, config.ui.enabled);
|
||||||
|
},
|
||||||
|
});
|
||||||
|
updateUi(ctx, config.ui.enabled);
|
||||||
|
return supervisor;
|
||||||
|
};
|
||||||
|
|
||||||
|
const resolve = (ctx: ExtensionContext, request: SpawnRequest): SpawnRequest => {
|
||||||
|
const diagnostics: Diagnostics = { warnings: [] };
|
||||||
|
const cwd = cwdOf(ctx);
|
||||||
|
const trusted = isProjectTrusted(ctx);
|
||||||
|
const config = loadConfig(cwd, trusted, diagnostics);
|
||||||
|
const agents = loadAgents(cwd, trusted, diagnostics);
|
||||||
|
lastDiagnostics = diagnostics;
|
||||||
|
const resolved = resolveSpawn(request, config, agents);
|
||||||
|
if (resolved.context === "fork") resolved.parentSessionFile = ctx.sessionManager.getSessionFile();
|
||||||
|
return resolved;
|
||||||
|
};
|
||||||
|
|
||||||
|
pi.registerTool({
|
||||||
|
name: "subagent_spawn",
|
||||||
|
label: "Spawn subagent",
|
||||||
|
description: "Start one ad hoc independent subagent and return immediately with its child id",
|
||||||
|
parameters: Type.Object({
|
||||||
|
prompt: Type.String({ description: "Prompt for the delegated subagent" }),
|
||||||
|
agent: Type.Optional(Type.String({ description: "Named agent definition to use" })),
|
||||||
|
context: Type.Optional(Type.Union([Type.Literal("independent"), Type.Literal("fork")])),
|
||||||
|
model: Type.Optional(Type.String({ description: "Optional model selector for the child" })),
|
||||||
|
thinking: Type.Optional(Type.String({ description: "Optional thinking level for the child" })),
|
||||||
|
tools: Type.Optional(Type.String({ description: "Tool profile name" })),
|
||||||
|
}),
|
||||||
|
async execute(_toolCallId, params, _signal, _onUpdate, ctx) {
|
||||||
|
const accepted = getSupervisor(ctx).spawn(resolve(ctx, params as SpawnRequest));
|
||||||
|
ctx.ui?.notify?.(`Started subagent ${accepted.id}`, "info");
|
||||||
|
return textResult(accepted);
|
||||||
|
},
|
||||||
|
});
|
||||||
|
|
||||||
|
pi.registerTool({
|
||||||
|
name: "subagent_batch",
|
||||||
|
label: "Spawn subagent batch",
|
||||||
|
description: "Start multiple subagents and return immediately with accepted child ids and per-entry failures",
|
||||||
|
parameters: Type.Object({
|
||||||
|
subagents: Type.Array(
|
||||||
|
Type.Object({
|
||||||
|
prompt: Type.String({ description: "Prompt for the delegated subagent" }),
|
||||||
|
agent: Type.Optional(Type.String({ description: "Named agent definition to use" })),
|
||||||
|
context: Type.Optional(Type.Union([Type.Literal("independent"), Type.Literal("fork")])),
|
||||||
|
model: Type.Optional(Type.String({ description: "Optional model selector for the child" })),
|
||||||
|
thinking: Type.Optional(Type.String({ description: "Optional thinking level for the child" })),
|
||||||
|
tools: Type.Optional(Type.String({ description: "Tool profile name" })),
|
||||||
|
}),
|
||||||
|
),
|
||||||
|
}),
|
||||||
|
async execute(_toolCallId, params, _signal, _onUpdate, ctx) {
|
||||||
|
const requests = Array.isArray((params as { subagents?: unknown }).subagents) ? ((params as { subagents: SpawnRequest[] }).subagents) : [];
|
||||||
|
const accepted: SpawnRequest[] = [];
|
||||||
|
const failed: Array<{ index: number; error: string }> = [];
|
||||||
|
requests.forEach((request, index) => {
|
||||||
|
try {
|
||||||
|
accepted.push(resolve(ctx, request));
|
||||||
|
} catch (error) {
|
||||||
|
failed.push({ index, error: error instanceof Error ? error.message : String(error) });
|
||||||
|
}
|
||||||
|
});
|
||||||
|
const result = getSupervisor(ctx).spawnBatch(accepted);
|
||||||
|
return textResult({ accepted: result.accepted, failed: [...failed, ...result.failed] });
|
||||||
|
},
|
||||||
|
});
|
||||||
|
|
||||||
|
pi.registerTool({
|
||||||
|
name: "subagent_list",
|
||||||
|
label: "List subagents",
|
||||||
|
description: "List active and recent subagents for this parent session",
|
||||||
|
parameters: Type.Object({}),
|
||||||
|
async execute(_toolCallId, _params, _signal, _onUpdate, ctx) {
|
||||||
|
return textResult(getSupervisor(ctx).list());
|
||||||
|
},
|
||||||
|
});
|
||||||
|
|
||||||
|
pi.registerTool({
|
||||||
|
name: "subagent_status",
|
||||||
|
label: "Get subagent status",
|
||||||
|
description: "Get current lifecycle status for one subagent",
|
||||||
|
parameters: Type.Object({
|
||||||
|
id: Type.String({ description: "Subagent id returned by subagent_spawn" }),
|
||||||
|
}),
|
||||||
|
async execute(_toolCallId, params, _signal, _onUpdate, ctx) {
|
||||||
|
return textResult(getSupervisor(ctx).status(String((params as { id: unknown }).id)));
|
||||||
|
},
|
||||||
|
});
|
||||||
|
|
||||||
|
pi.registerTool({
|
||||||
|
name: "subagent_result",
|
||||||
|
label: "Get subagent result",
|
||||||
|
description: "Return still-running before completion and the final answer after completion",
|
||||||
|
parameters: Type.Object({
|
||||||
|
id: Type.String({ description: "Subagent id returned by subagent_spawn" }),
|
||||||
|
}),
|
||||||
|
async execute(_toolCallId, params, _signal, _onUpdate, ctx) {
|
||||||
|
return textResult(getSupervisor(ctx).result(String((params as { id: unknown }).id)));
|
||||||
|
},
|
||||||
|
});
|
||||||
|
|
||||||
|
pi.registerTool({
|
||||||
|
name: "subagent_wait",
|
||||||
|
label: "Wait for subagents",
|
||||||
|
description: "Block until multiple subagents are terminal or a timeout expires. Prefer setting timeoutMs so the parent turn cannot hang forever",
|
||||||
|
parameters: Type.Object({
|
||||||
|
ids: Type.Array(Type.String({ description: "Subagent id returned by subagent_spawn or subagent_batch" })),
|
||||||
|
timeoutMs: Type.Optional(Type.Number({ description: "Maximum milliseconds to wait. Omit or use 0 to wait indefinitely" })),
|
||||||
|
mode: Type.Optional(Type.Union([Type.Literal("all"), Type.Literal("any")], { description: "Wait for all ids by default, or return after any id is terminal" })),
|
||||||
|
}),
|
||||||
|
async execute(_toolCallId, params, signal, _onUpdate, ctx) {
|
||||||
|
const input = params as { ids?: unknown; timeoutMs?: unknown; mode?: unknown };
|
||||||
|
const ids = Array.isArray(input.ids) ? input.ids.map(String) : [];
|
||||||
|
const timeoutMs = typeof input.timeoutMs === "number" && Number.isFinite(input.timeoutMs) ? input.timeoutMs : undefined;
|
||||||
|
const mode = input.mode === "any" ? "any" : "all";
|
||||||
|
return textResult(await getSupervisor(ctx).wait(ids, { timeoutMs, mode, signal }));
|
||||||
|
},
|
||||||
|
});
|
||||||
|
|
||||||
|
pi.registerTool({
|
||||||
|
name: "subagent_cancel",
|
||||||
|
label: "Cancel subagent",
|
||||||
|
description: "Cancel a running subagent",
|
||||||
|
parameters: Type.Object({
|
||||||
|
id: Type.String({ description: "Subagent id returned by subagent_spawn" }),
|
||||||
|
}),
|
||||||
|
async execute(_toolCallId, params, _signal, _onUpdate, ctx) {
|
||||||
|
return textResult(await getSupervisor(ctx).cancel(String((params as { id: unknown }).id)));
|
||||||
|
},
|
||||||
|
});
|
||||||
|
|
||||||
|
pi.registerCommand("subagent-spawn", {
|
||||||
|
description: "Start an ad hoc independent subagent",
|
||||||
|
handler: async (args, ctx) => {
|
||||||
|
const accepted = getSupervisor(ctx).spawn(resolve(ctx, parseSpawnArgs(args)));
|
||||||
|
ctx.ui.notify(`Started subagent ${accepted.id}`, "info");
|
||||||
|
},
|
||||||
|
});
|
||||||
|
|
||||||
|
pi.registerCommand("subagent-batch", {
|
||||||
|
description: "Start ad hoc independent subagents split by |",
|
||||||
|
handler: async (args, ctx) => {
|
||||||
|
const requests = args
|
||||||
|
.split("|")
|
||||||
|
.map((prompt) => prompt.trim())
|
||||||
|
.filter(Boolean)
|
||||||
|
.map((prompt) => resolve(ctx, { prompt }));
|
||||||
|
ctx.ui.notify(JSON.stringify(getSupervisor(ctx).spawnBatch(requests), null, 2), "info");
|
||||||
|
},
|
||||||
|
});
|
||||||
|
|
||||||
|
pi.registerCommand("subagent-list", {
|
||||||
|
description: "Show subagent status records",
|
||||||
|
handler: async (_args, ctx) => {
|
||||||
|
ctx.ui.notify(JSON.stringify(getSupervisor(ctx).list(), null, 2), "info");
|
||||||
|
},
|
||||||
|
});
|
||||||
|
|
||||||
|
pi.registerCommand("subagent-status", {
|
||||||
|
description: "Show a subagent status by id",
|
||||||
|
handler: async (args, ctx) => {
|
||||||
|
ctx.ui.notify(JSON.stringify(getSupervisor(ctx).status(args.trim()), null, 2), "info");
|
||||||
|
},
|
||||||
|
});
|
||||||
|
|
||||||
|
pi.registerCommand("subagent-result", {
|
||||||
|
description: "Show a subagent result by id",
|
||||||
|
handler: async (args, ctx) => {
|
||||||
|
ctx.ui.notify(JSON.stringify(getSupervisor(ctx).result(args.trim()), null, 2), "info");
|
||||||
|
},
|
||||||
|
});
|
||||||
|
|
||||||
|
pi.registerCommand("subagent-wait", {
|
||||||
|
description: "Wait for subagent ids separated by spaces",
|
||||||
|
handler: async (args, ctx) => {
|
||||||
|
const { ids, timeoutMs, mode } = parseWaitArgs(args);
|
||||||
|
ctx.ui.notify(JSON.stringify(await getSupervisor(ctx).wait(ids, { timeoutMs, mode }), null, 2), "info");
|
||||||
|
},
|
||||||
|
});
|
||||||
|
|
||||||
|
pi.registerCommand("subagent-ui", {
|
||||||
|
description: "Toggle the bundled subagent status inspector",
|
||||||
|
handler: async (_args, ctx) => {
|
||||||
|
uiExpanded = !uiExpanded;
|
||||||
|
updateUi(ctx, true);
|
||||||
|
ctx.ui.notify(`Subagent inspector ${uiExpanded ? "expanded" : "collapsed"}`, "info");
|
||||||
|
},
|
||||||
|
});
|
||||||
|
|
||||||
|
pi.registerCommand("subagent-diagnostics", {
|
||||||
|
description: "Show subagent configuration diagnostics from the last load",
|
||||||
|
handler: async (_args, ctx) => {
|
||||||
|
ctx.ui.notify(JSON.stringify(lastDiagnostics, null, 2), "info");
|
||||||
|
},
|
||||||
|
});
|
||||||
|
|
||||||
|
pi.registerCommand("subagent-cancel", {
|
||||||
|
description: "Cancel a running subagent by id",
|
||||||
|
handler: async (args, ctx) => {
|
||||||
|
ctx.ui.notify(JSON.stringify(await getSupervisor(ctx).cancel(args.trim()), null, 2), "info");
|
||||||
|
},
|
||||||
|
});
|
||||||
|
|
||||||
|
pi.on("session_shutdown", async () => {
|
||||||
|
await supervisor?.shutdown();
|
||||||
|
supervisor = undefined;
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
function updateUi(ctx: ExtensionContext, enabled: boolean) {
|
||||||
|
if (!ctx.hasUI) return;
|
||||||
|
ctx.ui.setWidget("subagents", enabled ? widget(lastStatuses, uiExpanded) : undefined);
|
||||||
|
}
|
||||||
|
|
||||||
|
function parseSpawnArgs(args: string): SpawnRequest {
|
||||||
|
const parts = args.trim().split(/\s+/u);
|
||||||
|
const request: Partial<SpawnRequest> = {};
|
||||||
|
while (parts.length >= 2 && parts[0].startsWith("--")) {
|
||||||
|
const flag = parts.shift();
|
||||||
|
const value = parts.shift();
|
||||||
|
if (flag === "--agent") request.agent = value;
|
||||||
|
else if (flag === "--context" && (value === "independent" || value === "fork")) request.context = value;
|
||||||
|
else if (flag === "--tools") request.tools = value;
|
||||||
|
else if (flag === "--model") request.model = value;
|
||||||
|
else if (flag === "--thinking") request.thinking = value;
|
||||||
|
}
|
||||||
|
return { ...request, prompt: parts.join(" ") || args } as SpawnRequest;
|
||||||
|
}
|
||||||
|
|
||||||
|
function parseWaitArgs(args: string): { ids: string[]; timeoutMs?: number; mode?: "all" | "any" } {
|
||||||
|
const parts = args.trim().split(/\s+/u).filter(Boolean);
|
||||||
|
let timeoutMs: number | undefined;
|
||||||
|
let mode: "all" | "any" | undefined;
|
||||||
|
const ids: string[] = [];
|
||||||
|
while (parts.length > 0) {
|
||||||
|
const part = parts.shift();
|
||||||
|
if (!part) continue;
|
||||||
|
if (part === "--timeout-ms" && parts[0]) {
|
||||||
|
const parsed = Number(parts.shift());
|
||||||
|
if (Number.isFinite(parsed)) timeoutMs = parsed;
|
||||||
|
} else if (part === "--mode" && (parts[0] === "all" || parts[0] === "any")) {
|
||||||
|
mode = parts.shift() as "all" | "any";
|
||||||
|
} else {
|
||||||
|
ids.push(part);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return { ids, timeoutMs, mode };
|
||||||
|
}
|
||||||
|
|
||||||
|
function isProjectTrusted(ctx: ExtensionContext): boolean {
|
||||||
|
const value = (ctx as unknown as { isProjectTrusted?: () => boolean }).isProjectTrusted?.();
|
||||||
|
return value === true;
|
||||||
|
}
|
||||||
|
|
||||||
|
function cwdOf(ctx: ExtensionContext): string {
|
||||||
|
const sessionCwd = (ctx as unknown as { sessionManager?: { getCwd?: () => string }; cwd?: string }).sessionManager?.getCwd?.();
|
||||||
|
return sessionCwd ?? (ctx as unknown as { cwd?: string }).cwd ?? process.cwd();
|
||||||
|
}
|
||||||
|
|
||||||
|
function textResult(value: unknown) {
|
||||||
|
return {
|
||||||
|
content: [{ type: "text" as const, text: JSON.stringify(value, null, 2) }],
|
||||||
|
details: value,
|
||||||
|
};
|
||||||
|
}
|
||||||
212
modules/agents/pi/extensions/subagents/runner.ts
Normal file
212
modules/agents/pi/extensions/subagents/runner.ts
Normal file
@@ -0,0 +1,212 @@
|
|||||||
|
import { spawn, type ChildProcessWithoutNullStreams } from "node:child_process";
|
||||||
|
import type { ChildHandle, ChildRunner, RunnerEvents, SpawnRequest } from "./types.ts";
|
||||||
|
|
||||||
|
interface PendingResponse {
|
||||||
|
resolve(value: unknown): void;
|
||||||
|
reject(error: Error): void;
|
||||||
|
command: string;
|
||||||
|
}
|
||||||
|
|
||||||
|
interface RpcLine {
|
||||||
|
id?: string;
|
||||||
|
type?: string;
|
||||||
|
command?: string;
|
||||||
|
success?: boolean;
|
||||||
|
data?: unknown;
|
||||||
|
error?: string;
|
||||||
|
message?: string;
|
||||||
|
}
|
||||||
|
|
||||||
|
class RpcChildHandle implements ChildHandle {
|
||||||
|
private buffer = "";
|
||||||
|
private nextRequest = 0;
|
||||||
|
private settled = false;
|
||||||
|
private finishing = false;
|
||||||
|
private cancelling = false;
|
||||||
|
private killed = false;
|
||||||
|
private readonly pending = new Map<string, PendingResponse>();
|
||||||
|
|
||||||
|
constructor(
|
||||||
|
private readonly child: ChildProcessWithoutNullStreams,
|
||||||
|
private readonly events: RunnerEvents,
|
||||||
|
) {
|
||||||
|
child.stdout.setEncoding("utf8");
|
||||||
|
child.stderr.setEncoding("utf8");
|
||||||
|
child.stdout.on("data", (chunk) => this.onStdout(chunk));
|
||||||
|
child.stderr.on("data", (chunk) => this.events.running(`stderr: ${String(chunk).trim().slice(0, 200)}`));
|
||||||
|
child.on("error", (error) => this.fail(error.message));
|
||||||
|
child.on("close", (code, signal) => {
|
||||||
|
for (const pending of this.pending.values()) {
|
||||||
|
pending.reject(new Error(`RPC process closed before ${pending.command} response`));
|
||||||
|
}
|
||||||
|
this.pending.clear();
|
||||||
|
if (!this.settled) this.fail(`RPC process closed with code ${code ?? "null"} signal ${signal ?? "null"}`);
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
async prompt(message: string): Promise<void> {
|
||||||
|
await this.send("prompt", { message });
|
||||||
|
}
|
||||||
|
|
||||||
|
async cancel(): Promise<void> {
|
||||||
|
if (this.cancelling) return;
|
||||||
|
this.cancelling = true;
|
||||||
|
try {
|
||||||
|
await Promise.race([this.send("abort", {}), delay(200)]);
|
||||||
|
} catch {}
|
||||||
|
this.terminate();
|
||||||
|
}
|
||||||
|
|
||||||
|
private onStdout(chunk: string) {
|
||||||
|
this.buffer += chunk;
|
||||||
|
while (true) {
|
||||||
|
const newline = this.buffer.indexOf("\n");
|
||||||
|
if (newline === -1) return;
|
||||||
|
const line = this.buffer.slice(0, newline).replace(/\r$/, "");
|
||||||
|
this.buffer = this.buffer.slice(newline + 1);
|
||||||
|
if (line.trim() === "") continue;
|
||||||
|
this.onLine(line);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private onLine(line: string) {
|
||||||
|
let payload: RpcLine;
|
||||||
|
try {
|
||||||
|
payload = JSON.parse(line);
|
||||||
|
} catch {
|
||||||
|
this.events.running(`non-json rpc output: ${line.slice(0, 200)}`);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
if (payload.type === "response" && payload.id) {
|
||||||
|
const pending = this.pending.get(payload.id);
|
||||||
|
if (!pending) return;
|
||||||
|
this.pending.delete(payload.id);
|
||||||
|
if (payload.success) pending.resolve(payload.data);
|
||||||
|
else pending.reject(new Error(payload.error ?? payload.message ?? `${pending.command} failed`));
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
if (payload.type === "agent_started") {
|
||||||
|
this.events.running("agent_started");
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
if (payload.type === "agent_settled") {
|
||||||
|
this.finish().catch((error) => this.fail(error instanceof Error ? error.message : String(error)));
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
if (payload.type) this.events.running(payload.type);
|
||||||
|
}
|
||||||
|
|
||||||
|
private async finish() {
|
||||||
|
if (this.settled || this.finishing) return;
|
||||||
|
this.finishing = true;
|
||||||
|
this.events.settling();
|
||||||
|
const result = await this.send("get_last_assistant_text", {});
|
||||||
|
const text = typeof result === "string" ? result : result && typeof result === "object" && "text" in result ? String((result as { text: unknown }).text) : "";
|
||||||
|
this.settled = true;
|
||||||
|
this.events.completed(text, "agent_settled");
|
||||||
|
this.terminate();
|
||||||
|
}
|
||||||
|
|
||||||
|
private terminate() {
|
||||||
|
if (this.killed) return;
|
||||||
|
this.killed = true;
|
||||||
|
this.child.stdin.end();
|
||||||
|
if (this.child.killed) return;
|
||||||
|
if (process.platform !== "win32" && this.child.pid) {
|
||||||
|
try {
|
||||||
|
process.kill(-this.child.pid, "SIGTERM");
|
||||||
|
} catch {
|
||||||
|
this.child.kill("SIGTERM");
|
||||||
|
}
|
||||||
|
setTimeout(() => {
|
||||||
|
if (this.child.killed || !this.child.pid) return;
|
||||||
|
try {
|
||||||
|
process.kill(-this.child.pid, "SIGKILL");
|
||||||
|
} catch {
|
||||||
|
this.child.kill("SIGKILL");
|
||||||
|
}
|
||||||
|
}, 2_000).unref();
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
this.child.kill("SIGTERM");
|
||||||
|
}
|
||||||
|
|
||||||
|
private fail(error: string) {
|
||||||
|
if (this.settled) return;
|
||||||
|
this.settled = true;
|
||||||
|
this.events.failed(error);
|
||||||
|
}
|
||||||
|
|
||||||
|
private send(command: string, body: Record<string, unknown>): Promise<unknown> {
|
||||||
|
const id = `subagent-${++this.nextRequest}`;
|
||||||
|
return new Promise((resolve, reject) => {
|
||||||
|
this.pending.set(id, { resolve, reject, command });
|
||||||
|
this.child.stdin.write(`${JSON.stringify({ id, type: command, ...body })}\n`, (error) => {
|
||||||
|
if (!error) return;
|
||||||
|
this.pending.delete(id);
|
||||||
|
reject(error);
|
||||||
|
});
|
||||||
|
});
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
export class SubprocessRpcRunner implements ChildRunner {
|
||||||
|
async start(id: string, request: SpawnRequest, cwd: string, events: RunnerEvents): Promise<ChildHandle> {
|
||||||
|
const args = [process.argv[1], "--mode", "rpc", "--no-extensions", "--name", `subagent ${id}`, ...contextArgs(request), ...toolArgs(request), ...modelArgs(request)];
|
||||||
|
const child = spawn(process.execPath, args, {
|
||||||
|
cwd,
|
||||||
|
env: childEnvironment(),
|
||||||
|
stdio: ["pipe", "pipe", "pipe"],
|
||||||
|
detached: process.platform !== "win32",
|
||||||
|
});
|
||||||
|
const handle = new RpcChildHandle(child, events);
|
||||||
|
events.accepted();
|
||||||
|
void handle.prompt(independentPrompt(request)).catch((error) => events.failed(error instanceof Error ? error.message : String(error)));
|
||||||
|
return handle;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
function delay(ms: number): Promise<void> {
|
||||||
|
return new Promise((resolve) => setTimeout(resolve, ms));
|
||||||
|
}
|
||||||
|
|
||||||
|
function contextArgs(request: SpawnRequest): string[] {
|
||||||
|
if (request.context !== "fork" || !request.parentSessionFile) return [];
|
||||||
|
return ["--fork", request.parentSessionFile];
|
||||||
|
}
|
||||||
|
|
||||||
|
function toolArgs(request: SpawnRequest): string[] {
|
||||||
|
const activeTools = request.toolProfile?.activeTools;
|
||||||
|
if (activeTools === undefined || activeTools === null) return [];
|
||||||
|
if (activeTools.length === 0) return ["--no-tools"];
|
||||||
|
return ["--tools", activeTools.join(",")];
|
||||||
|
}
|
||||||
|
|
||||||
|
function modelArgs(request: SpawnRequest): string[] {
|
||||||
|
const args: string[] = [];
|
||||||
|
if (request.model && request.model !== "inherit") args.push("--model", request.model);
|
||||||
|
if (request.thinking) args.push("--thinking", request.thinking);
|
||||||
|
return args;
|
||||||
|
}
|
||||||
|
|
||||||
|
function childEnvironment(): NodeJS.ProcessEnv {
|
||||||
|
const env = { ...process.env };
|
||||||
|
delete env.PI_SESSION_ID;
|
||||||
|
delete env.PI_SESSION_FILE;
|
||||||
|
delete env.PI_PROVIDER;
|
||||||
|
delete env.PI_MODEL;
|
||||||
|
delete env.PI_REASONING_LEVEL;
|
||||||
|
return env;
|
||||||
|
}
|
||||||
|
|
||||||
|
function independentPrompt(request: SpawnRequest): string {
|
||||||
|
const base = request.agentBody ? `${request.agentBody}\n\n` : "";
|
||||||
|
if (request.context === "fork") {
|
||||||
|
return `${base}You are running as a delegated subagent in fork context.\nUse the inherited parent session context, then return a concise final answer for the parent agent.\n\nTask:\n${request.prompt}`;
|
||||||
|
}
|
||||||
|
return `${base}You are running as a delegated subagent in independent context.\nDo not assume access to the parent conversation transcript.\nReturn a concise final answer for the parent agent.\n\nTask:\n${request.prompt}`;
|
||||||
|
}
|
||||||
38
modules/agents/pi/extensions/subagents/status.ts
Normal file
38
modules/agents/pi/extensions/subagents/status.ts
Normal file
@@ -0,0 +1,38 @@
|
|||||||
|
import type { ChildRecord, SpawnAccepted, SubagentResult, SubagentStatus } from "./types.ts";
|
||||||
|
|
||||||
|
export function toAccepted(status: SubagentStatus): SpawnAccepted {
|
||||||
|
return {
|
||||||
|
id: status.id,
|
||||||
|
label: status.label,
|
||||||
|
context: status.context,
|
||||||
|
tools: status.tools,
|
||||||
|
state: status.state,
|
||||||
|
hint: `Use subagent_status or subagent_result with id ${status.id}`,
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
export function cloneStatus(status: SubagentStatus): SubagentStatus {
|
||||||
|
return { ...status, elapsedMs: elapsedMs(status) };
|
||||||
|
}
|
||||||
|
|
||||||
|
export function cloneResult(record: ChildRecord): SubagentResult {
|
||||||
|
const status = cloneStatus(record.status);
|
||||||
|
const terminal = ["completed", "failed", "cancelled", "timed_out", "orphaned"].includes(status.state);
|
||||||
|
return {
|
||||||
|
id: status.id,
|
||||||
|
state: status.state,
|
||||||
|
running: !terminal,
|
||||||
|
resultAvailable: status.resultAvailable,
|
||||||
|
result: record.result,
|
||||||
|
error: status.error,
|
||||||
|
completedAt: status.completedAt,
|
||||||
|
elapsedMs: status.elapsedMs,
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
export function elapsedMs(status: Pick<SubagentStatus, "startedAt" | "completedAt">): number {
|
||||||
|
const start = Date.parse(status.startedAt);
|
||||||
|
const end = status.completedAt ? Date.parse(status.completedAt) : Date.now();
|
||||||
|
if (!Number.isFinite(start) || !Number.isFinite(end)) return 0;
|
||||||
|
return Math.max(0, end - start);
|
||||||
|
}
|
||||||
306
modules/agents/pi/extensions/subagents/supervisor.test.ts
Normal file
306
modules/agents/pi/extensions/subagents/supervisor.test.ts
Normal file
@@ -0,0 +1,306 @@
|
|||||||
|
import assert from "node:assert/strict";
|
||||||
|
import test from "node:test";
|
||||||
|
import { Supervisor } from "./supervisor.ts";
|
||||||
|
import type { ChildHandle, ChildRunner, RunnerEvents, SpawnRequest } from "./types.ts";
|
||||||
|
|
||||||
|
class FakeHandle implements ChildHandle {
|
||||||
|
cancelCalls = 0;
|
||||||
|
|
||||||
|
async cancel(): Promise<void> {
|
||||||
|
this.cancelCalls += 1;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
class FakeRunner implements ChildRunner {
|
||||||
|
starts: Array<{ id: string; request: SpawnRequest; events: RunnerEvents; handle: FakeHandle }> = [];
|
||||||
|
autoAccept = true;
|
||||||
|
|
||||||
|
async start(id: string, request: SpawnRequest, _cwd: string, events: RunnerEvents): Promise<ChildHandle> {
|
||||||
|
const handle = new FakeHandle();
|
||||||
|
this.starts.push({ id, request, events, handle });
|
||||||
|
if (this.autoAccept) events.accepted(`session-${id}`);
|
||||||
|
return handle;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));
|
||||||
|
|
||||||
|
async function spawnStarted(supervisor: Supervisor, prompt = "work") {
|
||||||
|
const accepted = supervisor.spawn({ prompt });
|
||||||
|
await sleep(0);
|
||||||
|
return accepted;
|
||||||
|
}
|
||||||
|
|
||||||
|
test("cancel is idempotent and reaches cancelled", async () => {
|
||||||
|
const runner = new FakeRunner();
|
||||||
|
const supervisor = new Supervisor(runner, "/tmp");
|
||||||
|
const accepted = await spawnStarted(supervisor);
|
||||||
|
|
||||||
|
const first = await supervisor.cancel(accepted.id);
|
||||||
|
const second = await supervisor.cancel(accepted.id);
|
||||||
|
|
||||||
|
assert.equal(first.state, "cancelled");
|
||||||
|
assert.equal(second.state, "cancelled");
|
||||||
|
assert.equal(runner.starts[0].handle.cancelCalls, 1);
|
||||||
|
});
|
||||||
|
|
||||||
|
test("startup timeout reaches timed_out", async () => {
|
||||||
|
const runner = new FakeRunner();
|
||||||
|
runner.autoAccept = false;
|
||||||
|
const supervisor = new Supervisor(runner, "/tmp", { timeouts: { startMs: 5 } });
|
||||||
|
const accepted = await spawnStarted(supervisor);
|
||||||
|
|
||||||
|
await sleep(20);
|
||||||
|
|
||||||
|
const status = supervisor.status(accepted.id);
|
||||||
|
assert.equal(status.state, "timed_out");
|
||||||
|
assert.equal(status.stopReason, "start_timeout");
|
||||||
|
assert.equal(runner.starts[0].handle.cancelCalls, 1);
|
||||||
|
});
|
||||||
|
|
||||||
|
test("runtime timeout reaches timed_out", async () => {
|
||||||
|
const runner = new FakeRunner();
|
||||||
|
const supervisor = new Supervisor(runner, "/tmp", { timeouts: { runMs: 5 } });
|
||||||
|
const accepted = await spawnStarted(supervisor);
|
||||||
|
|
||||||
|
await sleep(20);
|
||||||
|
|
||||||
|
const status = supervisor.status(accepted.id);
|
||||||
|
assert.equal(status.state, "timed_out");
|
||||||
|
assert.equal(status.stopReason, "run_timeout");
|
||||||
|
assert.equal(runner.starts[0].handle.cancelCalls, 1);
|
||||||
|
});
|
||||||
|
|
||||||
|
test("process failure reaches failed with diagnostics", async () => {
|
||||||
|
const runner = new FakeRunner();
|
||||||
|
const supervisor = new Supervisor(runner, "/tmp");
|
||||||
|
const accepted = await spawnStarted(supervisor);
|
||||||
|
|
||||||
|
runner.starts[0].events.failed("process closed with code 1");
|
||||||
|
|
||||||
|
const status = supervisor.status(accepted.id);
|
||||||
|
assert.equal(status.state, "failed");
|
||||||
|
assert.equal(status.error, "process closed with code 1");
|
||||||
|
});
|
||||||
|
|
||||||
|
test("shutdown cancels running children", async () => {
|
||||||
|
const runner = new FakeRunner();
|
||||||
|
const supervisor = new Supervisor(runner, "/tmp");
|
||||||
|
const accepted = await spawnStarted(supervisor);
|
||||||
|
|
||||||
|
await supervisor.shutdown();
|
||||||
|
|
||||||
|
const status = supervisor.status(accepted.id);
|
||||||
|
assert.equal(status.state, "cancelled");
|
||||||
|
assert.equal(status.stopReason, "shutdown");
|
||||||
|
assert.equal(runner.starts[0].handle.cancelCalls, 1);
|
||||||
|
});
|
||||||
|
|
||||||
|
test("completed children ignore later cancel", async () => {
|
||||||
|
const runner = new FakeRunner();
|
||||||
|
const supervisor = new Supervisor(runner, "/tmp");
|
||||||
|
const accepted = await spawnStarted(supervisor);
|
||||||
|
|
||||||
|
runner.starts[0].events.completed("done", "agent_settled");
|
||||||
|
await supervisor.cancel(accepted.id);
|
||||||
|
|
||||||
|
const result = supervisor.result(accepted.id);
|
||||||
|
assert.equal(result.state, "completed");
|
||||||
|
assert.equal(result.result, "done");
|
||||||
|
assert.equal(runner.starts[0].handle.cancelCalls, 0);
|
||||||
|
});
|
||||||
|
|
||||||
|
test("shutdown clears recent terminal expiry timer", async () => {
|
||||||
|
const runner = new FakeRunner();
|
||||||
|
let changes = 0;
|
||||||
|
const supervisor = new Supervisor(runner, "/tmp", {
|
||||||
|
recentTerminalTtlMs: 5,
|
||||||
|
onChange: () => {
|
||||||
|
changes += 1;
|
||||||
|
},
|
||||||
|
});
|
||||||
|
await spawnStarted(supervisor);
|
||||||
|
|
||||||
|
await supervisor.shutdown();
|
||||||
|
const afterShutdown = changes;
|
||||||
|
await sleep(15);
|
||||||
|
|
||||||
|
assert.equal(changes, afterShutdown);
|
||||||
|
});
|
||||||
|
|
||||||
|
test("batch spawn returns accepted ids and per-entry failures", async () => {
|
||||||
|
const runner = new FakeRunner();
|
||||||
|
const supervisor = new Supervisor(runner, "/tmp");
|
||||||
|
|
||||||
|
const result = supervisor.spawnBatch([{ prompt: "one" }, { prompt: "" }, { prompt: "two" }]);
|
||||||
|
await sleep(0);
|
||||||
|
|
||||||
|
assert.equal(result.accepted.length, 2);
|
||||||
|
assert.equal(result.failed.length, 1);
|
||||||
|
assert.equal(result.failed[0].index, 1);
|
||||||
|
assert.equal(runner.starts.length, 2);
|
||||||
|
});
|
||||||
|
|
||||||
|
test("maxConcurrent preserves queued records", async () => {
|
||||||
|
const runner = new FakeRunner();
|
||||||
|
const supervisor = new Supervisor(runner, "/tmp", { maxConcurrent: 1 });
|
||||||
|
|
||||||
|
const result = supervisor.spawnBatch([{ prompt: "one" }, { prompt: "two" }]);
|
||||||
|
await sleep(0);
|
||||||
|
|
||||||
|
assert.equal(result.accepted.length, 2);
|
||||||
|
assert.equal(runner.starts.length, 1);
|
||||||
|
assert.equal(supervisor.status(result.accepted[1].id).state, "queued");
|
||||||
|
|
||||||
|
runner.starts[0].events.completed("done", "agent_settled");
|
||||||
|
await sleep(0);
|
||||||
|
|
||||||
|
assert.equal(runner.starts.length, 2);
|
||||||
|
});
|
||||||
|
|
||||||
|
test("recent terminal statuses expire from list by ttl", 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);
|
||||||
|
|
||||||
|
await sleep(10);
|
||||||
|
|
||||||
|
assert.equal(supervisor.list().some((status) => status.id === accepted.id), false);
|
||||||
|
assert.equal(supervisor.result(accepted.id).result, "done");
|
||||||
|
});
|
||||||
|
|
||||||
|
test("recent terminal ttl does not hide active statuses", async () => {
|
||||||
|
const runner = new FakeRunner();
|
||||||
|
const supervisor = new Supervisor(runner, "/tmp", { recentTerminalTtlMs: 0 });
|
||||||
|
const accepted = await spawnStarted(supervisor);
|
||||||
|
|
||||||
|
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");
|
||||||
|
const first = await spawnStarted(supervisor, "one");
|
||||||
|
const second = await spawnStarted(supervisor, "two");
|
||||||
|
|
||||||
|
const waiting = supervisor.wait([first.id, second.id], { timeoutMs: 100 });
|
||||||
|
runner.starts[0].events.completed("one done", "agent_settled");
|
||||||
|
await sleep(0);
|
||||||
|
|
||||||
|
assert.equal(await Promise.race([waiting.then(() => "done"), sleep(10).then(() => "pending")]), "pending");
|
||||||
|
|
||||||
|
runner.starts[1].events.failed("two failed");
|
||||||
|
const result = await waiting;
|
||||||
|
|
||||||
|
assert.equal(result.timedOut, false);
|
||||||
|
assert.equal(result.ready, true);
|
||||||
|
assert.deepEqual(result.ids, [first.id, second.id]);
|
||||||
|
assert.equal(result.pending.length, 0);
|
||||||
|
assert.deepEqual(result.results.map((item) => item.state), ["completed", "failed"]);
|
||||||
|
assert.equal(result.results[0].result, "one done");
|
||||||
|
assert.equal(result.results[1].error, "two failed");
|
||||||
|
});
|
||||||
|
|
||||||
|
test("wait returns pending statuses on timeout", async () => {
|
||||||
|
const runner = new FakeRunner();
|
||||||
|
const supervisor = new Supervisor(runner, "/tmp");
|
||||||
|
const first = await spawnStarted(supervisor, "one");
|
||||||
|
const second = await spawnStarted(supervisor, "two");
|
||||||
|
|
||||||
|
runner.starts[0].events.completed("one done", "agent_settled");
|
||||||
|
const result = await supervisor.wait([first.id, second.id], { timeoutMs: 5 });
|
||||||
|
|
||||||
|
assert.equal(result.timedOut, true);
|
||||||
|
assert.equal(result.ready, false);
|
||||||
|
assert.deepEqual(result.results.map((item) => item.state), ["completed", "running"]);
|
||||||
|
assert.deepEqual(result.pending.map((item) => item.id), [second.id]);
|
||||||
|
});
|
||||||
|
|
||||||
|
test("wait any returns after the first terminal subagent", async () => {
|
||||||
|
const runner = new FakeRunner();
|
||||||
|
const supervisor = new Supervisor(runner, "/tmp");
|
||||||
|
const first = await spawnStarted(supervisor, "one");
|
||||||
|
const second = await spawnStarted(supervisor, "two");
|
||||||
|
|
||||||
|
const waiting = supervisor.wait([first.id, second.id], { mode: "any", timeoutMs: 100 });
|
||||||
|
runner.starts[1].events.completed("two done", "agent_settled");
|
||||||
|
const result = await waiting;
|
||||||
|
|
||||||
|
assert.equal(result.timedOut, false);
|
||||||
|
assert.equal(result.ready, true);
|
||||||
|
assert.deepEqual(result.results.map((item) => item.state), ["running", "completed"]);
|
||||||
|
assert.deepEqual(result.pending.map((item) => item.id), [first.id]);
|
||||||
|
});
|
||||||
|
|
||||||
|
test("wait rejects unknown and empty id sets", async () => {
|
||||||
|
const runner = new FakeRunner();
|
||||||
|
const supervisor = new Supervisor(runner, "/tmp");
|
||||||
|
|
||||||
|
await assert.rejects(() => supervisor.wait([]), /at least one subagent id is required/);
|
||||||
|
await assert.rejects(() => supervisor.wait(["missing"]), /unknown subagent id: missing/);
|
||||||
|
});
|
||||||
|
|
||||||
|
test("wait abort rejects without cancelling child", async () => {
|
||||||
|
const runner = new FakeRunner();
|
||||||
|
const supervisor = new Supervisor(runner, "/tmp");
|
||||||
|
const accepted = await spawnStarted(supervisor, "one");
|
||||||
|
const controller = new AbortController();
|
||||||
|
|
||||||
|
const waiting = supervisor.wait([accepted.id], { signal: controller.signal });
|
||||||
|
controller.abort();
|
||||||
|
|
||||||
|
await assert.rejects(waiting, /subagent wait aborted/);
|
||||||
|
assert.equal(runner.starts[0].handle.cancelCalls, 0);
|
||||||
|
});
|
||||||
|
|
||||||
|
test("wait follows queued subagents through queue start and completion", async () => {
|
||||||
|
const runner = new FakeRunner();
|
||||||
|
const supervisor = new Supervisor(runner, "/tmp", { maxConcurrent: 1 });
|
||||||
|
const batch = supervisor.spawnBatch([{ prompt: "one" }, { prompt: "two" }]);
|
||||||
|
await sleep(0);
|
||||||
|
|
||||||
|
const waiting = supervisor.wait([batch.accepted[1].id], { timeoutMs: 100 });
|
||||||
|
assert.equal(await Promise.race([waiting.then(() => "done"), sleep(10).then(() => "pending")]), "pending");
|
||||||
|
|
||||||
|
runner.starts[0].events.completed("one done", "agent_settled");
|
||||||
|
await sleep(0);
|
||||||
|
runner.starts[1].events.completed("two done", "agent_settled");
|
||||||
|
const result = await waiting;
|
||||||
|
|
||||||
|
assert.equal(result.timedOut, false);
|
||||||
|
assert.equal(result.ready, true);
|
||||||
|
assert.deepEqual(result.results.map((item) => item.result), ["two done"]);
|
||||||
|
});
|
||||||
423
modules/agents/pi/extensions/subagents/supervisor.ts
Normal file
423
modules/agents/pi/extensions/subagents/supervisor.ts
Normal file
@@ -0,0 +1,423 @@
|
|||||||
|
import type {
|
||||||
|
ChildHandle,
|
||||||
|
ChildRecord,
|
||||||
|
ChildRunner,
|
||||||
|
ContextMode,
|
||||||
|
RunnerEvents,
|
||||||
|
SpawnAccepted,
|
||||||
|
SpawnRequest,
|
||||||
|
SubagentResult,
|
||||||
|
SubagentStatus,
|
||||||
|
SubagentWaitMode,
|
||||||
|
SubagentWaitResult,
|
||||||
|
} from "./types.ts";
|
||||||
|
import { cloneResult, cloneStatus, toAccepted } from "./status.ts";
|
||||||
|
|
||||||
|
interface RunningChild {
|
||||||
|
record: ChildRecord;
|
||||||
|
request: SpawnRequest;
|
||||||
|
handle?: ChildHandle;
|
||||||
|
startTimer?: ReturnType<typeof setTimeout>;
|
||||||
|
runTimer?: ReturnType<typeof setTimeout>;
|
||||||
|
}
|
||||||
|
|
||||||
|
interface SupervisorOptions {
|
||||||
|
maxConcurrent?: number;
|
||||||
|
recentTerminalLimit?: number;
|
||||||
|
recentTerminalTtlMs?: number;
|
||||||
|
timeouts?: {
|
||||||
|
startMs?: number;
|
||||||
|
runMs?: number;
|
||||||
|
};
|
||||||
|
onMilestone?: (status: SubagentStatus, event: string) => void;
|
||||||
|
onChange?: (statuses: SubagentStatus[]) => void;
|
||||||
|
}
|
||||||
|
|
||||||
|
export interface BatchSpawnResult {
|
||||||
|
accepted: SpawnAccepted[];
|
||||||
|
failed: Array<{ index: number; error: string }>;
|
||||||
|
}
|
||||||
|
|
||||||
|
const DEFAULT_TIMEOUTS = {
|
||||||
|
startMs: 30_000,
|
||||||
|
runMs: 0,
|
||||||
|
};
|
||||||
|
|
||||||
|
export class Supervisor {
|
||||||
|
private nextChild = 0;
|
||||||
|
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,
|
||||||
|
private readonly cwd: string,
|
||||||
|
private readonly options: SupervisorOptions = {},
|
||||||
|
) {}
|
||||||
|
|
||||||
|
spawn(request: SpawnRequest): SpawnAccepted {
|
||||||
|
return this.createChild(request);
|
||||||
|
}
|
||||||
|
|
||||||
|
spawnBatch(requests: SpawnRequest[]): BatchSpawnResult {
|
||||||
|
const accepted: SpawnAccepted[] = [];
|
||||||
|
const failed: Array<{ index: number; error: string }> = [];
|
||||||
|
requests.forEach((request, index) => {
|
||||||
|
try {
|
||||||
|
accepted.push(this.createChild(request));
|
||||||
|
} catch (error) {
|
||||||
|
failed.push({ index, error: error instanceof Error ? error.message : String(error) });
|
||||||
|
}
|
||||||
|
});
|
||||||
|
return { accepted, failed };
|
||||||
|
}
|
||||||
|
|
||||||
|
list(): SubagentStatus[] {
|
||||||
|
const statuses = [...this.children.values()].map((child) => cloneStatus(child.record.status));
|
||||||
|
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);
|
||||||
|
return [...active, ...terminal];
|
||||||
|
}
|
||||||
|
|
||||||
|
status(id: string): SubagentStatus {
|
||||||
|
return cloneStatus(this.require(id).record.status);
|
||||||
|
}
|
||||||
|
|
||||||
|
result(id: string): SubagentResult {
|
||||||
|
return cloneResult(this.require(id).record);
|
||||||
|
}
|
||||||
|
|
||||||
|
async wait(
|
||||||
|
ids: string[],
|
||||||
|
options: { timeoutMs?: number; signal?: AbortSignal; mode?: SubagentWaitMode } = {},
|
||||||
|
): Promise<SubagentWaitResult> {
|
||||||
|
const uniqueIds = [...new Set(ids.map((id) => id.trim()).filter(Boolean))];
|
||||||
|
if (uniqueIds.length === 0) throw new Error("at least one subagent id is required");
|
||||||
|
for (const id of uniqueIds) this.require(id);
|
||||||
|
|
||||||
|
const startedAt = Date.now();
|
||||||
|
const mode = options.mode ?? "all";
|
||||||
|
if (mode !== "all" && mode !== "any") throw new Error(`unknown wait mode: ${mode}`);
|
||||||
|
const deadline = options.timeoutMs && options.timeoutMs > 0 ? startedAt + options.timeoutMs : undefined;
|
||||||
|
let timedOut = false;
|
||||||
|
|
||||||
|
while (!this.waitReady(uniqueIds, mode)) {
|
||||||
|
if (options.signal?.aborted) throw new Error("subagent wait aborted");
|
||||||
|
const remainingMs = deadline === undefined ? undefined : deadline - Date.now();
|
||||||
|
if (remainingMs !== undefined && remainingMs <= 0) {
|
||||||
|
timedOut = true;
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
await this.nextChange(remainingMs, options.signal).catch((error) => {
|
||||||
|
if (error instanceof Error && error.message === "subagent wait timed out") timedOut = true;
|
||||||
|
else throw error;
|
||||||
|
});
|
||||||
|
if (timedOut) break;
|
||||||
|
}
|
||||||
|
|
||||||
|
const results = uniqueIds.map((id) => this.result(id));
|
||||||
|
const pending = uniqueIds
|
||||||
|
.map((id) => this.status(id))
|
||||||
|
.filter((status) => !isTerminal(status.state));
|
||||||
|
return { ids: uniqueIds, mode, ready: this.waitReady(uniqueIds, mode), results, pending, timedOut, elapsedMs: Date.now() - startedAt };
|
||||||
|
}
|
||||||
|
|
||||||
|
async cancel(id: string): Promise<SubagentStatus> {
|
||||||
|
const child = this.require(id);
|
||||||
|
if (isTerminal(child.record.status.state)) return cloneStatus(child.record.status);
|
||||||
|
await child.handle?.cancel();
|
||||||
|
this.completeWithoutResult(child, "cancelled", "cancelled");
|
||||||
|
this.pumpQueue();
|
||||||
|
return cloneStatus(child.record.status);
|
||||||
|
}
|
||||||
|
|
||||||
|
async shutdown(): Promise<void> {
|
||||||
|
this.clearRecentTerminalTimer();
|
||||||
|
await Promise.allSettled(
|
||||||
|
[...this.children.values()].map(async (child) => {
|
||||||
|
if (!isTerminal(child.record.status.state)) {
|
||||||
|
await child.handle?.cancel();
|
||||||
|
this.completeWithoutResult(child, "cancelled", "shutdown");
|
||||||
|
}
|
||||||
|
}),
|
||||||
|
);
|
||||||
|
this.clearRecentTerminalTimer();
|
||||||
|
}
|
||||||
|
|
||||||
|
private createChild(request: SpawnRequest): SpawnAccepted {
|
||||||
|
const prompt = typeof request.prompt === "string" ? request.prompt.trim() : "";
|
||||||
|
if (!prompt) throw new Error("prompt is required");
|
||||||
|
|
||||||
|
const id = this.allocateId();
|
||||||
|
const now = new Date().toISOString();
|
||||||
|
const status: SubagentStatus = {
|
||||||
|
id,
|
||||||
|
label: request.agent ?? `ad-hoc ${id}`,
|
||||||
|
agent: request.agent,
|
||||||
|
adHoc: !request.agent,
|
||||||
|
context: this.resolveContext(request.context),
|
||||||
|
state: "queued",
|
||||||
|
cwd: this.cwd,
|
||||||
|
model: request.model,
|
||||||
|
thinking: request.thinking,
|
||||||
|
tools: request.tools ?? "read-only",
|
||||||
|
startedAt: now,
|
||||||
|
elapsedMs: 0,
|
||||||
|
lastEvent: "queued",
|
||||||
|
lastEventAt: now,
|
||||||
|
resultAvailable: false,
|
||||||
|
};
|
||||||
|
const child: RunningChild = { record: { status }, request: { ...request, prompt, context: status.context, tools: status.tools } };
|
||||||
|
this.children.set(id, child);
|
||||||
|
this.emitMilestone(child, "accepted");
|
||||||
|
this.queue.push(child);
|
||||||
|
this.pumpQueue();
|
||||||
|
return toAccepted(cloneStatus(status));
|
||||||
|
}
|
||||||
|
|
||||||
|
private pumpQueue() {
|
||||||
|
while (this.runningCount() < this.maxConcurrent()) {
|
||||||
|
const child = this.queue.shift();
|
||||||
|
if (!child) break;
|
||||||
|
if (isTerminal(child.record.status.state)) continue;
|
||||||
|
this.start(child);
|
||||||
|
}
|
||||||
|
this.emitChange();
|
||||||
|
}
|
||||||
|
|
||||||
|
private start(child: RunningChild) {
|
||||||
|
this.setState(child.record.status, "starting", "starting");
|
||||||
|
this.armStartTimer(child);
|
||||||
|
setTimeout(() => {
|
||||||
|
if (isTerminal(child.record.status.state)) return;
|
||||||
|
void this.runner
|
||||||
|
.start(child.record.status.id, child.request, this.cwd, this.eventsFor(child.record))
|
||||||
|
.then((handle) => {
|
||||||
|
child.handle = handle;
|
||||||
|
if (isTerminal(child.record.status.state)) void handle.cancel();
|
||||||
|
})
|
||||||
|
.catch((error) => {
|
||||||
|
this.fail(child.record, error instanceof Error ? error.message : String(error));
|
||||||
|
});
|
||||||
|
}, 0);
|
||||||
|
}
|
||||||
|
|
||||||
|
private eventsFor(record: ChildRecord): RunnerEvents {
|
||||||
|
return {
|
||||||
|
accepted: (childSession) => {
|
||||||
|
const child = this.findChild(record);
|
||||||
|
if (child) {
|
||||||
|
this.clearTimer(child, "startTimer");
|
||||||
|
this.armRunTimer(child);
|
||||||
|
}
|
||||||
|
if (childSession) record.status.childSession = childSession;
|
||||||
|
this.setState(record.status, "running", "prompt accepted");
|
||||||
|
},
|
||||||
|
running: (event) => {
|
||||||
|
if (!isTerminal(record.status.state)) this.setState(record.status, "running", event);
|
||||||
|
},
|
||||||
|
settling: () => {
|
||||||
|
if (!isTerminal(record.status.state)) this.setState(record.status, "settling", "agent_settled");
|
||||||
|
},
|
||||||
|
completed: (result, stopReason) => {
|
||||||
|
const now = new Date().toISOString();
|
||||||
|
const child = this.findChild(record);
|
||||||
|
if (child) this.clearTimers(child);
|
||||||
|
record.result = result;
|
||||||
|
record.status.state = "completed";
|
||||||
|
record.status.completedAt = now;
|
||||||
|
record.status.lastEvent = "completed";
|
||||||
|
record.status.lastEventAt = now;
|
||||||
|
record.status.stopReason = stopReason;
|
||||||
|
record.status.resultAvailable = true;
|
||||||
|
if (child) this.emitMilestone(child, "completed");
|
||||||
|
this.pumpQueue();
|
||||||
|
},
|
||||||
|
failed: (error) => this.fail(record, error),
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
private fail(record: ChildRecord, error: string) {
|
||||||
|
if (isTerminal(record.status.state)) return;
|
||||||
|
const child = this.findChild(record);
|
||||||
|
if (child) this.clearTimers(child);
|
||||||
|
const now = new Date().toISOString();
|
||||||
|
record.status.state = "failed";
|
||||||
|
record.status.completedAt = now;
|
||||||
|
record.status.lastEvent = "failed";
|
||||||
|
record.status.lastEventAt = now;
|
||||||
|
record.status.error = error;
|
||||||
|
record.status.stopReason = "failed";
|
||||||
|
if (child) this.emitMilestone(child, "failed");
|
||||||
|
this.pumpQueue();
|
||||||
|
}
|
||||||
|
|
||||||
|
private completeWithoutResult(child: RunningChild, state: "cancelled" | "timed_out", reason: string) {
|
||||||
|
if (isTerminal(child.record.status.state)) return;
|
||||||
|
this.clearTimers(child);
|
||||||
|
const now = new Date().toISOString();
|
||||||
|
child.record.status.state = state;
|
||||||
|
child.record.status.completedAt = now;
|
||||||
|
child.record.status.lastEvent = state;
|
||||||
|
child.record.status.lastEventAt = now;
|
||||||
|
child.record.status.stopReason = reason;
|
||||||
|
this.emitMilestone(child, state);
|
||||||
|
}
|
||||||
|
|
||||||
|
private armStartTimer(child: RunningChild) {
|
||||||
|
const timeout = this.options.timeouts?.startMs ?? DEFAULT_TIMEOUTS.startMs;
|
||||||
|
if (timeout <= 0) return;
|
||||||
|
child.startTimer = setTimeout(() => {
|
||||||
|
this.timeout(child, "start_timeout");
|
||||||
|
}, timeout);
|
||||||
|
}
|
||||||
|
|
||||||
|
private armRunTimer(child: RunningChild) {
|
||||||
|
const timeout = this.options.timeouts?.runMs ?? DEFAULT_TIMEOUTS.runMs;
|
||||||
|
if (timeout <= 0) return;
|
||||||
|
child.runTimer = setTimeout(() => {
|
||||||
|
this.timeout(child, "run_timeout");
|
||||||
|
}, timeout);
|
||||||
|
}
|
||||||
|
|
||||||
|
private timeout(child: RunningChild, reason: string) {
|
||||||
|
if (isTerminal(child.record.status.state)) return;
|
||||||
|
void child.handle?.cancel();
|
||||||
|
this.completeWithoutResult(child, "timed_out", reason);
|
||||||
|
this.pumpQueue();
|
||||||
|
}
|
||||||
|
|
||||||
|
private clearTimers(child: RunningChild) {
|
||||||
|
this.clearTimer(child, "startTimer");
|
||||||
|
this.clearTimer(child, "runTimer");
|
||||||
|
}
|
||||||
|
|
||||||
|
private clearTimer(child: RunningChild, key: "startTimer" | "runTimer") {
|
||||||
|
const timer = child[key];
|
||||||
|
if (!timer) return;
|
||||||
|
clearTimeout(timer);
|
||||||
|
child[key] = undefined;
|
||||||
|
}
|
||||||
|
|
||||||
|
private findChild(record: ChildRecord): RunningChild | undefined {
|
||||||
|
return [...this.children.values()].find((child) => child.record === record);
|
||||||
|
}
|
||||||
|
|
||||||
|
private setState(status: SubagentStatus, state: SubagentStatus["state"], event: string) {
|
||||||
|
if (isTerminal(status.state)) return;
|
||||||
|
const now = new Date().toISOString();
|
||||||
|
status.state = state;
|
||||||
|
status.lastEvent = event;
|
||||||
|
status.lastEventAt = now;
|
||||||
|
this.emitChange();
|
||||||
|
}
|
||||||
|
|
||||||
|
private require(id: string): RunningChild {
|
||||||
|
const child = this.children.get(id);
|
||||||
|
if (!child) throw new Error(`unknown subagent id: ${id}`);
|
||||||
|
return child;
|
||||||
|
}
|
||||||
|
|
||||||
|
private resolveContext(context: ContextMode | undefined): ContextMode {
|
||||||
|
if (context === undefined) return "independent";
|
||||||
|
if (context !== "independent" && context !== "fork") throw new Error(`unknown context: ${context}`);
|
||||||
|
return context;
|
||||||
|
}
|
||||||
|
|
||||||
|
private maxConcurrent(): number {
|
||||||
|
return Math.max(1, this.options.maxConcurrent ?? 3);
|
||||||
|
}
|
||||||
|
|
||||||
|
private runningCount(): number {
|
||||||
|
return [...this.children.values()].filter((child) => ["starting", "running", "settling"].includes(child.record.status.state)).length;
|
||||||
|
}
|
||||||
|
|
||||||
|
private emitMilestone(child: RunningChild, event: string) {
|
||||||
|
this.options.onMilestone?.(cloneStatus(child.record.status), event);
|
||||||
|
this.emitChange();
|
||||||
|
}
|
||||||
|
|
||||||
|
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 {
|
||||||
|
const terminal = (id: string) => isTerminal(this.require(id).record.status.state);
|
||||||
|
return mode === "all" ? ids.every(terminal) : ids.some(terminal);
|
||||||
|
}
|
||||||
|
|
||||||
|
private nextChange(timeoutMs: number | undefined, signal: AbortSignal | undefined): Promise<void> {
|
||||||
|
return new Promise((resolve, reject) => {
|
||||||
|
let timer: ReturnType<typeof setTimeout> | undefined;
|
||||||
|
const cleanup = () => {
|
||||||
|
this.waiters.delete(resolveOnce);
|
||||||
|
if (timer) clearTimeout(timer);
|
||||||
|
signal?.removeEventListener("abort", abort);
|
||||||
|
};
|
||||||
|
const resolveOnce = () => {
|
||||||
|
cleanup();
|
||||||
|
resolve();
|
||||||
|
};
|
||||||
|
const abort = () => {
|
||||||
|
cleanup();
|
||||||
|
reject(new Error("subagent wait aborted"));
|
||||||
|
};
|
||||||
|
this.waiters.add(resolveOnce);
|
||||||
|
signal?.addEventListener("abort", abort, { once: true });
|
||||||
|
if (timeoutMs !== undefined) {
|
||||||
|
timer = setTimeout(() => {
|
||||||
|
cleanup();
|
||||||
|
reject(new Error("subagent wait timed out"));
|
||||||
|
}, timeoutMs);
|
||||||
|
}
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
private allocateId(): string {
|
||||||
|
this.nextChild += 1;
|
||||||
|
return `sg-${Date.now().toString(36)}-${this.nextChild.toString(36)}`;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
function isTerminal(state: SubagentStatus["state"]): boolean {
|
||||||
|
return ["completed", "failed", "cancelled", "timed_out", "orphaned"].includes(state);
|
||||||
|
}
|
||||||
103
modules/agents/pi/extensions/subagents/types.ts
Normal file
103
modules/agents/pi/extensions/subagents/types.ts
Normal file
@@ -0,0 +1,103 @@
|
|||||||
|
export type ContextMode = "independent" | "fork";
|
||||||
|
|
||||||
|
export type SubagentState =
|
||||||
|
| "queued"
|
||||||
|
| "starting"
|
||||||
|
| "running"
|
||||||
|
| "settling"
|
||||||
|
| "completed"
|
||||||
|
| "failed"
|
||||||
|
| "cancelled"
|
||||||
|
| "timed_out"
|
||||||
|
| "orphaned";
|
||||||
|
|
||||||
|
export interface ToolProfile {
|
||||||
|
activeTools: string[] | null;
|
||||||
|
}
|
||||||
|
|
||||||
|
export interface SpawnRequest {
|
||||||
|
prompt: string;
|
||||||
|
context?: ContextMode;
|
||||||
|
agent?: string;
|
||||||
|
model?: string;
|
||||||
|
thinking?: string;
|
||||||
|
tools?: string;
|
||||||
|
toolProfile?: ToolProfile;
|
||||||
|
agentBody?: string;
|
||||||
|
parentSessionFile?: string;
|
||||||
|
}
|
||||||
|
|
||||||
|
export interface SpawnAccepted {
|
||||||
|
id: string;
|
||||||
|
label: string;
|
||||||
|
context: ContextMode;
|
||||||
|
tools: string;
|
||||||
|
state: SubagentState;
|
||||||
|
hint: string;
|
||||||
|
}
|
||||||
|
|
||||||
|
export interface SubagentStatus {
|
||||||
|
id: string;
|
||||||
|
label: string;
|
||||||
|
agent?: string;
|
||||||
|
adHoc: boolean;
|
||||||
|
context: ContextMode;
|
||||||
|
state: SubagentState;
|
||||||
|
cwd: string;
|
||||||
|
model?: string;
|
||||||
|
thinking?: string;
|
||||||
|
tools: string;
|
||||||
|
startedAt: string;
|
||||||
|
completedAt?: string;
|
||||||
|
elapsedMs: number;
|
||||||
|
lastEvent?: string;
|
||||||
|
lastEventAt?: string;
|
||||||
|
stopReason?: string;
|
||||||
|
resultAvailable: boolean;
|
||||||
|
childSession?: string;
|
||||||
|
error?: string;
|
||||||
|
}
|
||||||
|
|
||||||
|
export interface SubagentResult {
|
||||||
|
id: string;
|
||||||
|
state: SubagentState;
|
||||||
|
running: boolean;
|
||||||
|
resultAvailable: boolean;
|
||||||
|
result?: string;
|
||||||
|
error?: string;
|
||||||
|
completedAt?: string;
|
||||||
|
elapsedMs: number;
|
||||||
|
}
|
||||||
|
|
||||||
|
export type SubagentWaitMode = "all" | "any";
|
||||||
|
|
||||||
|
export interface SubagentWaitResult {
|
||||||
|
ids: string[];
|
||||||
|
mode: SubagentWaitMode;
|
||||||
|
ready: boolean;
|
||||||
|
results: SubagentResult[];
|
||||||
|
pending: SubagentStatus[];
|
||||||
|
timedOut: boolean;
|
||||||
|
elapsedMs: number;
|
||||||
|
}
|
||||||
|
|
||||||
|
export interface ChildRecord {
|
||||||
|
status: SubagentStatus;
|
||||||
|
result?: string;
|
||||||
|
}
|
||||||
|
|
||||||
|
export interface RunnerEvents {
|
||||||
|
accepted(childSession?: string): void;
|
||||||
|
running(event: string): void;
|
||||||
|
settling(): void;
|
||||||
|
completed(result: string, stopReason?: string): void;
|
||||||
|
failed(error: string): void;
|
||||||
|
}
|
||||||
|
|
||||||
|
export interface ChildHandle {
|
||||||
|
cancel(): Promise<void>;
|
||||||
|
}
|
||||||
|
|
||||||
|
export interface ChildRunner {
|
||||||
|
start(id: string, request: SpawnRequest, cwd: string, events: RunnerEvents): Promise<ChildHandle>;
|
||||||
|
}
|
||||||
28
modules/agents/pi/extensions/subagents/ui.ts
Normal file
28
modules/agents/pi/extensions/subagents/ui.ts
Normal file
@@ -0,0 +1,28 @@
|
|||||||
|
import type { SubagentStatus } from "./types.ts";
|
||||||
|
|
||||||
|
export function renderSummary(statuses: SubagentStatus[]): string[] {
|
||||||
|
const running = statuses.filter((status) => ["starting", "running", "settling"].includes(status.state)).length;
|
||||||
|
const queued = statuses.filter((status) => status.state === "queued").length;
|
||||||
|
const terminal = statuses.filter((status) => ["completed", "failed", "cancelled", "timed_out", "orphaned"].includes(status.state)).length;
|
||||||
|
if (running === 0 && queued === 0 && terminal === 0) return [];
|
||||||
|
return [`subagents: ${running} running · ${queued} queued · ${terminal} recent`];
|
||||||
|
}
|
||||||
|
|
||||||
|
export function renderInspector(statuses: SubagentStatus[]): string[] {
|
||||||
|
const lines = renderSummary(statuses);
|
||||||
|
for (const status of statuses) {
|
||||||
|
lines.push(
|
||||||
|
`${status.id} ${status.label} ${status.context} ${status.state} ${Math.round(status.elapsedMs / 1000)}s ${status.model ?? "inherit"} ${status.tools} ${status.lastEvent ?? "none"} result:${status.resultAvailable ? "yes" : "no"}`,
|
||||||
|
);
|
||||||
|
}
|
||||||
|
return lines;
|
||||||
|
}
|
||||||
|
|
||||||
|
export function widget(statuses: SubagentStatus[], expanded: boolean) {
|
||||||
|
return () => ({
|
||||||
|
invalidate() {},
|
||||||
|
render(width: number) {
|
||||||
|
return (expanded ? renderInspector(statuses) : renderSummary(statuses)).map((line) => (line.length > width ? line.slice(0, Math.max(0, width - 1)) : line));
|
||||||
|
},
|
||||||
|
});
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user