From 91da560cc79e46b42cbac0e40de23e7df3526881 Mon Sep 17 00:00:00 2001 From: can1357 Date: Tue, 27 Jan 2026 07:38:55 +0100 Subject: [PATCH] refactor(coding-agent/task): moved subagents in-process - run subagent execution in-process and drop worker protocol - align tool settings manager typing and docs - queue skill follow-ups, dedupe review findings, restore MCP timeouts --- packages/coding-agent/CHANGELOG.md | 8 + packages/coding-agent/DEVELOPMENT.md | 6 +- .../src/modes/controllers/input-controller.ts | 2 +- packages/coding-agent/src/task/executor.ts | 897 ++++++++--------- packages/coding-agent/src/task/types.ts | 2 +- .../coding-agent/src/task/worker-protocol.ts | 131 --- packages/coding-agent/src/task/worker.ts | 921 ------------------ packages/coding-agent/src/tools/index.ts | 8 +- 8 files changed, 438 insertions(+), 1537 deletions(-) delete mode 100644 packages/coding-agent/src/task/worker-protocol.ts delete mode 100644 packages/coding-agent/src/task/worker.ts diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index 61e63340e..0a4e2666a 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -7,6 +7,14 @@ - Added `--plan` CLI flag and `OMP_PLAN_MODEL` environment variable for ephemeral plan model override - Added plan model selection in model selector UI with PLAN badge +### Changed +- Task tool subagents now execute in-process instead of using worker threads + +### Fixed +- Queued skill commands as follow-ups when the agent is already streaming to avoid load failures +- Deduplicated repeated review findings in subagent progress rendering +- Restored MCP proxy tool timeout handling to prevent subagent hangs + ## [8.5.0] - 2026-01-27 ### Added diff --git a/packages/coding-agent/DEVELOPMENT.md b/packages/coding-agent/DEVELOPMENT.md index 60e22d5e1..1456963c2 100644 --- a/packages/coding-agent/DEVELOPMENT.md +++ b/packages/coding-agent/DEVELOPMENT.md @@ -256,11 +256,11 @@ src/ │ ├── agents.ts # Agent definitions │ ├── commands.ts # Task commands │ ├── discovery.ts # Task discovery -│ ├── executor.ts # Task execution +│ ├── executor.ts # In-process task execution │ ├── render.ts # Task rendering │ ├── subprocess-tool-registry.ts # Subprocess tool registry -│ ├── worker-protocol.ts # Worker protocol -│ └── worker.ts # Task worker +│ ├── parallel.ts # Task concurrency helpers +│ └── template.ts # Task templating ├── tools/ # Built-in tool implementations │ ├── index.ts # Tool exports, BUILTIN_TOOLS, createTools diff --git a/packages/coding-agent/src/modes/controllers/input-controller.ts b/packages/coding-agent/src/modes/controllers/input-controller.ts index e35925918..cd3c1dd9d 100644 --- a/packages/coding-agent/src/modes/controllers/input-controller.ts +++ b/packages/coding-agent/src/modes/controllers/input-controller.ts @@ -324,7 +324,7 @@ export class InputController { metaLines.push(`User: ${args}`); } const message = `${body}\n\n---\n\n${metaLines.join("\n")}`; - await this.ctx.session.prompt(message); + await this.ctx.session.prompt(message, { streamingBehavior: "followUp" }); } catch (err) { this.ctx.showError(`Failed to load skill: ${err instanceof Error ? err.message : String(err)}`); } diff --git a/packages/coding-agent/src/task/executor.ts b/packages/coding-agent/src/task/executor.ts index 3fd93c8ff..830c03ffb 100644 --- a/packages/coding-agent/src/task/executor.ts +++ b/packages/coding-agent/src/task/executor.ts @@ -1,24 +1,28 @@ /** - * Worker execution for subagents. + * In-process execution for subagents. * - * Runs each subagent in a Bun Worker and forwards AgentEvents for progress tracking. + * Runs each subagent on the main thread and forwards AgentEvents for progress tracking. */ import path from "node:path"; import type { AgentEvent, ThinkingLevel } from "@oh-my-pi/pi-agent-core"; -import type { ModelRegistry } from "../config/model-registry"; -import { formatModelString, parseModelPattern } from "../config/model-resolver"; -import type { PromptTemplate } from "../config/prompt-templates"; -import type { Skill } from "../extensibility/skills"; -import { getPreludeDocs } from "../ipy/executor"; -import { checkPythonKernelAvailability } from "../ipy/kernel"; -import { LspTool } from "../lsp"; -import type { LspParams } from "../lsp/types"; -import { callTool } from "../mcp/client"; -import type { MCPManager } from "../mcp/manager"; -import type { AuthStorage } from "../session/auth-storage"; -import type { ContextFileEntry, ToolSession } from "../tools"; -import { PythonTool, type PythonToolParams } from "../tools/python"; -import type { EventBus } from "../utils/event-bus"; +import type { Api, Model } from "@oh-my-pi/pi-ai"; +import type { ModelRegistry } from "@oh-my-pi/pi-coding-agent/config/model-registry"; +import { parseModelPattern } from "@oh-my-pi/pi-coding-agent/config/model-resolver"; +import type { PromptTemplate } from "@oh-my-pi/pi-coding-agent/config/prompt-templates"; +import { SettingsManager } from "@oh-my-pi/pi-coding-agent/config/settings-manager"; +import type { CustomTool } from "@oh-my-pi/pi-coding-agent/extensibility/custom-tools/types"; +import type { Skill } from "@oh-my-pi/pi-coding-agent/extensibility/skills"; +import { callTool } from "@oh-my-pi/pi-coding-agent/mcp/client"; +import type { MCPManager } from "@oh-my-pi/pi-coding-agent/mcp/manager"; +import { createAgentSession, discoverAuthStorage, discoverModels } from "@oh-my-pi/pi-coding-agent/sdk"; +import type { AgentSession, AgentSessionEvent } from "@oh-my-pi/pi-coding-agent/session/agent-session"; +import type { AuthStorage } from "@oh-my-pi/pi-coding-agent/session/auth-storage"; +import { SessionManager } from "@oh-my-pi/pi-coding-agent/session/session-manager"; +import type { ContextFileEntry } from "@oh-my-pi/pi-coding-agent/tools"; +import { ToolAbortError } from "@oh-my-pi/pi-coding-agent/tools/tool-errors"; +import type { EventBus } from "@oh-my-pi/pi-coding-agent/utils/event-bus"; +import { logger, untilAborted } from "@oh-my-pi/pi-utils"; +import type { TSchema } from "@sinclair/typebox"; import { subprocessToolRegistry } from "./subprocess-tool-registry"; import { type AgentDefinition, @@ -30,17 +34,26 @@ import { TASK_SUBAGENT_EVENT_CHANNEL, TASK_SUBAGENT_PROGRESS_CHANNEL, } from "./types"; -import type { - LspToolCallRequest, - MCPToolCallRequest, - MCPToolMetadata, - PythonToolCallCancel, - PythonToolCallRequest, - SubagentWorkerRequest, - SubagentWorkerResponse, -} from "./worker-protocol"; const DEFAULT_MODEL_ALIASES = new Set(["default", "pi/default", "omp/default"]); +const MCP_CALL_TIMEOUT_MS = 60_000; + +/** Agent event types to forward for progress tracking. */ +const agentEventTypes = new Set([ + "agent_start", + "agent_end", + "turn_start", + "turn_end", + "message_start", + "message_update", + "message_end", + "tool_execution_start", + "tool_execution_update", + "tool_execution_end", +]); + +const isAgentEvent = (event: AgentSessionEvent): event is AgentEvent => + agentEventTypes.has(event.type as AgentEvent["type"]); function normalizeModelPatterns(value: string | string[] | undefined): string[] { if (!value) return []; @@ -53,7 +66,93 @@ function normalizeModelPatterns(value: string | string[] | undefined): string[] .filter(Boolean); } -/** Options for worker execution */ +function withAbortTimeout(promise: Promise, timeoutMs: number, signal?: AbortSignal): Promise { + return new Promise((resolve, reject) => { + let settled = false; + const timeoutId = setTimeout(() => { + if (settled) return; + settled = true; + reject(new Error(`MCP tool call timed out after ${timeoutMs}ms`)); + }, timeoutMs); + + const onAbort = () => { + if (settled) return; + settled = true; + clearTimeout(timeoutId); + reject(new ToolAbortError()); + }; + + if (signal) { + if (signal.aborted) { + clearTimeout(timeoutId); + reject(new ToolAbortError()); + return; + } + signal.addEventListener("abort", onAbort, { once: true }); + } + + promise.then( + value => { + if (settled) return; + settled = true; + clearTimeout(timeoutId); + if (signal) signal.removeEventListener("abort", onAbort); + resolve(value); + }, + error => { + if (settled) return; + settled = true; + clearTimeout(timeoutId); + if (signal) signal.removeEventListener("abort", onAbort); + reject(error); + }, + ); + }); +} + +function getReportFindingKey(value: unknown): string | null { + if (!value || typeof value !== "object") return null; + const record = value as Record; + const title = typeof record.title === "string" ? record.title : null; + const filePath = typeof record.file_path === "string" ? record.file_path : null; + const lineStart = typeof record.line_start === "number" ? record.line_start : null; + const lineEnd = typeof record.line_end === "number" ? record.line_end : null; + const priority = typeof record.priority === "string" ? record.priority : null; + if (!title || !filePath || lineStart === null || lineEnd === null) { + return null; + } + return `${filePath}:${lineStart}:${lineEnd}:${priority ?? ""}:${title}`; +} + +function resolveModelOverride( + modelPatterns: string[], + modelRegistry: ModelRegistry, + settingsManager?: SettingsManager, +): { model?: Model; thinkingLevel?: ThinkingLevel } { + if (modelPatterns.length === 0) return {}; + const roles = settingsManager?.serialize().modelRoles as Record | undefined; + for (const pattern of modelPatterns) { + const normalized = pattern.trim().toLowerCase(); + if (!normalized || DEFAULT_MODEL_ALIASES.has(normalized)) { + continue; + } + let effectivePattern = pattern; + if (normalized.startsWith("omp/") || normalized.startsWith("pi/")) { + const role = normalized.startsWith("omp/") ? pattern.slice(4) : pattern.slice(3); + const configured = roles?.[role] ?? roles?.[role.toLowerCase()]; + if (configured) { + effectivePattern = configured; + } + } + const { model, thinkingLevel } = parseModelPattern(effectivePattern, modelRegistry.getAvailable()); + if (model) { + return { model, thinkingLevel: thinkingLevel !== "off" ? thinkingLevel : undefined }; + } + } + return {}; +} + +/** Options for subagent execution */ export interface ExecutorOptions { cwd: string; worktree?: string; @@ -80,13 +179,7 @@ export interface ExecutorOptions { mcpManager?: MCPManager; authStorage?: AuthStorage; modelRegistry?: ModelRegistry; - settingsManager?: { - serialize: () => import("../config/settings-manager").Settings; - getPlansDirectory: (cwd?: string) => string; - getPythonToolMode?: () => "ipy-only" | "bash-only" | "both"; - getPythonKernelMode?: () => "session" | "per-call"; - getPythonSharedGateway?: () => boolean; - }; + settingsManager?: SettingsManager; } /** @@ -184,29 +277,60 @@ function getUsageTokens(usage: unknown): number { } /** - * Extract MCP tool metadata from MCPManager for passing to worker. - * - * MCPTool and DeferredMCPTool expose mcpToolName (original MCP tool name) - * and mcpServerName properties. We use these directly when available, - * falling back to empty strings if not. + * Create proxy tools that reuse the parent's MCP connections. */ -function extractMCPToolMetadata(mcpManager: MCPManager): MCPToolMetadata[] { +function createMCPProxyTools(mcpManager: MCPManager): CustomTool[] { return mcpManager.getTools().map(tool => { - // MCPTool and DeferredMCPTool have these properties const mcpTool = tool as { mcpToolName?: string; mcpServerName?: string }; return { name: tool.name, label: tool.label ?? tool.name, description: tool.description ?? "", - parameters: tool.parameters, - serverName: mcpTool.mcpServerName ?? "", - mcpToolName: mcpTool.mcpToolName ?? "", + parameters: tool.parameters as TSchema, + execute: async (_toolCallId, params, _onUpdate, _ctx, signal) => { + if (signal?.aborted) { + throw new ToolAbortError(); + } + const serverName = mcpTool.mcpServerName ?? ""; + const mcpToolName = mcpTool.mcpToolName ?? ""; + try { + const result = await withAbortTimeout( + (async () => { + const connection = await mcpManager.waitForConnection(serverName); + return callTool(connection, mcpToolName, params as Record); + })(), + MCP_CALL_TIMEOUT_MS, + signal, + ); + return { + content: (result.content ?? []).map(item => + item.type === "text" + ? { type: "text" as const, text: item.text ?? "" } + : { type: "text" as const, text: JSON.stringify(item) }, + ), + details: { serverName, mcpToolName, isError: result.isError }, + }; + } catch (error) { + if (error instanceof ToolAbortError) { + throw error; + } + return { + content: [ + { + type: "text" as const, + text: `MCP error: ${error instanceof Error ? error.message : String(error)}`, + }, + ], + details: { serverName, mcpToolName, isError: true }, + }; + } + }, }; }); } /** - * Run a single agent in a worker. + * Run a single agent in-process. */ export async function runSubprocess(options: ExecutorOptions): Promise { const { @@ -282,7 +406,8 @@ export async function runSubprocess(options: ExecutorOptions): Promise name !== "exec"); if (pythonToolMode === "bash-only") { @@ -295,70 +420,12 @@ export async function runSubprocess(options: ExecutorOptions): Promise 0) { - const roles = serializedSettings?.modelRoles as Record | undefined; - for (const pattern of modelPatterns) { - const normalized = pattern.trim().toLowerCase(); - if (!normalized || DEFAULT_MODEL_ALIASES.has(normalized)) { - continue; - } - let effectivePattern = pattern; - if (normalized.startsWith("omp/") || normalized.startsWith("pi/")) { - const role = normalized.startsWith("omp/") ? pattern.slice(4) : pattern.slice(3); - const configured = roles?.[role] ?? roles?.[role.toLowerCase()]; - if (configured) { - effectivePattern = configured; - } - } - const { model } = parseModelPattern(effectivePattern, availableModels); - if (model) { - resolvedModel = formatModelString(model); - break; - } - } - } const sessionFile = subtaskSessionFile ?? null; const spawnsEnv = agent.spawns === undefined ? "" : agent.spawns === "*" ? "*" : agent.spawns.join(","); - const pythonToolRequested = toolNames === undefined || toolNames.includes("python"); - let pythonProxyEnabled = pythonToolRequested && pythonToolMode !== "bash-only"; - if (pythonProxyEnabled) { - const availability = await checkPythonKernelAvailability(cwd); - pythonProxyEnabled = availability.ok; - } - const lspEnabled = enableLsp ?? true; - const lspToolRequested = lspEnabled && (toolNames === undefined || toolNames.includes("lsp")); - const pythonPreludeDocs = getPreludeDocs(); - const pythonPreludeDocsPayload = pythonPreludeDocs.length > 0 ? pythonPreludeDocs : undefined; - - let worker: Worker; - try { - worker = new Worker(new URL("./worker.ts", import.meta.url), { type: "module" }); - } catch (err) { - return { - index, - id, - agent: agent.name, - agentSource: agent.source, - task, - description: options.description, - exitCode: 1, - output: "", - stderr: `Failed to create worker: ${err instanceof Error ? err.message : String(err)}`, - truncated: false, - durationMs: Date.now() - startTime, - tokens: 0, - modelOverride, - error: `Failed to create worker: ${err instanceof Error ? err.message : String(err)}`, - }; - } + const skipPythonPreflight = Array.isArray(toolNames) && !toolNames.includes("python"); const outputChunks: string[] = []; const finalOutputChunks: string[] = []; @@ -369,67 +436,14 @@ export async function runSubprocess(options: ExecutorOptions): Promise | null = null; let pendingTerminationTimeoutId: ReturnType | null = null; - let finalize: ((message: Extract) => void) | null = null; const listenerController = new AbortController(); const listenerSignal = listenerController.signal; - const withTimeout = async (promise: Promise, timeoutMs?: number): Promise => { - if (timeoutMs === undefined) return promise; - let timeoutId: ReturnType | undefined; - try { - return await Promise.race([ - promise, - new Promise((_resolve, reject) => { - timeoutId = setTimeout(() => { - reject(new Error(`Tool call timed out after ${timeoutMs}ms`)); - }, timeoutMs); - }), - ]); - } finally { - if (timeoutId) clearTimeout(timeoutId); - } - }; - - const combineSignals = (signals: Array): AbortSignal | undefined => { - const filtered = signals.filter((value): value is AbortSignal => Boolean(value)); - if (filtered.length === 0) return undefined; - if (filtered.length === 1) return filtered[0]; - return AbortSignal.any(filtered); - }; - - const createTimeoutSignal = (timeoutMs?: number): AbortSignal | undefined => { - if (typeof timeoutMs !== "number" || !Number.isFinite(timeoutMs) || timeoutMs <= 0) { - return undefined; - } - return AbortSignal.timeout(timeoutMs); - }; - - const pythonSessionFile = sessionFile ?? `subtask:${id}`; - const pythonToolSession: ToolSession = { - cwd, - hasUI: false, - enableLsp: false, - getSessionFile: () => pythonSessionFile, - getSessionSpawns: () => spawnsEnv, - settings: options.settingsManager as ToolSession["settings"], - settingsManager: options.settingsManager, - }; - const pythonTool = pythonProxyEnabled ? new PythonTool(pythonToolSession) : null; - const pythonCallControllers = new Map(); - - const lspToolSession: ToolSession = { - cwd, - hasUI: false, - enableLsp: lspEnabled, - getSessionFile: () => pythonSessionFile, - getSessionSpawns: () => spawnsEnv, - settings: options.settingsManager as ToolSession["settings"], - settingsManager: options.settingsManager, - }; - const lspTool = lspToolRequested ? new LspTool(lspToolSession) : null; + const abortController = new AbortController(); + const abortSignal = abortController.signal; + let activeSession: AgentSession | null = null; + let unsubscribe: (() => void) | null = null; + let completeCalled = false; // Accumulate usage incrementally from message_end events (no memory for streaming events) const accumulatedUsage = { @@ -442,30 +456,6 @@ export async function runSubprocess(options: ExecutorOptions): Promise { - if (terminationScheduled) return; - terminationScheduled = true; - terminationTimeoutId = setTimeout(() => { - terminationTimeoutId = null; - if (resolved || terminated) return; - terminated = true; - try { - worker.terminate(); - } catch { - // Ignore termination errors - } - if (finalize && !resolved) { - finalize({ - type: "done", - exitCode: 1, - durationMs: Date.now() - startTime, - error: abortReason === "signal" ? "Aborted" : "Worker terminated after tool completion", - aborted: abortReason === "signal", - }); - } - }, 2000); - }; - const requestAbort = (reason: AbortReason) => { if (abortSent) { if (reason === "signal" && abortReason !== "signal") { @@ -476,23 +466,15 @@ export async function runSubprocess(options: ExecutorOptions): Promise { - if (pendingTerminationTimeoutId || abortSent || terminationScheduled || resolved) return; + if (pendingTerminationTimeoutId || abortSent || resolved) return; pendingTerminationTimeoutId = setTimeout(() => { pendingTerminationTimeoutId = null; if (!resolved) { @@ -672,12 +654,23 @@ export async function runSubprocess(options: ExecutorOptions): Promise getReportFindingKey(item) === findingKey); + if (existingIndex >= 0) { + existing[existingIndex] = data; + } else { + existing.push(data); + } + } else { + existing.push(data); + } + progress.extractedToolData[event.toolName] = existing; } } - // Check if handler wants to terminate worker + // Check if handler wants to terminate the session if ( handler.shouldTerminate?.({ toolName: event.toolName, @@ -785,288 +778,242 @@ export async function runSubprocess(options: ExecutorOptions): Promise => { + const sessionAbortController = new AbortController(); + let exitCode = 0; + let error: string | undefined; + let aborted = false; + + const checkAbort = () => { + if (abortSignal.aborted) { + aborted = abortReason === "signal" || abortReason === undefined; + exitCode = 1; + throw new ToolAbortError(); + } + }; + + try { + checkAbort(); + const authStorage = options.authStorage ?? (await discoverAuthStorage()); + checkAbort(); + const modelRegistry = options.modelRegistry ?? discoverModels(authStorage); + checkAbort(); + + const { model, thinkingLevel: resolvedThinkingLevel } = resolveModelOverride( + modelPatterns, + modelRegistry, + settingsManager, + ); + const effectiveThinkingLevel = thinkingLevel ?? resolvedThinkingLevel; + + const sessionManager = sessionFile + ? await SessionManager.open(sessionFile) + : SessionManager.inMemory(worktree ?? cwd); + + const mcpProxyTools = options.mcpManager ? createMCPProxyTools(options.mcpManager) : []; + const enableMCP = !options.mcpManager; + + const completionInstruction = + "When finished, call the complete tool exactly once. Do not end with a plain-text final answer."; + const worktreeNotice = worktree + ? `You will work under this working tree: ${worktree}. CRITICAL: Do not touch the original repository; only make changes inside this worktree.` + : ""; + + const { session } = await createAgentSession({ + cwd: worktree ?? cwd, + authStorage, + modelRegistry, + settingsManager, + model, + thinkingLevel: effectiveThinkingLevel, + toolNames, + outputSchema, + requireCompleteTool: true, + contextFiles: options.contextFiles, + skills: options.skills, + preloadedSkills: options.preloadedSkills, + promptTemplates: options.promptTemplates, + systemPrompt: defaultPrompt => + `${defaultPrompt}\n\n${agent.systemPrompt}\n\n${worktreeNotice}\n\n${completionInstruction}`, + sessionManager, + hasUI: false, + spawns: spawnsEnv, + enableLsp: lspEnabled, + skipPythonPreflight, + enableMCP, + customTools: mcpProxyTools.length > 0 ? mcpProxyTools : undefined, + }); + + activeSession = session; + + session.sessionManager.appendSessionInit({ + systemPrompt: session.agent.state.systemPrompt, + task: fullTask, + tools: session.getAllToolNames(), + outputSchema, + }); + + abortSignal.addEventListener( + "abort", + () => { + void session.abort(); + }, + { once: true, signal: sessionAbortController.signal }, + ); + + const extensionRunner = session.extensionRunner; + if (extensionRunner) { + extensionRunner.initialize( + { + sendMessage: (message, options) => { + session.sendCustomMessage(message, options).catch(e => { + logger.error("Extension sendMessage failed", { + error: e instanceof Error ? e.message : String(e), + }); + }); + }, + sendUserMessage: (content, options) => { + session.sendUserMessage(content, options).catch(e => { + logger.error("Extension sendUserMessage failed", { + error: e instanceof Error ? e.message : String(e), + }); + }); + }, + appendEntry: (customType, data) => { + session.sessionManager.appendCustomEntry(customType, data); + }, + setLabel: (targetId, label) => { + session.sessionManager.appendLabelChange(targetId, label); + }, + getActiveTools: () => session.getActiveToolNames(), + getAllTools: () => session.getAllToolNames(), + setActiveTools: (toolNames: string[]) => session.setActiveToolsByName(toolNames), + setModel: async model => { + const key = await session.modelRegistry.getApiKey(model); + if (!key) return false; + await session.setModel(model); + return true; + }, + getThinkingLevel: () => session.thinkingLevel, + setThinkingLevel: level => session.setThinkingLevel(level), + }, + { + getModel: () => session.model, + isIdle: () => !session.isStreaming, + abort: () => session.abort(), + hasPendingMessages: () => session.queuedMessageCount > 0, + shutdown: () => {}, + getContextUsage: () => session.getContextUsage(), + compact: async instructionsOrOptions => { + const instructions = typeof instructionsOrOptions === "string" ? instructionsOrOptions : undefined; + const options = + instructionsOrOptions && typeof instructionsOrOptions === "object" + ? instructionsOrOptions + : undefined; + await session.compact(instructions, options); + }, + }, + ); + extensionRunner.onError(err => { + logger.error("Extension error", { path: err.extensionPath, error: err.error }); + }); + await extensionRunner.emit({ type: "session_start" }); + } + + const MAX_COMPLETE_RETRIES = 3; + unsubscribe = session.subscribe(event => { + if (event.type === "tool_execution_end" && event.toolName === "complete") { + completeCalled = true; + } + if (isAgentEvent(event)) { + try { + processEvent(event); + } catch (err) { + logger.error("Subagent event processing failed", { + error: err instanceof Error ? err.message : String(err), + }); + requestAbort("terminate"); + } + } + }); + + await session.prompt(fullTask); + + let retryCount = 0; + while (!completeCalled && retryCount < MAX_COMPLETE_RETRIES && !abortSignal.aborted) { + retryCount++; + const reminder = ` +CRITICAL: You stopped without calling the complete tool. This is reminder ${retryCount} of ${MAX_COMPLETE_RETRIES}. + +You MUST call the complete tool to finish your task. Options: +1. Call complete with your result data if you have completed the task +2. Call complete with status="aborted" and an error message if you cannot complete the task + +Failure to call complete after ${MAX_COMPLETE_RETRIES} reminders will result in task failure. + + +Call complete now.`; + + await session.prompt(reminder); + } + + const lastMessage = session.state.messages[session.state.messages.length - 1]; + if (lastMessage?.role === "assistant" && lastMessage.stopReason === "aborted") { + aborted = abortReason === "signal" || abortReason === undefined; + exitCode = 1; + } + } catch (err) { + exitCode = 1; + if (!abortSignal.aborted) { + error = err instanceof Error ? err.stack || err.message : String(err); + } + } finally { + if (abortSignal.aborted) { + aborted = abortReason === "signal" || abortReason === undefined; + if (exitCode === 0) exitCode = 1; + } + sessionAbortController.abort(); + if (unsubscribe) { + try { + unsubscribe(); + } catch { + // Ignore unsubscribe errors + } + unsubscribe = null; + } + if (activeSession) { + const session = activeSession; + activeSession = null; + try { + await untilAborted(AbortSignal.timeout(5000), () => session.dispose()); + } catch { + // Ignore cleanup errors + } + } + } + + return { + exitCode, + error, + aborted, + durationMs: Date.now() - startTime, + }; }; - interface WorkerMessageEvent { - data: T; - } - interface WorkerErrorEvent { - message: string; - } + const done = await runSubagent(); + resolved = true; + listenerController.abort(); - const done = await new Promise>(resolve => { - const cleanup = () => { - listenerController.abort(); - }; - finalize = message => { - if (resolved) return; - resolved = true; - cleanup(); - resolve(message); - }; - const postMessageSafe = (message: unknown) => { - if (resolved || terminated) return; - try { - worker.postMessage(message); - } catch { - // Worker already terminated - } - }; - const handleMCPCall = async (request: MCPToolCallRequest) => { - const mcpManager = options.mcpManager; - if (!mcpManager) { - postMessageSafe({ - type: "mcp_tool_result", - callId: request.callId, - error: "MCP not available", - }); - return; - } - try { - const result = await withTimeout( - (async () => { - const connection = await mcpManager.waitForConnection(request.serverName); - return callTool(connection, request.mcpToolName, request.params); - })(), - request.timeoutMs, - ); - postMessageSafe({ - type: "mcp_tool_result", - callId: request.callId, - result: { content: result.content ?? [], isError: result.isError }, - }); - } catch (error) { - postMessageSafe({ - type: "mcp_tool_result", - callId: request.callId, - error: error instanceof Error ? error.message : String(error), - }); - } - }; - - const getPythonCallTimeoutMs = (params: { timeout?: number }): number | undefined => { - const timeout = params.timeout; - if (typeof timeout === "number" && Number.isFinite(timeout) && timeout > 0) { - return Math.max(1000, Math.round(timeout * 1000) + 1000); - } - return undefined; - }; - - const handlePythonCall = async (request: PythonToolCallRequest) => { - if (!pythonTool) { - postMessageSafe({ - type: "python_tool_result", - callId: request.callId, - error: "Python proxy not available", - }); - return; - } - const callController = new AbortController(); - pythonCallControllers.set(request.callId, callController); - const timeoutMs = getPythonCallTimeoutMs(request.params as { timeout?: number }); - const timeoutSignal = createTimeoutSignal(timeoutMs); - const combinedSignal = combineSignals([signal, callController.signal, timeoutSignal]); - try { - const result = await pythonTool.execute(request.callId, request.params as PythonToolParams, combinedSignal); - postMessageSafe({ - type: "python_tool_result", - callId: request.callId, - result: { content: result.content ?? [], details: result.details }, - }); - } catch (error) { - const message = - timeoutSignal?.aborted && timeoutMs !== undefined - ? `Python tool call timed out after ${timeoutMs}ms` - : error instanceof Error - ? error.message - : String(error); - postMessageSafe({ - type: "python_tool_result", - callId: request.callId, - error: message, - }); - } finally { - pythonCallControllers.delete(request.callId); - } - }; - - const handlePythonCancel = (request: PythonToolCallCancel) => { - const controller = pythonCallControllers.get(request.callId); - if (controller) { - controller.abort(); - } - }; - - const handleLspCall = async (request: LspToolCallRequest) => { - if (!lspTool) { - postMessageSafe({ - type: "lsp_tool_result", - callId: request.callId, - error: "LSP proxy not available", - }); - return; - } - try { - const result = await withTimeout( - lspTool.execute(request.callId, request.params as LspParams, signal), - request.timeoutMs, - ); - postMessageSafe({ - type: "lsp_tool_result", - callId: request.callId, - result: { content: result.content ?? [], details: result.details }, - }); - } catch (error) { - const message = - request.timeoutMs !== undefined && error instanceof Error && error.message.includes("timed out") - ? `LSP tool call timed out after ${request.timeoutMs}ms` - : error instanceof Error - ? error.message - : String(error); - postMessageSafe({ - type: "lsp_tool_result", - callId: request.callId, - error: message, - }); - } - }; - - const onMessage = (event: WorkerMessageEvent) => { - const message = event.data; - if (!message || resolved) return; - if (message.type === "mcp_tool_call") { - handleMCPCall(message as MCPToolCallRequest); - return; - } - if (message.type === "python_tool_call") { - handlePythonCall(message as PythonToolCallRequest); - return; - } - if (message.type === "python_tool_cancel") { - handlePythonCancel(message as PythonToolCallCancel); - return; - } - if (message.type === "lsp_tool_call") { - handleLspCall(message as LspToolCallRequest); - return; - } - if (message.type === "event") { - try { - processEvent(message.event); - } catch (err) { - finalize?.({ - type: "done", - exitCode: 1, - durationMs: Date.now() - startTime, - error: `Failed to process worker event: ${err instanceof Error ? err.message : String(err)}`, - }); - } - return; - } - if (message.type === "done") { - // Worker is exiting - mark as terminated to prevent calling terminate() on dead worker - terminated = true; - finalize?.(message); - } - }; - const onError = (event: WorkerErrorEvent) => { - // Worker error likely means it's dead or dying - terminated = true; - finalize?.({ - type: "done", - exitCode: 1, - durationMs: Date.now() - startTime, - error: event.message, - }); - }; - const onMessageError = () => { - // Message error may indicate worker is in bad state - terminated = true; - finalize?.({ - type: "done", - exitCode: 1, - durationMs: Date.now() - startTime, - error: "Worker message deserialization failed", - }); - }; - const onClose = () => { - // Worker terminated unexpectedly (crashed or was killed without sending done) - // Mark as terminated since the worker is already dead - calling terminate() again would crash - terminated = true; - const abortMessage = - abortSent && abortReason === "signal" - ? "Worker terminated after abort" - : abortSent - ? "Worker terminated after tool completion" - : "Worker terminated unexpectedly"; - finalize?.({ - type: "done", - exitCode: 1, - durationMs: Date.now() - startTime, - error: abortMessage, - aborted: abortReason === "signal", - }); - }; - worker.addEventListener("message", onMessage, { signal: listenerSignal }); - worker.addEventListener("error", onError, { signal: listenerSignal }); - worker.addEventListener("close", onClose, { signal: listenerSignal }); - worker.addEventListener("messageerror", onMessageError, { signal: listenerSignal }); - try { - worker.postMessage(startMessage); - } catch (err) { - finalize({ - type: "done", - exitCode: 1, - durationMs: Date.now() - startTime, - error: `Failed to start worker: ${err instanceof Error ? err.message : String(err)}`, - }); - } - }); - - // Cleanup - cancel any pending timeouts first - if (terminationTimeoutId) { - clearTimeout(terminationTimeoutId); - terminationTimeoutId = null; - } if (progressTimeoutId) { clearTimeout(progressTimeoutId); progressTimeoutId = null; } cancelPendingTermination(); - if (!terminated) { - terminated = true; - try { - worker.terminate(); - } catch { - // Ignore termination errors - } - } let exitCode = done.exitCode; if (done.error) { diff --git a/packages/coding-agent/src/task/types.ts b/packages/coding-agent/src/task/types.ts index 7e727ef3d..35b064349 100644 --- a/packages/coding-agent/src/task/types.ts +++ b/packages/coding-agent/src/task/types.ts @@ -22,7 +22,7 @@ function getEnv(name: string, defaultValue: number): number { /** Maximum tasks per call */ export const MAX_PARALLEL_TASKS = getEnv("OMP_TASK_MAX_PARALLEL", 32); -/** Maximum concurrent workers */ +/** Maximum concurrent tasks */ export const MAX_CONCURRENCY = getEnv("OMP_TASK_MAX_CONCURRENCY", 16); /** Maximum output bytes per agent */ diff --git a/packages/coding-agent/src/task/worker-protocol.ts b/packages/coding-agent/src/task/worker-protocol.ts deleted file mode 100644 index 289141998..000000000 --- a/packages/coding-agent/src/task/worker-protocol.ts +++ /dev/null @@ -1,131 +0,0 @@ -import type { AgentEvent, ThinkingLevel } from "@oh-my-pi/pi-agent-core"; -import type { SerializedModelRegistry } from "../config/model-registry"; -import type { PromptTemplate } from "../config/prompt-templates"; -import type { Settings } from "../config/settings-manager"; -import type { Skill } from "../extensibility/skills"; -import type { PreludeHelper } from "../ipy/kernel"; -import type { SerializedAuthStorage } from "../session/auth-storage"; -import type { ContextFileEntry } from "../tools"; - -/** - * MCP tool metadata passed from parent to worker for proxy tool creation. - */ -export interface MCPToolMetadata { - name: string; - label: string; - description: string; - parameters: unknown; - serverName: string; - mcpToolName: string; - timeoutMs?: number; -} - -/** - * Worker -> Parent: request to execute an MCP tool via parent's connection. - */ -export interface MCPToolCallRequest { - type: "mcp_tool_call"; - callId: string; - /** Server name for routing to correct connection */ - serverName: string; - /** Original MCP tool name (as expected by the MCP server) */ - mcpToolName: string; - params: Record; - timeoutMs?: number; -} - -/** - * Parent -> Worker: result of an MCP tool call. - */ -export interface MCPToolCallResponse { - type: "mcp_tool_result"; - callId: string; - result?: { - content: Array<{ type: string; text?: string; [key: string]: unknown }>; - isError?: boolean; - }; - error?: string; -} - -export interface PythonToolCallRequest { - type: "python_tool_call"; - callId: string; - params: Record; - timeoutMs?: number; -} - -export interface PythonToolCallResponse { - type: "python_tool_result"; - callId: string; - result?: { - content: Array<{ type: string; text?: string; [key: string]: unknown }>; - details?: unknown; - isError?: boolean; - }; - error?: string; -} - -export interface PythonToolCallCancel { - type: "python_tool_cancel"; - callId: string; - reason?: string; -} - -export interface LspToolCallRequest { - type: "lsp_tool_call"; - callId: string; - params: Record; - timeoutMs?: number; -} - -export interface LspToolCallResponse { - type: "lsp_tool_result"; - callId: string; - result?: { - content: Array<{ type: string; text?: string; [key: string]: unknown }>; - details?: unknown; - isError?: boolean; - }; - error?: string; -} - -export interface SubagentWorkerStartPayload { - cwd: string; - worktree?: string; - task: string; - systemPrompt: string; - model?: string; - thinkingLevel?: ThinkingLevel; - toolNames?: string[]; - outputSchema?: unknown; - enableLsp?: boolean; - sessionFile?: string | null; - spawnsEnv?: string; - serializedAuth?: SerializedAuthStorage; - serializedModels?: SerializedModelRegistry; - serializedSettings?: Settings; - pythonPreludeDocs?: PreludeHelper[]; - contextFiles?: ContextFileEntry[]; - skills?: Skill[]; - preloadedSkills?: Skill[]; - promptTemplates?: PromptTemplate[]; - mcpTools?: MCPToolMetadata[]; - pythonToolProxy?: boolean; - lspToolProxy?: boolean; -} - -export type SubagentWorkerRequest = - | { type: "start"; payload: SubagentWorkerStartPayload } - | { type: "abort" } - | MCPToolCallResponse - | PythonToolCallResponse - | PythonToolCallCancel - | LspToolCallResponse; - -export type SubagentWorkerResponse = - | { type: "event"; event: AgentEvent } - | { type: "done"; exitCode: number; durationMs: number; error?: string; aborted?: boolean } - | MCPToolCallRequest - | PythonToolCallRequest - | PythonToolCallCancel - | LspToolCallRequest; diff --git a/packages/coding-agent/src/task/worker.ts b/packages/coding-agent/src/task/worker.ts deleted file mode 100644 index 431b96709..000000000 --- a/packages/coding-agent/src/task/worker.ts +++ /dev/null @@ -1,921 +0,0 @@ -/** - * Worker thread for subagent execution. - * - * This worker runs in a separate thread via Bun's Worker API. It creates a minimal - * AgentSession and forwards events back to the parent thread. - * - * ## Event Flow - * - * 1. Parent sends { type: "start", payload } with task config - * 2. Worker creates AgentSession and subscribes to events - * 3. Worker forwards AgentEvent messages via postMessage - * 4. Worker sends { type: "done", exitCode, ... } on completion - * 5. Parent can send { type: "abort" } to request cancellation - */ -import type { AgentEvent, ThinkingLevel } from "@oh-my-pi/pi-agent-core"; -import type { Api, Model } from "@oh-my-pi/pi-ai"; -import { logger, postmortem, untilAborted } from "@oh-my-pi/pi-utils"; -import type { TSchema } from "@sinclair/typebox"; -import { ModelRegistry } from "../config/model-registry"; -import { parseModelPattern, parseModelString } from "../config/model-resolver"; -import { renderPromptTemplate } from "../config/prompt-templates"; -import { SettingsManager } from "../config/settings-manager"; -import type { CustomTool } from "../extensibility/custom-tools/types"; -import { setPreludeDocsCache } from "../ipy/executor"; -import { type LspToolDetails, lspSchema } from "../lsp/types"; -import lspDescription from "../prompts/tools/lsp.md" with { type: "text" }; -import { createAgentSession, discoverAuthStorage, discoverModels } from "../sdk"; -import type { AgentSessionEvent } from "../session/agent-session"; -import { AuthStorage } from "../session/auth-storage"; -import { SessionManager } from "../session/session-manager"; -import { getPythonToolDescription, type PythonToolDetails, type PythonToolParams, pythonSchema } from "../tools/python"; -import { ToolAbortError } from "../tools/tool-errors"; -import type { - LspToolCallResponse, - MCPToolCallResponse, - MCPToolMetadata, - PythonToolCallResponse, - SubagentWorkerRequest, - SubagentWorkerResponse, - SubagentWorkerStartPayload, -} from "./worker-protocol"; - -type PostMessageFn = (message: SubagentWorkerResponse) => void; - -const postMessageSafe: PostMessageFn = message => { - try { - (globalThis as typeof globalThis & { postMessage: PostMessageFn }).postMessage(message); - } catch { - // Parent may have terminated worker, nothing we can do - } -}; - -interface PendingMCPCall { - resolve: (result: MCPToolCallResponse["result"]) => void; - reject: (error: Error) => void; - timeoutId: ReturnType; -} - -interface PendingPythonCall { - resolve: (result: PythonToolCallResponse["result"]) => void; - reject: (error: Error) => void; - timeoutId?: ReturnType; -} - -interface PendingLspCall { - resolve: (result: LspToolCallResponse["result"]) => void; - reject: (error: Error) => void; - timeoutId?: ReturnType; -} - -const pendingMCPCalls = new Map(); -const pendingPythonCalls = new Map(); -const pendingLspCalls = new Map(); -const MCP_CALL_TIMEOUT_MS = 60_000; -let mcpCallIdCounter = 0; -let pythonCallIdCounter = 0; -let lspCallIdCounter = 0; - -function generateMCPCallId(): string { - return `mcp_${Date.now()}_${++mcpCallIdCounter}`; -} - -function generatePythonCallId(): string { - return `python_${Date.now()}_${++pythonCallIdCounter}`; -} - -function generateLspCallId(): string { - return `lsp_${Date.now()}_${++lspCallIdCounter}`; -} - -function callMCPToolViaParent( - serverName: string, - mcpToolName: string, - params: Record, - signal?: AbortSignal, - timeoutMs = MCP_CALL_TIMEOUT_MS, -): Promise<{ content: Array<{ type: string; text?: string; [key: string]: unknown }>; isError?: boolean }> { - const { promise, resolve, reject } = Promise.withResolvers<{ - content: Array<{ type: string; text?: string; [key: string]: unknown }>; - isError?: boolean; - }>(); - const callId = generateMCPCallId(); - if (signal?.aborted) { - reject(new ToolAbortError()); - return promise; - } - - const timeoutId = setTimeout(() => { - pendingMCPCalls.delete(callId); - reject(new Error(`MCP call timed out after ${timeoutMs}ms`)); - }, timeoutMs); - - const cleanup = () => { - clearTimeout(timeoutId); - pendingMCPCalls.delete(callId); - }; - - if (typeof signal?.addEventListener === "function") { - signal.addEventListener( - "abort", - () => { - cleanup(); - reject(new ToolAbortError()); - }, - { once: true }, - ); - } - - pendingMCPCalls.set(callId, { - resolve: result => { - cleanup(); - resolve(result ?? { content: [] }); - }, - reject: error => { - cleanup(); - reject(error); - }, - timeoutId, - }); - - postMessageSafe({ - type: "mcp_tool_call", - callId, - serverName, - mcpToolName, - params, - timeoutMs, - } as SubagentWorkerResponse); - - return promise; -} - -function callPythonToolViaParent( - params: PythonToolParams, - signal?: AbortSignal, - timeoutMs?: number, -): Promise { - const { promise, resolve, reject } = Promise.withResolvers(); - const callId = generatePythonCallId(); - if (signal?.aborted) { - reject(new ToolAbortError()); - return promise; - } - - const sendCancel = (reason: string) => { - postMessageSafe({ type: "python_tool_cancel", callId, reason } as SubagentWorkerResponse); - }; - - const timeoutId = - typeof timeoutMs === "number" && Number.isFinite(timeoutMs) - ? setTimeout(() => { - pendingPythonCalls.delete(callId); - sendCancel(`Python call timed out after ${timeoutMs}ms`); - reject(new Error(`Python call timed out after ${timeoutMs}ms`)); - }, timeoutMs) - : undefined; - - const cleanup = () => { - if (timeoutId) { - clearTimeout(timeoutId); - } - pendingPythonCalls.delete(callId); - }; - - if (typeof signal?.addEventListener === "function") { - signal.addEventListener( - "abort", - () => { - cleanup(); - sendCancel("Aborted"); - reject(new ToolAbortError()); - }, - { once: true }, - ); - } - - pendingPythonCalls.set(callId, { - resolve: result => { - cleanup(); - resolve(result ?? { content: [] }); - }, - reject: error => { - cleanup(); - reject(error); - }, - timeoutId, - }); - - postMessageSafe({ - type: "python_tool_call", - callId, - params, - timeoutMs, - } as SubagentWorkerResponse); - - return promise; -} - -function callLspToolViaParent( - params: Record, - signal?: AbortSignal, - timeoutMs?: number, -): Promise { - const { promise, resolve, reject } = Promise.withResolvers(); - const callId = generateLspCallId(); - if (signal?.aborted) { - reject(new ToolAbortError()); - return promise; - } - - const timeoutId = - typeof timeoutMs === "number" && Number.isFinite(timeoutMs) - ? setTimeout(() => { - pendingLspCalls.delete(callId); - reject(new Error(`LSP call timed out after ${timeoutMs}ms`)); - }, timeoutMs) - : undefined; - - const cleanup = () => { - if (timeoutId) { - clearTimeout(timeoutId); - } - pendingLspCalls.delete(callId); - }; - - if (typeof signal?.addEventListener === "function") { - signal.addEventListener( - "abort", - () => { - cleanup(); - reject(new ToolAbortError()); - }, - { once: true }, - ); - } - - pendingLspCalls.set(callId, { - resolve: result => { - cleanup(); - resolve(result ?? { content: [] }); - }, - reject: error => { - cleanup(); - reject(error); - }, - timeoutId, - }); - - postMessageSafe({ - type: "lsp_tool_call", - callId, - params, - timeoutMs, - } as SubagentWorkerResponse); - - return promise; -} - -function handleMCPToolResult(response: MCPToolCallResponse): void { - const pending = pendingMCPCalls.get(response.callId); - if (!pending) return; - if (response.error) { - pending.reject(new Error(response.error)); - } else { - pending.resolve(response.result); - } -} - -function handlePythonToolResult(response: PythonToolCallResponse): void { - const pending = pendingPythonCalls.get(response.callId); - if (!pending) return; - if (response.error) { - pending.reject(new Error(response.error)); - } else { - pending.resolve(response.result); - } -} - -function handleLspToolResult(response: LspToolCallResponse): void { - const pending = pendingLspCalls.get(response.callId); - if (!pending) return; - if (response.error) { - pending.reject(new Error(response.error)); - } else { - pending.resolve(response.result); - } -} - -function rejectPendingCalls(reason: string): void { - const error = new Error(reason); - const mcpCalls = Array.from(pendingMCPCalls.values()); - const pythonCalls = Array.from(pendingPythonCalls.values()); - const lspCalls = Array.from(pendingLspCalls.values()); - pendingMCPCalls.clear(); - pendingPythonCalls.clear(); - pendingLspCalls.clear(); - for (const pending of mcpCalls) { - clearTimeout(pending.timeoutId); - pending.reject(error); - } - for (const pending of pythonCalls) { - clearTimeout(pending.timeoutId); - pending.reject(error); - } - for (const pending of lspCalls) { - clearTimeout(pending.timeoutId); - pending.reject(error); - } -} - -function createMCPProxyTool(metadata: MCPToolMetadata): CustomTool { - return { - name: metadata.name, - label: metadata.label, - description: metadata.description, - parameters: metadata.parameters as TSchema, - execute: async (_toolCallId, params, _onUpdate, _ctx, signal) => { - try { - const result = await callMCPToolViaParent( - metadata.serverName, - metadata.mcpToolName, - params as Record, - signal, - metadata.timeoutMs, - ); - return { - content: result.content.map(c => - c.type === "text" - ? { type: "text" as const, text: c.text ?? "" } - : { type: "text" as const, text: JSON.stringify(c) }, - ), - details: { serverName: metadata.serverName, mcpToolName: metadata.mcpToolName, isError: result.isError }, - }; - } catch (error) { - return { - content: [ - { - type: "text" as const, - text: `MCP error: ${error instanceof Error ? error.message : String(error)}`, - }, - ], - details: { serverName: metadata.serverName, mcpToolName: metadata.mcpToolName, isError: true }, - }; - } - }, - }; -} - -function getPythonCallTimeoutMs(params: PythonToolParams): number | undefined { - const timeout = params.timeout; - if (typeof timeout === "number" && Number.isFinite(timeout) && timeout > 0) { - return Math.max(1000, Math.round(timeout * 1000) + 1000); - } - return undefined; -} - -function createPythonProxyTool(): CustomTool { - return { - name: "python", - label: "Python", - description: getPythonToolDescription(), - parameters: pythonSchema, - execute: async (_toolCallId, params, _onUpdate, _ctx, signal) => { - try { - const timeoutMs = getPythonCallTimeoutMs(params as PythonToolParams); - const result = await callPythonToolViaParent(params as PythonToolParams, signal, timeoutMs); - return { - content: - result?.content?.map(c => - c.type === "text" - ? { type: "text" as const, text: c.text ?? "" } - : { type: "text" as const, text: JSON.stringify(c) }, - ) ?? [], - details: result?.details as PythonToolDetails | undefined, - }; - } catch (error) { - return { - content: [ - { - type: "text" as const, - text: `Python error: ${error instanceof Error ? error.message : String(error)}`, - }, - ], - details: { isError: true } as PythonToolDetails, - }; - } - }, - }; -} - -function createLspProxyTool(): CustomTool { - return { - name: "lsp", - label: "LSP", - description: renderPromptTemplate(lspDescription), - parameters: lspSchema, - execute: async (_toolCallId, params, _onUpdate, _ctx, signal) => { - try { - const result = await callLspToolViaParent(params as Record, signal); - return { - content: - result?.content?.map(c => - c.type === "text" - ? { type: "text" as const, text: c.text ?? "" } - : { type: "text" as const, text: JSON.stringify(c) }, - ) ?? [], - details: result?.details as LspToolDetails | undefined, - }; - } catch (error) { - const { action } = params; - return { - content: [ - { - type: "text" as const, - text: `LSP error: ${error instanceof Error ? error.message : String(error)}`, - }, - ], - details: { action, success: false } as LspToolDetails, - }; - } - }, - }; -} - -interface WorkerMessageEvent { - data: T; -} - -/** Agent event types to forward to parent (excludes session-only events like compaction) */ -const agentEventTypes = new Set([ - "agent_start", - "agent_end", - "turn_start", - "turn_end", - "message_start", - "message_update", - "message_end", - "tool_execution_start", - "tool_execution_update", - "tool_execution_end", -]); - -const isAgentEvent = (event: AgentSessionEvent): event is AgentEvent => { - return agentEventTypes.has(event.type as AgentEvent["type"]); -}; - -class RunState { - abortController = new AbortController(); - startTime = Date.now(); - session: { abort: () => Promise; dispose: () => Promise } | null = null; - unsubscribe: (() => void) | null = null; - - private doneSent = false; - - sendDoneOnce(message: Extract): void { - if (this.doneSent) return; - this.doneSent = true; - postMessageSafe(message); - } -} - -let activeRun: RunState | null = null; -let pendingAbort = false; - -/** - * Resolve model string to Model object with optional thinking level. - * Supports both exact "provider/id" format and fuzzy matching ("sonnet", "opus"). - */ -function resolveModelOverride( - override: string | undefined, - modelRegistry: { getAvailable: () => Model[]; find: (provider: string, id: string) => Model | undefined }, -): { model?: Model; thinkingLevel?: ThinkingLevel } { - if (!override) return {}; - - // Try exact "provider/id" format first - const parsed = parseModelString(override); - if (parsed) { - return { model: modelRegistry.find(parsed.provider, parsed.id) }; - } - - // Fall back to fuzzy pattern matching - const result = parseModelPattern(override, modelRegistry.getAvailable()); - return { - model: result.model, - thinkingLevel: result.thinkingLevel !== "off" ? result.thinkingLevel : undefined, - }; -} - -/** - * Main task execution function. - * - * Equivalent to CLI flow: - * 1. omp --mode json --non-interactive - * 2. --append-system-prompt - * 3. --tools (if specified) - * 4. --model (if specified) - * 5. --session OR --no-session - * 6. --prompt - * - * Environment equivalent: - * - OMP_BLOCKED_AGENT: payload.blockedAgent (prevents same-agent recursion) - * - OMP_SPAWNS: payload.spawnsEnv (controls nested spawn permissions) - */ -async function runTask(runState: RunState, payload: SubagentWorkerStartPayload): Promise { - const { signal } = runState.abortController; - const startTime = runState.startTime; - let exitCode = 0; - let error: string | undefined; - let aborted = false; - const sessionAbortController = new AbortController(); - if (payload.pythonPreludeDocs && payload.pythonPreludeDocs.length > 0) { - setPreludeDocsCache(payload.pythonPreludeDocs); - } - - // Helper to check abort status - throws if aborted to exit early - const checkAbort = (): void => { - if (signal.aborted) { - aborted = true; - exitCode = 1; - throw new ToolAbortError(); - } - }; - - try { - // Check for pre-start abort - checkAbort(); - - // Use serialized auth/models if provided, otherwise discover from disk - let authStorage: AuthStorage; - let modelRegistry: ModelRegistry; - - if (payload.serializedAuth && payload.serializedModels) { - authStorage = await AuthStorage.fromSerialized(payload.serializedAuth); - modelRegistry = ModelRegistry.fromSerialized(payload.serializedModels, authStorage); - } else { - authStorage = await discoverAuthStorage(); - checkAbort(); - modelRegistry = await discoverModels(authStorage); - checkAbort(); - } - - // Create MCP/python/LSP proxy tools if provided - const mcpProxyTools: CustomTool[] = payload.mcpTools?.map(createMCPProxyTool) ?? []; - const pythonProxyTools: CustomTool[] = payload.pythonToolProxy - ? [createPythonProxyTool() as unknown as CustomTool] - : []; - const lspProxyTools: CustomTool[] = payload.lspToolProxy - ? [createLspProxyTool() as unknown as CustomTool] - : []; - const proxyTools = [...mcpProxyTools, ...pythonProxyTools, ...lspProxyTools]; - const enableLsp = payload.enableLsp ?? true; - const lspProxyEnabled = payload.lspToolProxy ?? false; - - // Resolve model override (equivalent to CLI's parseModelPattern with --model) - const { model, thinkingLevel: modelThinkingLevel } = resolveModelOverride(payload.model, modelRegistry); - const thinkingLevel = modelThinkingLevel ?? payload.thinkingLevel; - - // Create session manager (equivalent to CLI's --session or --no-session) - const sessionManager = payload.sessionFile - ? await SessionManager.open(payload.sessionFile) - : SessionManager.inMemory(payload.worktree ?? payload.cwd); - checkAbort(); - - // Use serialized settings if provided, otherwise use empty in-memory settings - // This avoids opening the SQLite database in worker threads - const settingsManager = SettingsManager.inMemory(payload.serializedSettings ?? {}); - - // Create agent session (equivalent to CLI's createAgentSession) - // Note: hasUI: false disables interactive features - const completionInstruction = - "When finished, call the complete tool exactly once. Do not end with a plain-text final answer."; - const worktreeNotice = payload.worktree - ? `You will work under this working tree: ${payload.worktree}. CRITICAL: Do not touch the original repository; only make changes inside this worktree.` - : ""; - - const skipPythonPreflight = - payload.pythonToolProxy === true || - (Array.isArray(payload.toolNames) && !payload.toolNames.includes("python")); - const { session } = await createAgentSession({ - cwd: payload.worktree ?? payload.cwd, - authStorage, - modelRegistry, - settingsManager, - model, - thinkingLevel, - toolNames: payload.toolNames, - outputSchema: payload.outputSchema, - requireCompleteTool: true, - contextFiles: payload.contextFiles, - skills: payload.skills, - preloadedSkills: payload.preloadedSkills, - promptTemplates: payload.promptTemplates, - // Append system prompt (equivalent to CLI's --append-system-prompt) - systemPrompt: defaultPrompt => - `${defaultPrompt}\n\n${payload.systemPrompt}\n\n${worktreeNotice}\n\n${completionInstruction}`, - sessionManager, - hasUI: false, - // Pass spawn restrictions to nested tasks - spawns: payload.spawnsEnv, - enableLsp: enableLsp && !lspProxyEnabled, - skipPythonPreflight, - // Disable local MCP discovery if using proxy tools - enableMCP: !payload.mcpTools, - // Add proxy tools - customTools: proxyTools.length > 0 ? proxyTools : undefined, - }); - - runState.session = session; - checkAbort(); - - // Write session init metadata for debugging/replay - session.sessionManager.appendSessionInit({ - systemPrompt: session.agent.state.systemPrompt, - task: payload.task, - tools: session.getAllToolNames(), - outputSchema: payload.outputSchema, - }); - - signal.addEventListener( - "abort", - () => { - void session.abort(); - }, - { once: true, signal: sessionAbortController.signal }, - ); - - // Initialize extensions (equivalent to CLI's extension initialization) - // Note: Does not support --extension CLI flag or extension CLI flags - const extensionRunner = session.extensionRunner; - if (extensionRunner) { - extensionRunner.initialize( - // ExtensionActions - { - sendMessage: (message, options) => { - session.sendCustomMessage(message, options).catch(e => { - logger.error("Extension sendMessage failed", { - error: e instanceof Error ? e.message : String(e), - }); - }); - }, - sendUserMessage: (content, options) => { - session.sendUserMessage(content, options).catch(e => { - logger.error("Extension sendUserMessage failed", { - error: e instanceof Error ? e.message : String(e), - }); - }); - }, - appendEntry: (customType, data) => { - session.sessionManager.appendCustomEntry(customType, data); - }, - setLabel: (targetId, label) => { - session.sessionManager.appendLabelChange(targetId, label); - }, - getActiveTools: () => session.getActiveToolNames(), - getAllTools: () => session.getAllToolNames(), - setActiveTools: (toolNames: string[]) => session.setActiveToolsByName(toolNames), - setModel: async model => { - const key = await session.modelRegistry.getApiKey(model); - if (!key) return false; - await session.setModel(model); - return true; - }, - getThinkingLevel: () => session.thinkingLevel, - setThinkingLevel: level => session.setThinkingLevel(level), - }, - // ExtensionContextActions - { - getModel: () => session.model, - isIdle: () => !session.isStreaming, - abort: () => session.abort(), - hasPendingMessages: () => session.queuedMessageCount > 0, - shutdown: () => {}, - getContextUsage: () => session.getContextUsage(), - compact: async instructionsOrOptions => { - const instructions = typeof instructionsOrOptions === "string" ? instructionsOrOptions : undefined; - const options = - instructionsOrOptions && typeof instructionsOrOptions === "object" - ? instructionsOrOptions - : undefined; - await session.compact(instructions, options); - }, - }, - ); - extensionRunner.onError(err => { - logger.error("Extension error", { path: err.extensionPath, error: err.error }); - }); - await extensionRunner.emit({ type: "session_start" }); - } - - // Track complete tool calls - const MAX_COMPLETE_RETRIES = 3; - let completeCalled = false; - - // Subscribe to events and forward to parent (equivalent to --mode json output) - runState.unsubscribe = session.subscribe((event: AgentSessionEvent) => { - if (isAgentEvent(event)) { - postMessageSafe({ type: "event", event }); - // Track when complete tool is called - if (event.type === "tool_execution_end" && event.toolName === "complete") { - completeCalled = true; - } - } - }); - - // Run the prompt (equivalent to --prompt flag) - await session.prompt(payload.task); - - // Retry loop if complete was not called - let retryCount = 0; - while (!completeCalled && retryCount < MAX_COMPLETE_RETRIES && !signal.aborted) { - retryCount++; - const reminder = ` -CRITICAL: You stopped without calling the complete tool. This is reminder ${retryCount} of ${MAX_COMPLETE_RETRIES}. - -You MUST call the complete tool to finish your task. Options: -1. Call complete with your result data if you have completed the task -2. Call complete with status="aborted" and an error message if you cannot complete the task - -Failure to call complete after ${MAX_COMPLETE_RETRIES} reminders will result in task failure. - - -Call complete now.`; - - await session.prompt(reminder); - } - - // Check if aborted during execution - const lastMessage = session.state.messages[session.state.messages.length - 1]; - if (lastMessage?.role === "assistant" && lastMessage.stopReason === "aborted") { - aborted = true; - exitCode = 1; - } - } catch (err) { - exitCode = 1; - // Don't record abort as error - it's handled via the aborted flag - if (!signal.aborted) { - error = err instanceof Error ? err.stack || err.message : String(err); - } - } finally { - // Handle abort requested during execution - if (signal.aborted) { - aborted = true; - if (exitCode === 0) exitCode = 1; - } - - sessionAbortController.abort(); - rejectPendingCalls("Worker finished"); - - if (runState.unsubscribe) { - try { - runState.unsubscribe(); - } catch { - // Ignore unsubscribe errors - } - runState.unsubscribe = null; - } - - // Cleanup session with timeout to prevent hanging - if (runState.session) { - const session = runState.session; - runState.session = null; - try { - await untilAborted(AbortSignal.timeout(5000), () => session.dispose()); - } catch { - // Ignore cleanup errors - } - } - - if (activeRun === runState) { - activeRun = null; - } - - // Send completion message to parent (only once) - runState.sendDoneOnce({ - type: "done", - exitCode, - durationMs: Date.now() - startTime, - error, - aborted, - }); - } -} - -/** Handle abort request from parent */ -function handleAbort(): void { - const runState = activeRun; - if (!runState) { - pendingAbort = true; - rejectPendingCalls("Aborted"); - return; - } - rejectPendingCalls("Aborted"); - runState.abortController.abort(); - if (runState.session) { - void runState.session.abort(); - } -} - -const reportFatal = async (message: string): Promise => { - // Run postmortem cleanup first to ensure child processes are killed - try { - await postmortem.cleanup(); - } catch { - // Ignore cleanup errors - } - const error = new Error(message); - - const runState = activeRun; - if (runState) { - runState.abortController.abort(error); - if (runState.session) { - void runState.session.abort(); - } - runState.sendDoneOnce({ - type: "done", - exitCode: 1, - durationMs: Date.now() - runState.startTime, - error: message, - aborted: false, - }); - return; - } - - postMessageSafe({ - type: "done", - exitCode: 1, - durationMs: 0, - error: message, - aborted: false, - }); -}; - -// Global error handlers to ensure we always send a done message -// Using self instead of globalThis for proper worker scope typing -declare const self: { - addEventListener(type: "error", listener: (event: ErrorEvent) => void): void; - addEventListener(type: "unhandledrejection", listener: (event: { reason: unknown }) => void): void; - addEventListener(type: "messageerror", listener: (event: MessageEvent) => void): void; -}; - -self.addEventListener("error", event => { - reportFatal(`Uncaught error: ${event.message || "Unknown error"}`); -}); - -self.addEventListener("unhandledrejection", event => { - const reason = event.reason; - const message = reason instanceof Error ? reason.stack || reason.message : String(reason); - - // Avoid terminating active runs on tool-level errors that bubble as rejections. - if (activeRun) { - logger.error("Unhandled rejection in subagent worker", { error: message }); - if ("preventDefault" in event && typeof event.preventDefault === "function") { - event.preventDefault(); - } - return; - } - - reportFatal(`Unhandled rejection: ${message}`); -}); - -self.addEventListener("messageerror", () => { - reportFatal("Failed to deserialize parent message"); -}); - -// Message handler - receives start/abort/tool_result commands from parent -globalThis.addEventListener("message", (event: WorkerMessageEvent) => { - const message = event.data; - if (!message) return; - - if (message.type === "abort") { - handleAbort(); - return; - } - - if (message.type === "mcp_tool_result") { - handleMCPToolResult(message); - return; - } - - if (message.type === "python_tool_result") { - handlePythonToolResult(message); - return; - } - - if (message.type === "lsp_tool_result") { - handleLspToolResult(message); - return; - } - - if (message.type === "start") { - // Only allow one task per worker - if (activeRun) return; - const runState = new RunState(); - if (pendingAbort) { - pendingAbort = false; - runState.abortController.abort(); - } - activeRun = runState; - void runTask(runState, message.payload); - } -}); diff --git a/packages/coding-agent/src/tools/index.ts b/packages/coding-agent/src/tools/index.ts index 616d12630..a844d9802 100644 --- a/packages/coding-agent/src/tools/index.ts +++ b/packages/coding-agent/src/tools/index.ts @@ -1,4 +1,5 @@ import type { AgentTool } from "@oh-my-pi/pi-agent-core"; +import type { SettingsManager } from "@oh-my-pi/pi-coding-agent/config/settings-manager"; import { logger } from "@oh-my-pi/pi-utils"; import type { PromptTemplate } from "../config/prompt-templates"; import type { BashInterceptorRule } from "../config/settings-manager"; @@ -151,11 +152,8 @@ export interface ToolSession { internalRouter?: InternalUrlRouter; /** Agent output manager for unique agent:// IDs across task invocations */ agentOutputManager?: AgentOutputManager; - /** Settings manager for passing to subagents (avoids SQLite access in workers) */ - settingsManager?: { - serialize: () => import("../config/settings-manager").Settings; - getPlansDirectory: (cwd?: string) => string; - }; + /** Settings manager for passing to subagents */ + settingsManager?: SettingsManager; /** Settings manager (optional) */ settings?: { getImageAutoResize(): boolean;