Files
oh-my-pi/packages/coding-agent/src/task/index.ts
T
can1357 779ca4872b style(deps): migrated from Prettier to Biome and updated formatting rules
- Removed Prettier configuration files (.prettierignore and .prettierrc) and migrated formatting to Biome.
- Updated Biome configuration from version 2.3.11 to 2.3.12 and changed arrowParentheses rule from 'always' to 'asNeeded'.
- Pinned @biomejs/biome dependency to exact version 2.3.12 in package.json and bun.lock.
- Applied consistent arrow function formatting across 489 files by removing unnecessary parentheses around single parameters.
- Removed blank lines after comment blocks and reorganized imports for consistency across the codebase.
2026-01-24 04:20:19 +01:00

680 lines
21 KiB
TypeScript

/**
* Task tool - Delegate tasks to specialized agents.
*
* Discovers agent definitions from:
* - Bundled agents (shipped with omp-coding-agent)
* - ~/.omp/agent/agents/*.md (user-level)
* - .omp/agents/*.md (project-level)
*
* Supports:
* - Single agent execution
* - Parallel execution with concurrency limits
* - Progress tracking via JSON events
* - Session artifacts for debugging
*/
import * as fs from "node:fs/promises";
import * as os from "node:os";
import path from "node:path";
import type { AgentTool, AgentToolResult, AgentToolUpdateCallback } from "@oh-my-pi/pi-agent-core";
import type { Usage } from "@oh-my-pi/pi-ai";
import { $ } from "bun";
import { nanoid } from "nanoid";
import type { ToolSession } from "..";
import { renderPromptTemplate } from "../config/prompt-templates";
import type { Theme } from "../modes/theme/theme";
import taskDescriptionTemplate from "../prompts/tools/task.md" with { type: "text" };
import { formatDuration } from "../tools/render-utils";
// Import review tools for side effects (registers subagent tool handlers)
import "../tools/review";
import { discoverAgents, getAgent } from "./discovery";
import { runSubprocess } from "./executor";
import { AgentOutputManager } from "./output-manager";
import { mapWithConcurrencyLimit } from "./parallel";
import { renderCall, renderResult } from "./render";
import { renderTemplate } from "./template";
import {
type AgentProgress,
MAX_CONCURRENCY,
MAX_PARALLEL_TASKS,
type SingleResult,
type TaskParams,
type TaskToolDetails,
taskSchema,
} from "./types";
import {
applyBaseline,
captureBaseline,
captureDeltaPatch,
cleanupWorktree,
ensureWorktree,
getRepoRoot,
} from "./worktree";
/** Format byte count for display */
function formatBytes(bytes: number): string {
if (bytes < 1024) return `${bytes}B`;
if (bytes < 1024 * 1024) return `${(bytes / 1024).toFixed(1)}K`;
return `${(bytes / (1024 * 1024)).toFixed(1)}M`;
}
function createUsageTotals(): Usage {
return {
input: 0,
output: 0,
cacheRead: 0,
cacheWrite: 0,
totalTokens: 0,
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
};
}
function addUsageTotals(target: Usage, usage: Partial<Usage>): void {
const input = usage.input ?? 0;
const output = usage.output ?? 0;
const cacheRead = usage.cacheRead ?? 0;
const cacheWrite = usage.cacheWrite ?? 0;
const totalTokens = usage.totalTokens ?? input + output + cacheRead + cacheWrite;
const cost =
usage.cost ??
({
input: 0,
output: 0,
cacheRead: 0,
cacheWrite: 0,
total: 0,
} satisfies Usage["cost"]);
target.input += input;
target.output += output;
target.cacheRead += cacheRead;
target.cacheWrite += cacheWrite;
target.totalTokens += totalTokens;
target.cost.input += cost.input;
target.cost.output += cost.output;
target.cost.cacheRead += cost.cacheRead;
target.cost.cacheWrite += cost.cacheWrite;
target.cost.total += cost.total;
}
// Re-export types and utilities
export { loadBundledAgents as BUNDLED_AGENTS } from "./agents";
export { discoverCommands, expandCommand, getCommand } from "./commands";
export { discoverAgents, getAgent } from "./discovery";
export { AgentOutputManager } from "./output-manager";
export type { AgentDefinition, AgentProgress, SingleResult, TaskParams, TaskToolDetails } from "./types";
export { taskSchema } from "./types";
/**
* Build dynamic tool description listing available agents.
*/
async function buildDescription(cwd: string): Promise<string> {
const { agents } = await discoverAgents(cwd);
return renderPromptTemplate(taskDescriptionTemplate, {
agents,
MAX_PARALLEL_TASKS,
MAX_CONCURRENCY,
});
}
// ═══════════════════════════════════════════════════════════════════════════
// Tool Class
// ═══════════════════════════════════════════════════════════════════════════
/**
* Task tool - Delegate tasks to specialized agents.
*
* Requires async initialization to discover available agents.
* Use `TaskTool.create(session)` to instantiate.
*/
export class TaskTool implements AgentTool<typeof taskSchema, TaskToolDetails, Theme> {
public readonly name = "task";
public readonly label = "Task";
public readonly description: string;
public readonly parameters = taskSchema;
public readonly renderCall = renderCall;
public readonly renderResult = renderResult;
private readonly session: ToolSession;
private readonly blockedAgent: string | undefined;
private constructor(session: ToolSession, description: string) {
this.session = session;
this.description = description;
this.blockedAgent = process.env.OMP_BLOCKED_AGENT;
}
/**
* Create a TaskTool instance with async agent discovery.
*/
public static async create(session: ToolSession): Promise<TaskTool> {
const description = await buildDescription(session.cwd);
return new TaskTool(session, description);
}
public async execute(
_toolCallId: string,
params: TaskParams,
signal?: AbortSignal,
onUpdate?: AgentToolUpdateCallback<TaskToolDetails>,
): Promise<AgentToolResult<TaskToolDetails>> {
const startTime = Date.now();
const { agents, projectAgentsDir } = await discoverAgents(this.session.cwd);
const { agent: agentName, context, output: outputSchema, isolated } = params;
const isIsolated = isolated === true;
const isDefaultModelAlias = (value: string | undefined): boolean => {
if (!value) return true;
const normalized = value.trim().toLowerCase();
return normalized === "default" || normalized === "pi/default" || normalized === "omp/default";
};
// Validate agent exists
const agent = getAgent(agents, agentName);
if (!agent) {
const available = agents.map(a => a.name).join(", ") || "none";
return {
content: [
{
type: "text",
text: `Unknown agent "${agentName}". Available: ${available}`,
},
],
details: {
projectAgentsDir,
results: [],
totalDurationMs: 0,
},
};
}
const effectiveAgentModel = isDefaultModelAlias(agent.model) ? undefined : agent.model;
const modelOverride =
effectiveAgentModel ?? this.session.getActiveModelString?.() ?? this.session.getModelString?.();
const thinkingLevelOverride = agent.thinkingLevel;
// Output schema priority: agent frontmatter > params > inherited from parent session
const schemaOverridden = outputSchema !== undefined && agent.output !== undefined;
const effectiveOutputSchema = agent.output ?? outputSchema ?? this.session.outputSchema;
// Handle empty or missing tasks
if (!params.tasks || params.tasks.length === 0) {
return {
content: [
{
type: "text",
text: `No tasks provided. Use: { agent, context, tasks: [{id, description, args}, ...] }`,
},
],
details: {
projectAgentsDir,
results: [],
totalDurationMs: 0,
},
};
}
// Validate task count
if (params.tasks.length > MAX_PARALLEL_TASKS) {
return {
content: [
{
type: "text",
text: `Too many tasks (${params.tasks.length}). Max is ${MAX_PARALLEL_TASKS}.`,
},
],
details: {
projectAgentsDir,
results: [],
totalDurationMs: 0,
},
};
}
const tasks = params.tasks;
const missingTaskIndexes: number[] = [];
const idIndexes = new Map<string, number[]>();
for (let i = 0; i < tasks.length; i++) {
const id = tasks[i]?.id;
if (typeof id !== "string" || id.trim() === "") {
missingTaskIndexes.push(i);
continue;
}
const normalizedId = id.toLowerCase();
const indexes = idIndexes.get(normalizedId);
if (indexes) {
indexes.push(i);
} else {
idIndexes.set(normalizedId, [i]);
}
}
const duplicateIds: Array<{ id: string; indexes: number[] }> = [];
for (const [normalizedId, indexes] of idIndexes.entries()) {
if (indexes.length > 1) {
duplicateIds.push({
id: tasks[indexes[0]]?.id ?? normalizedId,
indexes,
});
}
}
if (missingTaskIndexes.length > 0 || duplicateIds.length > 0) {
const problems: string[] = [];
if (missingTaskIndexes.length > 0) {
problems.push(`Missing task ids at indexes: ${missingTaskIndexes.join(", ")}`);
}
if (duplicateIds.length > 0) {
const details = duplicateIds.map(entry => `${entry.id} (indexes ${entry.indexes.join(", ")})`).join("; ");
problems.push(`Duplicate task ids detected (case-insensitive): ${details}`);
}
return {
content: [{ type: "text", text: `Invalid tasks: ${problems.join(". ")}` }],
details: {
projectAgentsDir,
results: [],
totalDurationMs: 0,
},
};
}
let repoRoot: string | null = null;
let baseline = null as Awaited<ReturnType<typeof captureBaseline>> | null;
if (isIsolated) {
try {
repoRoot = await getRepoRoot(this.session.cwd);
baseline = await captureBaseline(repoRoot);
} catch (err) {
const message = err instanceof Error ? err.message : String(err);
return {
content: [
{
type: "text",
text: `Isolated task execution requires a git repository. ${message}`,
},
],
details: {
projectAgentsDir,
results: [],
totalDurationMs: Date.now() - startTime,
},
};
}
}
// Derive artifacts directory
const sessionFile = this.session.getSessionFile();
const artifactsDir = sessionFile ? sessionFile.slice(0, -6) : null;
const tempArtifactsDir = artifactsDir ? null : path.join(os.tmpdir(), `omp-task-${nanoid()}`);
const effectiveArtifactsDir = artifactsDir || tempArtifactsDir!;
// Initialize progress tracking
const progressMap = new Map<number, AgentProgress>();
// Update callback
const emitProgress = () => {
const progress = Array.from(progressMap.values()).sort((a, b) => a.index - b.index);
onUpdate?.({
content: [{ type: "text", text: `Running ${params.tasks.length} agents...` }],
details: {
projectAgentsDir,
results: [],
totalDurationMs: Date.now() - startTime,
progress,
},
});
};
try {
// Check self-recursion prevention
if (this.blockedAgent && agentName === this.blockedAgent) {
return {
content: [
{
type: "text",
text: `Cannot spawn ${this.blockedAgent} agent from within itself (recursion prevention). Use a different agent type.`,
},
],
details: {
projectAgentsDir,
results: [],
totalDurationMs: Date.now() - startTime,
},
};
}
// Check spawn restrictions from parent
const parentSpawns = this.session.getSessionSpawns() ?? "*";
const allowedSpawns = parentSpawns.split(",").map(s => s.trim());
const isSpawnAllowed = (): boolean => {
if (parentSpawns === "") return false; // Empty = deny all
if (parentSpawns === "*") return true; // Wildcard = allow all
return allowedSpawns.includes(agentName);
};
if (!isSpawnAllowed()) {
const allowed = parentSpawns === "" ? "none (spawns disabled for this agent)" : parentSpawns;
return {
content: [{ type: "text", text: `Cannot spawn '${agentName}'. Allowed: ${allowed}` }],
details: {
projectAgentsDir,
results: [],
totalDurationMs: Date.now() - startTime,
},
};
}
// Build full prompts with context prepended
// Allocate unique IDs across the session to prevent artifact collisions
const outputManager =
this.session.agentOutputManager ?? new AgentOutputManager(this.session.getArtifactsDir ?? (() => null));
const uniqueIds = await outputManager.allocateBatch(tasks.map(t => t.id));
const tasksWithUniqueIds = tasks.map((t, i) => ({ ...t, id: uniqueIds[i] }));
// Build full prompts with context prepended
const tasksWithContext = tasksWithUniqueIds.map(t => renderTemplate(context, t));
// Initialize progress for all tasks
for (let i = 0; i < tasksWithContext.length; i++) {
const t = tasksWithContext[i];
progressMap.set(i, {
index: i,
id: t.id,
agent: agentName,
agentSource: agent.source,
status: "pending",
task: t.task,
args: t.args,
recentTools: [],
recentOutput: [],
toolCount: 0,
tokens: 0,
durationMs: 0,
modelOverride,
description: t.description,
});
}
emitProgress();
const runTask = async (task: (typeof tasksWithContext)[number], index: number) => {
if (!isIsolated) {
return runSubprocess({
cwd: this.session.cwd,
agent,
task: task.task,
description: task.description,
index,
id: task.id,
context: undefined, // Already prepended above
modelOverride,
thinkingLevel: thinkingLevelOverride,
outputSchema: effectiveOutputSchema,
sessionFile,
persistArtifacts: !!artifactsDir,
artifactsDir: effectiveArtifactsDir,
enableLsp: false,
signal,
eventBus: undefined,
onProgress: progress => {
progressMap.set(index, {
...structuredClone(progress),
args: tasksWithContext[index]?.args,
});
emitProgress();
},
authStorage: this.session.authStorage,
modelRegistry: this.session.modelRegistry,
settingsManager: this.session.settingsManager,
mcpManager: this.session.mcpManager,
});
}
const taskStart = Date.now();
let worktreeDir: string | undefined;
try {
if (!repoRoot || !baseline) {
throw new Error("Isolated task execution not initialized.");
}
worktreeDir = await ensureWorktree(repoRoot, task.id);
await applyBaseline(worktreeDir, baseline);
const result = await runSubprocess({
cwd: this.session.cwd,
worktree: worktreeDir,
agent,
task: task.task,
description: task.description,
index,
id: task.id,
context: undefined, // Already prepended above
modelOverride,
thinkingLevel: thinkingLevelOverride,
outputSchema: effectiveOutputSchema,
sessionFile,
persistArtifacts: !!artifactsDir,
artifactsDir: effectiveArtifactsDir,
enableLsp: false,
signal,
eventBus: undefined,
onProgress: progress => {
progressMap.set(index, {
...structuredClone(progress),
args: tasksWithContext[index]?.args,
});
emitProgress();
},
authStorage: this.session.authStorage,
modelRegistry: this.session.modelRegistry,
settingsManager: this.session.settingsManager,
mcpManager: this.session.mcpManager,
});
const patch = await captureDeltaPatch(worktreeDir, baseline);
const patchPath = path.join(effectiveArtifactsDir, `${task.id}.patch`);
await Bun.write(patchPath, patch);
return {
...result,
patchPath,
};
} catch (err) {
const message = err instanceof Error ? err.message : String(err);
return {
index,
id: task.id,
agent: agent.name,
agentSource: agent.source,
task: task.task,
description: task.description,
exitCode: 1,
output: "",
stderr: message,
truncated: false,
durationMs: Date.now() - taskStart,
tokens: 0,
modelOverride,
error: message,
};
} finally {
if (worktreeDir) {
await cleanupWorktree(worktreeDir);
}
}
};
// Execute in parallel with concurrency limit
const { results: partialResults, aborted } = await mapWithConcurrencyLimit(
tasksWithContext,
MAX_CONCURRENCY,
runTask,
signal,
);
// Fill in skipped tasks (undefined entries from abort) with placeholder results
const results: SingleResult[] = partialResults.map((result, index) => {
if (result !== undefined) {
return {
...result,
args: tasksWithContext[index]?.args,
};
}
const task = tasksWithContext[index];
return {
index,
id: task.id,
agent: agentName,
agentSource: agent.source,
task: task.task,
args: task.args,
description: task.description,
exitCode: 1,
output: "",
stderr: "Skipped (cancelled before start)",
truncated: false,
durationMs: 0,
tokens: 0,
modelOverride,
error: "Skipped",
aborted: true,
};
});
// Aggregate usage from executor results (already accumulated incrementally)
const aggregatedUsage = createUsageTotals();
let hasAggregatedUsage = false;
for (const result of results) {
if (result.usage) {
addUsageTotals(aggregatedUsage, result.usage);
hasAggregatedUsage = true;
}
}
// Collect output paths (artifacts already written by executor in real-time)
const outputPaths: string[] = [];
const patchPaths: string[] = [];
for (const result of results) {
if (result.outputPath) {
outputPaths.push(result.outputPath);
}
if (result.patchPath) {
patchPaths.push(result.patchPath);
}
}
let patchApplySummary = "";
let patchesApplied: boolean | null = null;
if (isIsolated) {
const patchesInOrder = results.map(result => result.patchPath).filter(Boolean) as string[];
const missingPatch = results.some(result => !result.patchPath);
if (!repoRoot || missingPatch) {
patchesApplied = false;
} else {
const patchStats = await Promise.all(
patchesInOrder.map(async patchPath => ({
patchPath,
size: (await fs.stat(patchPath)).size,
})),
);
const nonEmptyPatches = patchStats.filter(patch => patch.size > 0).map(patch => patch.patchPath);
if (nonEmptyPatches.length === 0) {
patchesApplied = true;
} else {
const patchTexts = await Promise.all(
nonEmptyPatches.map(async patchPath => Bun.file(patchPath).text()),
);
const combinedPatch = patchTexts.map(text => (text.endsWith("\n") ? text : `${text}\n`)).join("");
if (!combinedPatch.trim()) {
patchesApplied = true;
} else {
const combinedPatchPath = path.join(os.tmpdir(), `omp-task-combined-${nanoid()}.patch`);
try {
await Bun.write(combinedPatchPath, combinedPatch);
const checkResult = await $`git apply --check --binary ${combinedPatchPath}`
.cwd(repoRoot)
.quiet()
.nothrow();
if (checkResult.exitCode !== 0) {
patchesApplied = false;
} else {
const applyResult = await $`git apply --binary ${combinedPatchPath}`
.cwd(repoRoot)
.quiet()
.nothrow();
patchesApplied = applyResult.exitCode === 0;
}
} finally {
await fs.rm(combinedPatchPath, { force: true });
}
}
}
}
if (patchesApplied) {
patchApplySummary = "\n\nApplied patches: yes";
} else {
const notification =
"<system-notification>Patches were not applied and must be handled manually.</system-notification>";
const patchList =
patchPaths.length > 0
? `\n\nPatch artifacts:\n${patchPaths.map(patch => `- ${patch}`).join("\n")}`
: "";
patchApplySummary = `\n\n${notification}${patchList}`;
}
}
// Build final output - match plugin format
const successCount = results.filter(r => r.exitCode === 0).length;
const cancelledCount = results.filter(r => r.aborted).length;
const totalDuration = Date.now() - startTime;
const summaries = results.map(r => {
const status = r.aborted ? "cancelled" : r.exitCode === 0 ? "completed" : `failed (exit ${r.exitCode})`;
const output = r.output.trim() || r.stderr.trim() || "(no output)";
const preview = output.split("\n").slice(0, 5).join("\n");
const meta = r.outputMeta
? ` [${r.outputMeta.lineCount} lines, ${formatBytes(r.outputMeta.charCount)}]`
: "";
return `[${r.agent}] ${status}${meta} ${r.id}\n${preview}`;
});
const outputIds = results.filter(r => !r.aborted || r.output.trim()).map(r => r.id);
const outputHint =
outputIds.length > 0
? `\n\nUse read with agent:// for full logs: ${outputIds.map(id => `agent://${id}`).join(", ")}`
: "";
const schemaNote = schemaOverridden
? `\n\nNote: Agent '${agentName}' has a fixed output schema; your 'output' parameter was ignored.\nRequired schema: ${JSON.stringify(agent.output)}`
: "";
const cancelledNote = aborted && cancelledCount > 0 ? ` (${cancelledCount} cancelled)` : "";
const summary = `${successCount}/${results.length} succeeded${cancelledNote} [${formatDuration(
totalDuration,
)}]\n\n${summaries.join("\n\n---\n\n")}${outputHint}${schemaNote}${patchApplySummary}`;
// Cleanup temp directory if used
const shouldCleanupTempArtifacts =
tempArtifactsDir && (!isIsolated || patchesApplied === true || patchesApplied === null);
if (shouldCleanupTempArtifacts) {
await fs.rm(tempArtifactsDir, { recursive: true, force: true });
}
return {
content: [{ type: "text", text: summary }],
details: {
projectAgentsDir,
results: results,
totalDurationMs: totalDuration,
usage: hasAggregatedUsage ? aggregatedUsage : undefined,
outputPaths,
},
};
} catch (err) {
return {
content: [{ type: "text", text: `Task execution failed: ${err}` }],
details: {
projectAgentsDir,
results: [],
totalDurationMs: Date.now() - startTime,
},
};
}
}
}