diff --git a/README.md b/README.md index 8d80cb2f..4dde2659 100644 --- a/README.md +++ b/README.md @@ -28,7 +28,7 @@ > **⚠️ Early Development Notice:** This project is in early development and is not yet ready for production use. Features may change, break, or be incomplete.

- + Join our Discord    diff --git a/prompts/templates/dev-codemachine/main-agents/00-init.md b/prompts/templates/dev-codemachine/main-agents/00-init.md index 31c28918..edc53b9d 100644 --- a/prompts/templates/dev-codemachine/main-agents/00-init.md +++ b/prompts/templates/dev-codemachine/main-agents/00-init.md @@ -2,10 +2,10 @@ **Task:** -1. **Ensure the current branch is `codemacine/dev` by following this specific logic:** - * First, check if the *current* branch is already `codemacine/dev`. If it is, this step is complete. - * If not, check if a branch named `codemacine/dev` *already exists*. If it does, switch to it. - * If it does not exist, create it as a new branch and switch to it (e.g., `git checkout -b codemacine/dev`). +1. **Ensure the current branch is `codemachine/dev` by following this specific logic:** + * First, check if the *current* branch is already `codemachine/dev`. If it is, this step is complete. + * If not, check if a branch named `codemachine/dev` *already exists*. If it does, switch to it. + * If it does not exist, create it as a new branch and switch to it (e.g., `git checkout -b codemachine/dev`). 2. **Append the following lines to the `.gitignore` file, skipping any that already exist:** ``` diff --git a/src/agents/runner/runner.ts b/src/agents/runner/runner.ts index c50bd02b..87cd7db6 100644 --- a/src/agents/runner/runner.ts +++ b/src/agents/runner/runner.ts @@ -1,14 +1,17 @@ -import * as path from 'node:path'; - -import type { EngineType } from '../../infra/engines/index.js'; -import { getEngine } from '../../infra/engines/index.js'; -import { MemoryAdapter } from '../../infra/fs/memory-adapter.js'; -import { MemoryStore } from '../index.js'; -import { loadAgentConfig } from './config.js'; -import { AgentMonitorService, AgentLoggerService } from '../monitoring/index.js'; -import type { ParsedTelemetry } from '../../infra/engines/core/types.js'; -import { formatForLogFile } from '../../shared/formatters/logFileFormatter.js'; -import { info, error } from '../../shared/logging/logger.js'; +import * as path from "node:path"; + +import type { EngineType } from "../../infra/engines/index.js"; +import { getEngine } from "../../infra/engines/index.js"; +import { MemoryAdapter } from "../../infra/fs/memory-adapter.js"; +import { MemoryStore } from "../index.js"; +import { loadAgentConfig } from "./config.js"; +import { + AgentMonitorService, + AgentLoggerService, +} from "../monitoring/index.js"; +import type { ParsedTelemetry } from "../../infra/engines/core/types.js"; +import { formatForLogFile } from "../../shared/formatters/logFileFormatter.js"; +import { info, error } from "../../shared/logging/logger.js"; /** * Cache for engine authentication status with TTL (shared across all subagents) @@ -16,29 +19,33 @@ import { info, error } from '../../shared/logging/logger.js'; * CRITICAL: This fixes the 5-minute delay bug when spawning multiple subagents */ class EngineAuthCache { - private cache: Map = new Map(); - private ttlMs: number = 5 * 60 * 1000; // 5 minutes TTL - - async isAuthenticated(engineId: string, checkFn: () => Promise): Promise { - const cached = this.cache.get(engineId); - const now = Date.now(); - - // Return cached value if still valid - if (cached && (now - cached.timestamp) < this.ttlMs) { - return cached.isAuthenticated; - } - - // Cache miss or expired - perform actual check - const result = await checkFn(); - - // Cache the result - this.cache.set(engineId, { - isAuthenticated: result, - timestamp: now - }); - - return result; - } + private cache: Map = + new Map(); + private ttlMs: number = 5 * 60 * 1000; // 5 minutes TTL + + async isAuthenticated( + engineId: string, + checkFn: () => Promise, + ): Promise { + const cached = this.cache.get(engineId); + const now = Date.now(); + + // Return cached value if still valid + if (cached && now - cached.timestamp < this.ttlMs) { + return cached.isAuthenticated; + } + + // Cache miss or expired - perform actual check + const result = await checkFn(); + + // Cache the result + this.cache.set(engineId, { + isAuthenticated: result, + timestamp: now, + }); + + return result; + } } // Global auth cache instance (shared across all subagent executions) @@ -48,103 +55,115 @@ const authCache = new EngineAuthCache(); * Minimal UI interface for agent execution */ export interface AgentExecutionUI { - registerMonitoringId(uiAgentId: string, monitoringAgentId: number): void; + registerMonitoringId(uiAgentId: string, monitoringAgentId: number): void; } export interface ExecuteAgentOptions { - /** - * Engine to use (overrides agent config) - */ - engine?: EngineType; - - /** - * Model to use (overrides agent config) - */ - model?: string; - - /** - * Working directory for execution - */ - workingDir: string; - - /** - * Project root for config lookup (defaults to workingDir) - */ - projectRoot?: string; - - /** - * Logger for stdout - */ - logger?: (chunk: string) => void; - - /** - * Logger for stderr - */ - stderrLogger?: (chunk: string) => void; - - /** - * Telemetry callback (for UI updates) - */ - onTelemetry?: (telemetry: ParsedTelemetry) => void; - - /** - * Abort signal - */ - abortSignal?: AbortSignal; - - /** - * Timeout in milliseconds - */ - timeout?: number; - - /** - * Parent agent ID (for tracking parent-child relationships) - */ - parentId?: number; - - /** - * Disable monitoring (for special cases where monitoring is not desired) - */ - disableMonitoring?: boolean; - - /** - * UI manager (for registering monitoring IDs) - */ - ui?: AgentExecutionUI; - - /** - * Unique agent ID for UI (for registering monitoring IDs) - */ - uniqueAgentId?: string; - - /** - * Display prompt (for logging/monitoring - shows user's actual request) - * If not provided, uses the full execution prompt - */ - displayPrompt?: string; + /** + * Engine to use (overrides agent config) + */ + engine?: EngineType; + + /** + * Model to use (overrides agent config) + */ + model?: string; + + /** + * Working directory for execution + */ + workingDir: string; + + /** + * Project root for config lookup (defaults to workingDir) + */ + projectRoot?: string; + + /** + * Logger for stdout + */ + logger?: (chunk: string) => void; + + /** + * Logger for stderr + */ + stderrLogger?: (chunk: string) => void; + + /** + * Telemetry callback (for UI updates) + */ + onTelemetry?: (telemetry: ParsedTelemetry) => void; + + /** + * Abort signal + */ + abortSignal?: AbortSignal; + + /** + * Timeout in milliseconds + */ + timeout?: number; + + /** + * Parent agent ID (for tracking parent-child relationships) + */ + parentId?: number; + + /** + * Disable monitoring (for special cases where monitoring is not desired) + */ + disableMonitoring?: boolean; + + /** + * UI manager (for registering monitoring IDs) + */ + ui?: AgentExecutionUI; + + /** + * Unique agent ID for UI (for registering monitoring IDs) + */ + uniqueAgentId?: string; + + /** + * Display prompt (for logging/monitoring - shows user's actual request) + * If not provided, uses the full execution prompt + */ + displayPrompt?: string; + + /** + * Agent name for engines that support agent selection (e.g., OpenCode --agent flag) + */ + agent?: string; + + /** + * URL of running server to attach to (e.g., http://localhost:4096) + * For OpenCode, uses --attach flag to connect to existing server. + * This avoids MCP server cold boot times on every run. + */ + attach?: string; } /** * Ensures the engine is authenticated */ async function ensureEngineAuth(engineType: EngineType): Promise { - const { registry } = await import('../../infra/engines/index.js'); - const engine = registry.get(engineType); - - if (!engine) { - const availableEngines = registry.getAllIds().join(', '); - throw new Error( - `Unknown engine type: ${engineType}. Available engines: ${availableEngines}` - ); - } - - const isAuthed = await engine.auth.isAuthenticated(); - if (!isAuthed) { - console.error(`\n${engine.metadata.name} authentication required`); - console.error(`\nRun the following command to authenticate:\n`); - console.error(` codemachine auth login\n`); - throw new Error(`${engine.metadata.name} authentication required`); - } + const { registry } = await import("../../infra/engines/index.js"); + const engine = registry.get(engineType); + + if (!engine) { + const availableEngines = registry.getAllIds().join(", "); + throw new Error( + `Unknown engine type: ${engineType}. Available engines: ${availableEngines}`, + ); + } + + const isAuthed = await engine.auth.isAuthenticated(); + if (!isAuthed) { + console.error(`\n${engine.metadata.name} authentication required`); + console.error(`\nRun the following command to authenticate:\n`); + console.error(` codemachine auth login\n`); + throw new Error(`${engine.metadata.name} authentication required`); + } } /** @@ -172,204 +191,236 @@ async function ensureEngineAuth(engineType: EngineType): Promise { * - CLI commands (via orchestration) */ export interface AgentExecutionOutput { - output: string; - agentId?: number; + output: string; + agentId?: number; } export async function executeAgent( - agentId: string, - prompt: string, - options: ExecuteAgentOptions, + agentId: string, + prompt: string, + options: ExecuteAgentOptions, ): Promise { - const { workingDir, projectRoot, engine: engineOverride, model: modelOverride, logger, stderrLogger, onTelemetry, abortSignal, timeout, parentId, disableMonitoring, ui, uniqueAgentId, displayPrompt } = options; - - // Load agent config to determine engine and model - const agentConfig = await loadAgentConfig(agentId, projectRoot ?? workingDir); - - // Determine engine: CLI override > agent config > first authenticated engine - const { registry } = await import('../../infra/engines/index.js'); - let engineType: EngineType; - - if (engineOverride) { - engineType = engineOverride; - } else if (agentConfig.engine) { - engineType = agentConfig.engine; - } else { - // Fallback: find first authenticated engine by order (WITH CACHING - critical for subagents) - const engines = registry.getAll(); - let foundEngine = null; - - for (const engine of engines) { - // Use cached auth check to avoid 10-30 second delays per subagent - const isAuth = await authCache.isAuthenticated( - engine.metadata.id, - () => engine.auth.isAuthenticated() - ); - if (isAuth) { - foundEngine = engine; - break; - } - } - - if (!foundEngine) { - // If no authenticated engine, use default (first by order) - foundEngine = registry.getDefault(); - } - - if (!foundEngine) { - throw new Error('No engines registered. Please install at least one engine.'); - } - - engineType = foundEngine.metadata.id; - info(`No engine specified for agent '${agentId}', using ${foundEngine.metadata.name} (${engineType})`); - } - - // Ensure authentication - await ensureEngineAuth(engineType); - - // Get engine module for defaults - const engineModule = registry.get(engineType); - if (!engineModule) { - throw new Error(`Engine not found: ${engineType}`); - } - - // Model resolution: CLI override > agent config (legacy) > engine default - const model = modelOverride ?? (agentConfig.model as string | undefined) ?? engineModule.metadata.defaultModel; - const modelReasoningEffort = (agentConfig.modelReasoningEffort as 'low' | 'medium' | 'high' | undefined) ?? engineModule.metadata.defaultModelReasoningEffort; - - // Initialize monitoring with engine/model info (unless explicitly disabled) - const monitor = !disableMonitoring ? AgentMonitorService.getInstance() : null; - const loggerService = !disableMonitoring ? AgentLoggerService.getInstance() : null; - let monitoringAgentId: number | undefined; - - if (monitor && loggerService) { - // For registration: use displayPrompt (short user request) if provided, otherwise full prompt - const promptForDisplay = displayPrompt || prompt; - monitoringAgentId = await monitor.register({ - name: agentId, - prompt: promptForDisplay, // This gets truncated in monitor for memory efficiency - parentId, - engine: engineType, - engineProvider: engineType, - modelName: model, - }); - - // Store FULL prompt for debug mode logging (not the display prompt) - // In debug mode, we want to see the complete composite prompt with template + input files - loggerService.storeFullPrompt(monitoringAgentId, prompt); - - // Register monitoring ID with UI immediately so it can load logs - if (ui && uniqueAgentId && monitoringAgentId !== undefined) { - ui.registerMonitoringId(uniqueAgentId, monitoringAgentId); - } - } - - // Set up memory - const memoryDir = path.resolve(workingDir, '.codemachine', 'memory'); - const adapter = new MemoryAdapter(memoryDir); - const store = new MemoryStore(adapter); - - // Get engine and execute - // NOTE: Prompt is already complete - no template loading or building here - const engine = getEngine(engineType); - - let totalStdout = ''; - - try { - const result = await engine.run({ - prompt, // Already complete and ready to use - workingDir, - model, - modelReasoningEffort, - env: { - ...process.env, - // Pass parent agent ID to child processes (for orchestration context) - ...(monitoringAgentId !== undefined && { - CODEMACHINE_PARENT_AGENT_ID: monitoringAgentId.toString() - }) - }, - onData: (chunk) => { - totalStdout += chunk; - - // Dual-stream: write to log file (with status text) AND original logger (with colors) - if (loggerService && monitoringAgentId !== undefined) { - // Transform color markers to status text for log file readability - const logChunk = formatForLogFile(chunk); - loggerService.write(monitoringAgentId, logChunk); - } - - // Keep original format with color markers for UI display - if (logger) { - logger(chunk); - } else { - try { - process.stdout.write(chunk); - } catch { - // ignore streaming failures - } - } - }, - onErrorData: (chunk) => { - // Also log stderr to file (with status text transformation) - if (loggerService && monitoringAgentId !== undefined) { - const logChunk = formatForLogFile(chunk); - loggerService.write(monitoringAgentId, `[STDERR] ${logChunk}`); - } - - if (stderrLogger) { - stderrLogger(chunk); - } else { - try { - process.stderr.write(chunk); - } catch { - // ignore streaming failures - } - } - }, - onTelemetry: (telemetry) => { - // Update telemetry in monitoring (fire and forget - don't block streaming) - if (monitor && monitoringAgentId !== undefined) { - monitor.updateTelemetry(monitoringAgentId, telemetry).catch(err => - error(`Failed to update telemetry: ${err}`) - ); - } - - // Forward to caller's telemetry callback (for UI updates) - if (onTelemetry) { - onTelemetry(telemetry); - } - }, - abortSignal, - timeout, - }); - - // Store output in memory - const stdout = result.stdout || totalStdout; - const slice = stdout.slice(-2000); - await store.append({ - agentId, - content: slice, - timestamp: new Date().toISOString(), - }); - - // Mark agent as completed - if (monitor && monitoringAgentId !== undefined) { - await monitor.complete(monitoringAgentId); - // Note: Don't close stream here - workflow may write more messages - // Streams will be closed by cleanup handlers or monitoring service shutdown - } - - return { - output: stdout, - agentId: monitoringAgentId - }; - } catch (error) { - // Mark agent as failed - if (monitor && monitoringAgentId !== undefined) { - await monitor.fail(monitoringAgentId, error as Error); - // Note: Don't close stream here - workflow may write more messages - // Streams will be closed by cleanup handlers or monitoring service shutdown - } - throw error; - } + const { + workingDir, + projectRoot, + engine: engineOverride, + model: modelOverride, + logger, + stderrLogger, + onTelemetry, + abortSignal, + timeout, + parentId, + disableMonitoring, + ui, + uniqueAgentId, + displayPrompt, + } = options; + + // Load agent config to determine engine and model + const agentConfig = await loadAgentConfig(agentId, projectRoot ?? workingDir); + + // Determine engine: CLI override > agent config > first authenticated engine + const { registry } = await import("../../infra/engines/index.js"); + let engineType: EngineType; + + if (engineOverride) { + engineType = engineOverride; + } else if (agentConfig.engine) { + engineType = agentConfig.engine; + } else { + // Fallback: find first authenticated engine by order (WITH CACHING - critical for subagents) + const engines = registry.getAll(); + let foundEngine = null; + + for (const engine of engines) { + // Use cached auth check to avoid 10-30 second delays per subagent + const isAuth = await authCache.isAuthenticated(engine.metadata.id, () => + engine.auth.isAuthenticated(), + ); + if (isAuth) { + foundEngine = engine; + break; + } + } + + if (!foundEngine) { + // If no authenticated engine, use default (first by order) + foundEngine = registry.getDefault(); + } + + if (!foundEngine) { + throw new Error( + "No engines registered. Please install at least one engine.", + ); + } + + engineType = foundEngine.metadata.id; + info( + `No engine specified for agent '${agentId}', using ${foundEngine.metadata.name} (${engineType})`, + ); + } + + // Ensure authentication + await ensureEngineAuth(engineType); + + // Get engine module for defaults + const engineModule = registry.get(engineType); + if (!engineModule) { + throw new Error(`Engine not found: ${engineType}`); + } + + // Model resolution: CLI override > agent config (legacy) > engine default + const model = + modelOverride ?? + (agentConfig.model as string | undefined) ?? + engineModule.metadata.defaultModel; + const modelReasoningEffort = + (agentConfig.modelReasoningEffort as + | "low" + | "medium" + | "high" + | undefined) ?? engineModule.metadata.defaultModelReasoningEffort; + + // Initialize monitoring with engine/model info (unless explicitly disabled) + const monitor = !disableMonitoring ? AgentMonitorService.getInstance() : null; + const loggerService = !disableMonitoring + ? AgentLoggerService.getInstance() + : null; + let monitoringAgentId: number | undefined; + + if (monitor && loggerService) { + // For registration: use displayPrompt (short user request) if provided, otherwise full prompt + const promptForDisplay = displayPrompt || prompt; + monitoringAgentId = await monitor.register({ + name: agentId, + prompt: promptForDisplay, // This gets truncated in monitor for memory efficiency + parentId, + engine: engineType, + engineProvider: engineType, + modelName: model, + }); + + // Store FULL prompt for debug mode logging (not the display prompt) + // In debug mode, we want to see the complete composite prompt with template + input files + loggerService.storeFullPrompt(monitoringAgentId, prompt); + + // Register monitoring ID with UI immediately so it can load logs + if (ui && uniqueAgentId && monitoringAgentId !== undefined) { + ui.registerMonitoringId(uniqueAgentId, monitoringAgentId); + } + } + + // Set up memory + const memoryDir = path.resolve(workingDir, ".codemachine", "memory"); + const adapter = new MemoryAdapter(memoryDir); + const store = new MemoryStore(adapter); + + // Get engine and execute + // NOTE: Prompt is already complete - no template loading or building here + const engine = getEngine(engineType); + + let totalStdout = ""; + + try { + const result = await engine.run({ + prompt, // Already complete and ready to use + workingDir, + model, + modelReasoningEffort, + agent: options.agent, // Pass agent name for engines that support it (e.g., OpenCode --agent) + attach: options.attach, // Pass server URL for engines that support it (e.g., OpenCode --attach) + env: { + ...process.env, + // Pass parent agent ID to child processes (for orchestration context) + ...(monitoringAgentId !== undefined && { + CODEMACHINE_PARENT_AGENT_ID: monitoringAgentId.toString(), + }), + }, + onData: (chunk) => { + totalStdout += chunk; + + // Dual-stream: write to log file (with status text) AND original logger (with colors) + if (loggerService && monitoringAgentId !== undefined) { + // Transform color markers to status text for log file readability + const logChunk = formatForLogFile(chunk); + loggerService.write(monitoringAgentId, logChunk); + } + + // Keep original format with color markers for UI display + if (logger) { + logger(chunk); + } else { + try { + process.stdout.write(chunk); + } catch { + // ignore streaming failures + } + } + }, + onErrorData: (chunk) => { + // Also log stderr to file (with status text transformation) + if (loggerService && monitoringAgentId !== undefined) { + const logChunk = formatForLogFile(chunk); + loggerService.write(monitoringAgentId, `[STDERR] ${logChunk}`); + } + + if (stderrLogger) { + stderrLogger(chunk); + } else { + try { + process.stderr.write(chunk); + } catch { + // ignore streaming failures + } + } + }, + onTelemetry: (telemetry) => { + // Update telemetry in monitoring (fire and forget - don't block streaming) + if (monitor && monitoringAgentId !== undefined) { + monitor + .updateTelemetry(monitoringAgentId, telemetry) + .catch((err) => error(`Failed to update telemetry: ${err}`)); + } + + // Forward to caller's telemetry callback (for UI updates) + if (onTelemetry) { + onTelemetry(telemetry); + } + }, + abortSignal, + timeout, + }); + + // Store output in memory (skip if empty to avoid validation error) + const stdout = result.stdout || totalStdout; + const slice = stdout.slice(-2000).trim(); + if (slice) { + await store.append({ + agentId, + content: slice, + timestamp: new Date().toISOString(), + }); + } + + // Mark agent as completed + if (monitor && monitoringAgentId !== undefined) { + await monitor.complete(monitoringAgentId); + // Note: Don't close stream here - workflow may write more messages + // Streams will be closed by cleanup handlers or monitoring service shutdown + } + + return { + output: stdout, + agentId: monitoringAgentId, + }; + } catch (error) { + // Mark agent as failed + if (monitor && monitoringAgentId !== undefined) { + await monitor.fail(monitoringAgentId, error as Error); + // Note: Don't close stream here - workflow may write more messages + // Streams will be closed by cleanup handlers or monitoring service shutdown + } + throw error; + } } diff --git a/src/cli/commands/start.command.ts b/src/cli/commands/start.command.ts index 2aa0e812..519d9815 100644 --- a/src/cli/commands/start.command.ts +++ b/src/cli/commands/start.command.ts @@ -1,84 +1,141 @@ -import * as path from 'node:path'; -import type { Command } from 'commander'; +import * as path from "node:path"; +import type { Command } from "commander"; -import { debug } from '../../shared/logging/logger.js'; -import { clearTerminal } from '../../shared/utils/terminal.js'; +import { debug } from "../../shared/logging/logger.js"; +import { clearTerminal } from "../../shared/utils/terminal.js"; -const DEFAULT_SPEC_PATH = '.codemachine/inputs/specifications.md'; +const DEFAULT_SPEC_PATH = ".codemachine/inputs/specifications.md"; type StartCommandOptions = { - spec?: string; + spec?: string; + opencodeServer?: boolean; + opencodeAttach?: string; }; export function registerStartCommand(program: Command): void { - program - .command('start') - .description('Run the workflow queue until completion (non-interactive)') - .option('--spec ', 'Path to the planning specification file') - .action(async (options: StartCommandOptions, command: Command) => { - const cwd = process.env.CODEMACHINE_CWD || process.cwd(); + program + .command("start") + .description("Run the workflow queue until completion (non-interactive)") + .option("--spec ", "Path to the planning specification file") + .option( + "--opencode-server", + "[Experimental] Start a consolidated OpenCode server for all workflow steps. " + + "Reduces MCP cold boot times by reusing a single server instance.", + ) + .option( + "--opencode-attach ", + "Attach to an existing OpenCode server (e.g., http://localhost:4096). " + + "Mutually exclusive with --opencode-server.", + ) + .action(async (options: StartCommandOptions, command: Command) => { + const cwd = process.env.CODEMACHINE_CWD || process.cwd(); - // Use command-specific --spec if provided, otherwise fall back to global --spec, then default - const globalOpts = command.optsWithGlobals ? command.optsWithGlobals() : command.opts(); - const specPath = options.spec ?? globalOpts.spec ?? DEFAULT_SPEC_PATH; - const specificationPath = path.resolve(cwd, specPath); + // Use command-specific --spec if provided, otherwise fall back to global --spec, then default + const globalOpts = command.optsWithGlobals + ? command.optsWithGlobals() + : command.opts(); + const specPath = options.spec ?? globalOpts.spec ?? DEFAULT_SPEC_PATH; + const specificationPath = path.resolve(cwd, specPath); - debug(`Starting workflow (spec: ${specificationPath})`); + // Validate mutually exclusive options + if (options.opencodeServer && options.opencodeAttach) { + console.error( + "Error: --opencode-server and --opencode-attach are mutually exclusive", + ); + process.exit(1); + } - // Comprehensive terminal clearing - clearTerminal(); + debug(`Starting workflow (spec: ${specificationPath})`); + if (options.opencodeServer) { + debug("OpenCode consolidated server mode enabled (experimental)"); + } + if (options.opencodeAttach) { + debug( + `Attaching to external OpenCode server: ${options.opencodeAttach}`, + ); + } - // Determine execution method based on environment: - // - Dev mode: Import and run workflow directly (no SolidJS preload in dev) - // - Production: Spawn workflow binary (prevents JSX conflicts) - const isDev = import.meta.url.includes('/src/') + // Comprehensive terminal clearing + clearTerminal(); - if (isDev) { - // Development mode - directly import and run (SolidJS preload not active) - const { runWorkflowQueue } = await import('../../workflows/index.js'); - const { ValidationError } = await import('../../runtime/services/validation.js'); - try { - await runWorkflowQueue({ cwd, specificationPath }); - console.log('\n✓ Workflow completed successfully'); - process.exit(0); - } catch (error) { - // Show friendly instructional message for validation errors (no stack trace) - if (error instanceof ValidationError) { - console.log(`\n${error.message}\n`); - process.exit(1); - } - // Show detailed error for other failures - console.error('\n✗ Workflow failed:', error instanceof Error ? error.message : String(error)); - process.exit(1); - } - } else { - // Production mode - spawn workflow binary to avoid JSX conflicts - // The main binary has SolidJS transform, so we must use separate workflow binary - const { spawnProcess } = await import('../../infra/process/spawn.js'); - const { resolveWorkflowBinary } = await import('../../shared/utils/resolve-workflow-binary.js'); + // Determine execution method based on environment: + // - Dev mode: Import and run workflow directly (no SolidJS preload in dev) + // - Production: Spawn workflow binary (prevents JSX conflicts) + const isDev = import.meta.url.includes("/src/"); - try { - const result = await spawnProcess({ - command: resolveWorkflowBinary(), - args: [cwd, specificationPath], - // Pass CODEMACHINE_INSTALL_DIR from parent process to child - env: process.env.CODEMACHINE_INSTALL_DIR ? { - CODEMACHINE_INSTALL_DIR: process.env.CODEMACHINE_INSTALL_DIR - } : undefined, - stdioMode: 'inherit', // Let workflow take full terminal control - }); + if (isDev) { + // Development mode - directly import and run (SolidJS preload not active) + const { runWorkflowQueue } = await import("../../workflows/index.js"); + const { ValidationError } = await import( + "../../runtime/services/validation.js" + ); + try { + await runWorkflowQueue({ + cwd, + specificationPath, + opencodeServer: options.opencodeServer, + opencodeAttach: options.opencodeAttach, + }); + console.log("\n✓ Workflow completed successfully"); + process.exit(0); + } catch (error) { + // Show friendly instructional message for validation errors (no stack trace) + if (error instanceof ValidationError) { + console.log(`\n${error.message}\n`); + process.exit(1); + } + // Show detailed error for other failures + console.error( + "\n✗ Workflow failed:", + error instanceof Error ? error.message : String(error), + ); + process.exit(1); + } + } else { + // Production mode - spawn workflow binary to avoid JSX conflicts + // The main binary has SolidJS transform, so we must use separate workflow binary + const { spawnProcess } = await import("../../infra/process/spawn.js"); + const { resolveWorkflowBinary } = await import( + "../../shared/utils/resolve-workflow-binary.js" + ); - if (result.exitCode === 0) { - console.log('\n✓ Workflow completed successfully'); - process.exit(0); - } else { - console.error(`\n✗ Workflow failed with exit code ${result.exitCode}`); - process.exit(result.exitCode); - } - } catch (error) { - console.error('\n✗ Failed to spawn workflow:', error instanceof Error ? error.message : String(error)); - process.exit(1); - } - } - }); + try { + // Build environment with optional OpenCode server settings + const spawnEnv: Record = {}; + if (process.env.CODEMACHINE_INSTALL_DIR) { + spawnEnv.CODEMACHINE_INSTALL_DIR = + process.env.CODEMACHINE_INSTALL_DIR; + } + if (options.opencodeServer) { + spawnEnv.CODEMACHINE_OPENCODE_SERVER = "1"; + } + if (options.opencodeAttach) { + spawnEnv.CODEMACHINE_OPENCODE_ATTACH = options.opencodeAttach; + } + + const result = await spawnProcess({ + command: resolveWorkflowBinary(), + args: [cwd, specificationPath], + env: Object.keys(spawnEnv).length > 0 ? spawnEnv : undefined, + stdioMode: "inherit", // Let workflow take full terminal control + }); + + if (result.exitCode === 0) { + console.log("\n✓ Workflow completed successfully"); + process.exit(0); + } else { + console.error( + `\n✗ Workflow failed with exit code ${result.exitCode}`, + ); + process.exit(result.exitCode); + } + } catch (error) { + console.error( + "\n✗ Failed to spawn workflow:", + error instanceof Error ? error.message : String(error), + ); + process.exit(1); + } + } + }); } diff --git a/src/infra/engines/core/types.ts b/src/infra/engines/core/types.ts index c7ceafc4..8cdcb4ed 100644 --- a/src/infra/engines/core/types.ts +++ b/src/infra/engines/core/types.ts @@ -9,36 +9,46 @@ export type EngineType = string; export interface ParsedTelemetry { - tokensIn: number; - tokensOut: number; - cached?: number; - cost?: number; - duration?: number; - cacheCreationTokens?: number; - cacheReadTokens?: number; + tokensIn: number; + tokensOut: number; + cached?: number; + cost?: number; + duration?: number; + cacheCreationTokens?: number; + cacheReadTokens?: number; } export interface EngineRunOptions { - prompt: string; - workingDir: string; - model?: string; - modelReasoningEffort?: 'low' | 'medium' | 'high'; - env?: NodeJS.ProcessEnv; - onData?: (chunk: string) => void; - onErrorData?: (chunk: string) => void; - onTelemetry?: (telemetry: ParsedTelemetry) => void; - abortSignal?: AbortSignal; - timeout?: number; + prompt: string; + workingDir: string; + model?: string; + modelReasoningEffort?: "low" | "medium" | "high"; + /** + * Agent name for engines that support agent selection (e.g., OpenCode --agent flag) + */ + agent?: string; + /** + * URL of running server to attach to (e.g., http://localhost:4096) + * For OpenCode, uses --attach flag to connect to existing server. + * This avoids MCP server cold boot times on every run. + */ + attach?: string; + env?: NodeJS.ProcessEnv; + onData?: (chunk: string) => void; + onErrorData?: (chunk: string) => void; + onTelemetry?: (telemetry: ParsedTelemetry) => void; + abortSignal?: AbortSignal; + timeout?: number; } export interface EngineRunResult { - stdout: string; - stderr: string; + stdout: string; + stderr: string; } export interface Engine { - type: EngineType; - run(options: EngineRunOptions): Promise; + type: EngineType; + run(options: EngineRunOptions): Promise; } /** @@ -46,7 +56,7 @@ export interface Engine { * Basic validation - registry-specific validation should be done at runtime */ export function isValidEngineType(type: string): boolean { - return typeof type === 'string' && type.length > 0; + return typeof type === "string" && type.length > 0; } /** @@ -54,8 +64,10 @@ export function isValidEngineType(type: string): boolean { * Note: This basic validation can be enhanced when registry access is needed */ export function normalizeEngineType(type: string): string { - if (isValidEngineType(type)) { - return type; - } - throw new Error(`Invalid engine type "${type}". Engine type must be a non-empty string.`); + if (isValidEngineType(type)) { + return type; + } + throw new Error( + `Invalid engine type "${type}". Engine type must be a non-empty string.`, + ); } diff --git a/src/infra/engines/providers/opencode/execution/commands.ts b/src/infra/engines/providers/opencode/execution/commands.ts index 056d7fc4..eaf5660c 100644 --- a/src/infra/engines/providers/opencode/execution/commands.ts +++ b/src/infra/engines/providers/opencode/execution/commands.ts @@ -1,33 +1,71 @@ export interface OpenCodeCommandOptions { - /** - * Provider/model identifier (e.g., anthropic/claude-3.7-sonnet) - */ - model?: string; - /** - * Agent name to run (defaults to 'build') - */ - agent?: string; + /** + * Provider/model identifier (e.g., anthropic/claude-3.7-sonnet) + */ + model?: string; + /** + * Agent name to run (defaults to 'build') + */ + agent?: string; + /** + * URL of running OpenCode server to attach to (e.g., http://localhost:4096) + * When provided, uses --attach flag to connect to existing server instead of spawning new process. + * This avoids MCP server cold boot times on every run. + */ + attach?: string; } export interface OpenCodeCommand { - command: string; - args: string[]; + command: string; + args: string[]; } -export function buildOpenCodeRunCommand(options: OpenCodeCommandOptions = {}): OpenCodeCommand { - const args: string[] = ['run', '--format', 'json']; +/** + * Build command for `opencode run` + */ +export function buildOpenCodeRunCommand( + options: OpenCodeCommandOptions = {}, +): OpenCodeCommand { + const args: string[] = ["run", "--format", "json"]; - const agentName = options.agent?.trim() || 'build'; - if (agentName) { - args.push('--agent', agentName); - } + // Attach to existing server if URL provided + if (options.attach?.trim()) { + args.push("--attach", options.attach.trim()); + } - if (options.model?.trim()) { - args.push('--model', options.model.trim()); - } + const agentName = options.agent?.trim() || "build"; + if (agentName) { + args.push("--agent", agentName); + } - return { - command: 'opencode', - args, - }; + if (options.model?.trim()) { + args.push("--model", options.model.trim()); + } + + return { + command: "opencode", + args, + }; +} + +/** + * Build command for `opencode serve` + */ +export function buildOpenCodeServeCommand( + options: { port?: number; hostname?: string } = {}, +): OpenCodeCommand { + const args: string[] = ["serve"]; + + if (options.port) { + args.push("--port", options.port.toString()); + } + + if (options.hostname?.trim()) { + args.push("--hostname", options.hostname.trim()); + } + + return { + command: "opencode", + args, + }; } diff --git a/src/infra/engines/providers/opencode/execution/index.ts b/src/infra/engines/providers/opencode/execution/index.ts index c6d3990c..7bb32581 100644 --- a/src/infra/engines/providers/opencode/execution/index.ts +++ b/src/infra/engines/providers/opencode/execution/index.ts @@ -1,3 +1,4 @@ -export * from './commands.js'; -export * from './executor.js'; -export * from './runner.js'; +export * from "./commands.js"; +export * from "./executor.js"; +export * from "./runner.js"; +export * from "./server.js"; diff --git a/src/infra/engines/providers/opencode/execution/runner.ts b/src/infra/engines/providers/opencode/execution/runner.ts index adc024d3..38e49fe0 100644 --- a/src/infra/engines/providers/opencode/execution/runner.ts +++ b/src/infra/engines/providers/opencode/execution/runner.ts @@ -1,346 +1,389 @@ -import * as path from 'node:path'; - -import { spawnProcess } from '../../../../process/spawn.js'; -import { buildOpenCodeRunCommand } from './commands.js'; -import { metadata } from '../metadata.js'; -import { resolveOpenCodeHome } from '../auth.js'; -import { formatCommand, formatResult, formatStatus, formatMessage } from '../../../../../shared/formatters/outputMarkers.js'; -import { logger } from '../../../../../shared/logging/index.js'; -import { createTelemetryCapture } from '../../../../../shared/telemetry/index.js'; -import type { ParsedTelemetry } from '../../../core/types.js'; +import * as path from "node:path"; + +import { spawnProcess } from "../../../../process/spawn.js"; +import { buildOpenCodeRunCommand } from "./commands.js"; +import { metadata } from "../metadata.js"; +import { resolveOpenCodeHome } from "../auth.js"; +import { + formatCommand, + formatResult, + formatStatus, + formatMessage, +} from "../../../../../shared/formatters/outputMarkers.js"; +import { logger } from "../../../../../shared/logging/index.js"; +import { createTelemetryCapture } from "../../../../../shared/telemetry/index.js"; +import type { ParsedTelemetry } from "../../../core/types.js"; export interface RunOpenCodeOptions { - prompt: string; - workingDir: string; - model?: string; - agent?: string; - env?: NodeJS.ProcessEnv; - onData?: (chunk: string) => void; - onErrorData?: (chunk: string) => void; - onTelemetry?: (telemetry: ParsedTelemetry) => void; - abortSignal?: AbortSignal; - timeout?: number; // Timeout in milliseconds (default: 1800000ms = 30 minutes) + prompt: string; + workingDir: string; + model?: string; + agent?: string; + /** + * URL of running OpenCode server to attach to (e.g., http://localhost:4096) + * When provided, uses --attach flag to connect to existing server instead of spawning new process. + * This avoids MCP server cold boot times on every run. + */ + attach?: string; + env?: NodeJS.ProcessEnv; + onData?: (chunk: string) => void; + onErrorData?: (chunk: string) => void; + onTelemetry?: (telemetry: ParsedTelemetry) => void; + abortSignal?: AbortSignal; + timeout?: number; // Timeout in milliseconds (default: 1800000ms = 30 minutes) } export interface RunOpenCodeResult { - stdout: string; - stderr: string; + stdout: string; + stderr: string; } -const ANSI_ESCAPE_SEQUENCE = new RegExp(String.raw`\u001B\[[0-9;?]*[ -/]*[@-~]`, 'g'); +const ANSI_ESCAPE_SEQUENCE = new RegExp( + String.raw`\u001B\[[0-9;?]*[ -/]*[@-~]`, + "g", +); -function shouldApplyDefault(key: string, overrides?: NodeJS.ProcessEnv): boolean { - return overrides?.[key] === undefined && process.env[key] === undefined; +function shouldApplyDefault( + key: string, + overrides?: NodeJS.ProcessEnv, +): boolean { + return overrides?.[key] === undefined && process.env[key] === undefined; } function resolveRunnerEnv(env?: NodeJS.ProcessEnv): NodeJS.ProcessEnv { - const runnerEnv: NodeJS.ProcessEnv = { ...process.env, ...(env ?? {}) }; + const runnerEnv: NodeJS.ProcessEnv = { ...process.env, ...(env ?? {}) }; - // Set all three XDG environment variables to subdirectories under OPENCODE_HOME - // This centralizes all OpenCode data under ~/.codemachine/opencode by default - const opencodeHome = resolveOpenCodeHome(runnerEnv.OPENCODE_HOME); + // Set all three XDG environment variables to subdirectories under OPENCODE_HOME + // This centralizes all OpenCode data under ~/.codemachine/opencode by default + const opencodeHome = resolveOpenCodeHome(runnerEnv.OPENCODE_HOME); - if (shouldApplyDefault('XDG_CONFIG_HOME', env)) { - runnerEnv.XDG_CONFIG_HOME = path.join(opencodeHome, 'config'); - } + if (shouldApplyDefault("XDG_CONFIG_HOME", env)) { + runnerEnv.XDG_CONFIG_HOME = path.join(opencodeHome, "config"); + } - if (shouldApplyDefault('XDG_CACHE_HOME', env)) { - runnerEnv.XDG_CACHE_HOME = path.join(opencodeHome, 'cache'); - } + if (shouldApplyDefault("XDG_CACHE_HOME", env)) { + runnerEnv.XDG_CACHE_HOME = path.join(opencodeHome, "cache"); + } - if (shouldApplyDefault('XDG_DATA_HOME', env)) { - runnerEnv.XDG_DATA_HOME = path.join(opencodeHome, 'data'); - } + if (shouldApplyDefault("XDG_DATA_HOME", env)) { + runnerEnv.XDG_DATA_HOME = path.join(opencodeHome, "data"); + } - return runnerEnv; + return runnerEnv; } const truncate = (value: string, length = 100): string => - value.length > length ? `${value.slice(0, length)}...` : value; + value.length > length ? `${value.slice(0, length)}...` : value; function cleanAnsi(text: string, plainLogs: boolean): string { - if (!plainLogs) return text; - return text.replace(ANSI_ESCAPE_SEQUENCE, ''); + if (!plainLogs) return text; + return text.replace(ANSI_ESCAPE_SEQUENCE, ""); } function formatToolUse(part: unknown, plainLogs: boolean): string { - // eslint-disable-next-line @typescript-eslint/no-explicit-any - const partObj = (typeof part === 'object' && part !== null ? part : {}) as Record; - const tool = partObj?.tool ?? 'tool'; - const base = formatCommand(tool, 'success'); - const state = partObj?.state ?? {}; - - if (tool === 'bash') { - const outputRaw = - typeof state?.output === 'string' - ? state.output - : state?.output - ? JSON.stringify(state.output) - : ''; - const output = cleanAnsi(outputRaw?.trim() ?? '', plainLogs); - if (output) { - return `${base}\n${formatResult(output, false)}`; - } - return base; - } - - const previewSource = - (typeof state?.title === 'string' && state.title.trim()) || - (typeof state?.output === 'string' && state.output.trim()) || - (state?.input && Object.keys(state.input).length > 0 ? JSON.stringify(state.input) : ''); - - if (previewSource) { - const preview = cleanAnsi(previewSource.trim(), plainLogs); - return `${base}\n${formatResult(truncate(preview), false)}`; - } - - return base; + // eslint-disable-next-line @typescript-eslint/no-explicit-any + const partObj = ( + typeof part === "object" && part !== null ? part : {} + ) as Record; + const tool = partObj?.tool ?? "tool"; + const base = formatCommand(tool, "success"); + const state = partObj?.state ?? {}; + + if (tool === "bash") { + const outputRaw = + typeof state?.output === "string" + ? state.output + : state?.output + ? JSON.stringify(state.output) + : ""; + const output = cleanAnsi(outputRaw?.trim() ?? "", plainLogs); + if (output) { + return `${base}\n${formatResult(output, false)}`; + } + return base; + } + + const previewSource = + (typeof state?.title === "string" && state.title.trim()) || + (typeof state?.output === "string" && state.output.trim()) || + (state?.input && Object.keys(state.input).length > 0 + ? JSON.stringify(state.input) + : ""); + + if (previewSource) { + const preview = cleanAnsi(previewSource.trim(), plainLogs); + return `${base}\n${formatResult(truncate(preview), false)}`; + } + + return base; } function formatStepEvent(type: string, part: unknown): string | null { - // eslint-disable-next-line @typescript-eslint/no-explicit-any - const partObj = (typeof part === 'object' && part !== null ? part : {}) as Record; - const reason = typeof partObj?.reason === 'string' ? partObj.reason : undefined; - - // Only show final step (reason: 'stop'), skip intermediate steps (reason: 'tool-calls') - if (reason !== 'stop') { - return null; - } - - const tokens = partObj?.tokens; - if (!tokens) { - return null; - } - - const cache = (tokens.cache?.read ?? 0) + (tokens.cache?.write ?? 0); - const totalIn = (tokens.input ?? 0) + cache; - const tokenSummary = `⏱️ Tokens: ${totalIn}in/${tokens.output ?? 0}out${cache > 0 ? ` (${cache} cached)` : ''}`; - - return tokenSummary; + // eslint-disable-next-line @typescript-eslint/no-explicit-any + const partObj = ( + typeof part === "object" && part !== null ? part : {} + ) as Record; + const reason = + typeof partObj?.reason === "string" ? partObj.reason : undefined; + + // Only show final step (reason: 'stop'), skip intermediate steps (reason: 'tool-calls') + if (reason !== "stop") { + return null; + } + + const tokens = partObj?.tokens; + if (!tokens) { + return null; + } + + const cache = (tokens.cache?.read ?? 0) + (tokens.cache?.write ?? 0); + const totalIn = (tokens.input ?? 0) + cache; + const tokenSummary = `⏱️ Tokens: ${totalIn}in/${tokens.output ?? 0}out${cache > 0 ? ` (${cache} cached)` : ""}`; + + return tokenSummary; } function formatErrorEvent(error: unknown, plainLogs: boolean): string { - // eslint-disable-next-line @typescript-eslint/no-explicit-any - const errorObj = (typeof error === 'object' && error !== null ? error : {}) as Record; - const dataMessage = - typeof errorObj?.data?.message === 'string' - ? errorObj.data.message - : typeof errorObj?.message === 'string' - ? errorObj.message - : typeof errorObj?.name === 'string' - ? errorObj.name - : 'OpenCode reported an unknown error'; - - const cleaned = cleanAnsi(dataMessage, plainLogs); - return `${formatCommand('OpenCode Error', 'error')}\n${formatResult(cleaned, true)}`; + // eslint-disable-next-line @typescript-eslint/no-explicit-any + const errorObj = ( + typeof error === "object" && error !== null ? error : {} + ) as Record; + const dataMessage = + typeof errorObj?.data?.message === "string" + ? errorObj.data.message + : typeof errorObj?.message === "string" + ? errorObj.message + : typeof errorObj?.name === "string" + ? errorObj.name + : "OpenCode reported an unknown error"; + + const cleaned = cleanAnsi(dataMessage, plainLogs); + return `${formatCommand("OpenCode Error", "error")}\n${formatResult(cleaned, true)}`; } -export async function runOpenCode(options: RunOpenCodeOptions): Promise { - const { - prompt, - workingDir, - model, - agent, - env, - onData, - onErrorData, - onTelemetry, - abortSignal, - timeout = 1800000, - } = options; - - if (!prompt) { - throw new Error('runOpenCode requires a prompt.'); - } - - if (!workingDir) { - throw new Error('runOpenCode requires a working directory.'); - } - - const runnerEnv = resolveRunnerEnv(env); - const plainLogs = - (env?.CODEMACHINE_PLAIN_LOGS ?? process.env.CODEMACHINE_PLAIN_LOGS ?? '').toString() === '1'; - const { command, args } = buildOpenCodeRunCommand({ model, agent }); - - logger.debug( - `OpenCode runner - prompt length: ${prompt.length}, lines: ${prompt.split('\n').length}, agent: ${ - agent ?? 'build' - }, model: ${model ?? 'default'}`, - ); - - const telemetryCapture = createTelemetryCapture('opencode', model, prompt, workingDir); - let jsonBuffer = ''; - let isFirstStep = true; - - const processLine = (line: string): void => { - if (!line.trim()) { - return; - } - - let parsed: unknown; - try { - parsed = JSON.parse(line); - } catch { - return; - } - - telemetryCapture.captureFromStreamJson(line); - - if (onTelemetry) { - const captured = telemetryCapture.getCaptured(); - if (captured?.tokens) { - const totalIn = - (captured.tokens.input ?? 0) + (captured.tokens.cached ?? 0); - onTelemetry({ - tokensIn: totalIn, - tokensOut: captured.tokens.output ?? 0, - cached: captured.tokens.cached, - cost: captured.cost, - duration: captured.duration, - }); - } - } - - // Type guard for parsed JSON - if (typeof parsed !== 'object' || parsed === null) { - return; - } - // eslint-disable-next-line @typescript-eslint/no-explicit-any - const parsedObj = parsed as Record; - - let formatted: string | null = null; - switch (parsedObj.type) { - case 'tool_use': - formatted = formatToolUse(parsedObj.part, plainLogs); - break; - case 'step_start': - if (isFirstStep) { - isFirstStep = false; - formatted = formatStatus('OpenCode is analyzing your request...'); - } - // Subsequent step_start events are silent - break; - case 'step_finish': - formatted = formatStepEvent(parsedObj.type, parsedObj.part); - break; - case 'text': { - const textPart = parsedObj.part; - const textValue = - typeof textPart?.text === 'string' - ? cleanAnsi(textPart.text, plainLogs) - : ''; - formatted = textValue ? formatMessage(textValue) : null; - break; - } - case 'error': - formatted = formatErrorEvent(parsedObj.error, plainLogs); - break; - default: - break; - } - - if (formatted) { - const suffix = formatted.endsWith('\n') ? '' : '\n'; - onData?.(formatted + suffix); - } - }; - - const normalizeChunk = (chunk: string): string => { - let result = chunk; - - // Convert line endings to \n - result = result.replace(/\r\n/g, '\n').replace(/\r/g, '\n'); - - // Handle carriage returns that cause line overwrites - result = result.replace(/^.*\r([^\r\n]*)/gm, '$1'); - - // Strip ANSI sequences in plain mode - if (plainLogs) { - result = result.replace(ANSI_ESCAPE_SEQUENCE, ''); - } - - // Collapse excessive newlines - result = result.replace(/\n{3,}/g, '\n\n'); - - return result; - }; - - let result; - try { - result = await spawnProcess({ - command, - args, - cwd: workingDir, - env: runnerEnv, - stdinInput: prompt, - stdioMode: 'pipe', - onStdout: (chunk) => { - const normalized = normalizeChunk(chunk); - jsonBuffer += normalized; - - const lines = jsonBuffer.split('\n'); - jsonBuffer = lines.pop() ?? ''; - - for (const line of lines) { - processLine(line); - } - }, - onStderr: (chunk) => { - const normalized = normalizeChunk(chunk); - const cleaned = cleanAnsi(normalized, plainLogs); - onErrorData?.(cleaned); - }, - signal: abortSignal, - timeout, - }); - } catch (error) { - const err = error as { code?: string; message?: string }; - const message = err?.message ?? ''; - const notFound = - err?.code === 'ENOENT' || - /not recognized as an internal or external command/i.test(message) || - /command not found/i.test(message); - - if (notFound) { - const installMessage = [ - `'${command}' is not available on this system.`, - 'Install OpenCode via:', - ' npm i -g opencode-ai@latest', - ' brew install opencode', - ' scoop bucket add extras && scoop install extras/opencode', - ' choco install opencode', - 'Docs: https://opencode.ai/docs', - ].join('\n'); - logger.error(`${metadata.name} CLI not found when executing: ${command} ${args.join(' ')}`); - throw new Error(installMessage); - } - - throw error; - } - - if (jsonBuffer.trim()) { - processLine(jsonBuffer); - jsonBuffer = ''; - } - - if (result.exitCode !== 0) { - const stderr = result.stderr.trim(); - const stdout = result.stdout.trim(); - const sample = (stderr || stdout || 'no error output').split('\n').slice(0, 10).join('\n'); - - logger.error('OpenCode CLI execution failed', { - exitCode: result.exitCode, - sample, - command: `${command} ${args.join(' ')}`, - }); - - throw new Error(`OpenCode CLI exited with code ${result.exitCode}`); - } - - telemetryCapture.logCapturedTelemetry(result.exitCode); - - return { - stdout: result.stdout, - stderr: result.stderr, - }; +export async function runOpenCode( + options: RunOpenCodeOptions, +): Promise { + const { + prompt, + workingDir, + model, + agent, + attach, + env, + onData, + onErrorData, + onTelemetry, + abortSignal, + timeout = 1800000, + } = options; + + if (!prompt) { + throw new Error("runOpenCode requires a prompt."); + } + + if (!workingDir) { + throw new Error("runOpenCode requires a working directory."); + } + + const runnerEnv = resolveRunnerEnv(env); + const plainLogs = + ( + env?.CODEMACHINE_PLAIN_LOGS ?? + process.env.CODEMACHINE_PLAIN_LOGS ?? + "" + ).toString() === "1"; + const { command, args } = buildOpenCodeRunCommand({ model, agent, attach }); + + logger.debug( + `OpenCode runner - prompt length: ${prompt.length}, lines: ${prompt.split("\n").length}, agent: ${ + agent ?? "build" + }, model: ${model ?? "default"}${attach ? `, attach: ${attach}` : ""}`, + ); + + const telemetryCapture = createTelemetryCapture( + "opencode", + model, + prompt, + workingDir, + ); + let jsonBuffer = ""; + let isFirstStep = true; + + const processLine = (line: string): void => { + if (!line.trim()) { + return; + } + + let parsed: unknown; + try { + parsed = JSON.parse(line); + } catch { + return; + } + + telemetryCapture.captureFromStreamJson(line); + + if (onTelemetry) { + const captured = telemetryCapture.getCaptured(); + if (captured?.tokens) { + const totalIn = + (captured.tokens.input ?? 0) + (captured.tokens.cached ?? 0); + onTelemetry({ + tokensIn: totalIn, + tokensOut: captured.tokens.output ?? 0, + cached: captured.tokens.cached, + cost: captured.cost, + duration: captured.duration, + }); + } + } + + // Type guard for parsed JSON + if (typeof parsed !== "object" || parsed === null) { + return; + } + // eslint-disable-next-line @typescript-eslint/no-explicit-any + const parsedObj = parsed as Record; + + let formatted: string | null = null; + switch (parsedObj.type) { + case "tool_use": + formatted = formatToolUse(parsedObj.part, plainLogs); + break; + case "step_start": + if (isFirstStep) { + isFirstStep = false; + formatted = formatStatus("OpenCode is analyzing your request..."); + } + // Subsequent step_start events are silent + break; + case "step_finish": + formatted = formatStepEvent(parsedObj.type, parsedObj.part); + break; + case "text": { + const textPart = parsedObj.part; + const textValue = + typeof textPart?.text === "string" + ? cleanAnsi(textPart.text, plainLogs) + : ""; + formatted = textValue ? formatMessage(textValue) : null; + break; + } + case "error": + formatted = formatErrorEvent(parsedObj.error, plainLogs); + break; + default: + break; + } + + if (formatted) { + const suffix = formatted.endsWith("\n") ? "" : "\n"; + onData?.(formatted + suffix); + } + }; + + const normalizeChunk = (chunk: string): string => { + let result = chunk; + + // Convert line endings to \n + result = result.replace(/\r\n/g, "\n").replace(/\r/g, "\n"); + + // Handle carriage returns that cause line overwrites + result = result.replace(/^.*\r([^\r\n]*)/gm, "$1"); + + // Strip ANSI sequences in plain mode + if (plainLogs) { + result = result.replace(ANSI_ESCAPE_SEQUENCE, ""); + } + + // Collapse excessive newlines + result = result.replace(/\n{3,}/g, "\n\n"); + + return result; + }; + + let result; + try { + result = await spawnProcess({ + command, + args, + cwd: workingDir, + env: runnerEnv, + stdinInput: prompt, + stdioMode: "pipe", + onStdout: (chunk) => { + const normalized = normalizeChunk(chunk); + jsonBuffer += normalized; + + const lines = jsonBuffer.split("\n"); + jsonBuffer = lines.pop() ?? ""; + + for (const line of lines) { + processLine(line); + } + }, + onStderr: (chunk) => { + const normalized = normalizeChunk(chunk); + const cleaned = cleanAnsi(normalized, plainLogs); + onErrorData?.(cleaned); + }, + signal: abortSignal, + timeout, + }); + } catch (error) { + const err = error as { code?: string; message?: string }; + const message = err?.message ?? ""; + const notFound = + err?.code === "ENOENT" || + /not recognized as an internal or external command/i.test(message) || + /command not found/i.test(message); + + if (notFound) { + const installMessage = [ + `'${command}' is not available on this system.`, + "Install OpenCode via:", + " npm i -g opencode-ai@latest", + " brew install opencode", + " scoop bucket add extras && scoop install extras/opencode", + " choco install opencode", + "Docs: https://opencode.ai/docs", + ].join("\n"); + logger.error( + `${metadata.name} CLI not found when executing: ${command} ${args.join(" ")}`, + ); + throw new Error(installMessage); + } + + throw error; + } + + if (jsonBuffer.trim()) { + processLine(jsonBuffer); + jsonBuffer = ""; + } + + if (result.exitCode !== 0) { + const stderr = result.stderr.trim(); + const stdout = result.stdout.trim(); + const sample = (stderr || stdout || "no error output") + .split("\n") + .slice(0, 10) + .join("\n"); + + logger.error("OpenCode CLI execution failed", { + exitCode: result.exitCode, + sample, + command: `${command} ${args.join(" ")}`, + }); + + throw new Error(`OpenCode CLI exited with code ${result.exitCode}`); + } + + telemetryCapture.logCapturedTelemetry(result.exitCode); + + return { + stdout: result.stdout, + stderr: result.stderr, + }; } diff --git a/src/infra/engines/providers/opencode/execution/server.ts b/src/infra/engines/providers/opencode/execution/server.ts new file mode 100644 index 00000000..0f0812e7 --- /dev/null +++ b/src/infra/engines/providers/opencode/execution/server.ts @@ -0,0 +1,285 @@ +import { spawn, type ChildProcess } from "node:child_process"; +import { createServer } from "node:net"; +import { buildOpenCodeServeCommand } from "./commands.js"; +import { + info, + error as logError, + debug, +} from "../../../../../shared/logging/logger.js"; + +/** + * Agent information returned by OpenCode server + */ +export interface OpenCodeAgent { + id: string; + name?: string; + description?: string; +} + +export interface OpenCodeServerOptions { + /** + * Port to listen on. If not provided, finds an available port. + */ + port?: number; + /** + * Hostname to listen on. Defaults to 127.0.0.1. + */ + hostname?: string; + /** + * Working directory for the server process. + */ + workingDir?: string; + /** + * Timeout in ms to wait for server to be ready. Defaults to 30000 (30s). + */ + startupTimeout?: number; +} + +export interface OpenCodeServer { + /** + * URL to connect to (e.g., http://127.0.0.1:4096) + */ + url: string; + /** + * Port the server is listening on + */ + port: number; + /** + * Hostname the server is listening on + */ + hostname: string; + /** + * Stop the server + */ + stop: () => Promise; + /** + * Check if server is healthy + */ + isHealthy: () => Promise; + /** + * List available agents from the server + */ + listAgents: () => Promise; +} + +/** + * Find an available port by attempting to bind to port 0 + */ +async function findAvailablePort(): Promise { + return new Promise((resolve, reject) => { + const server = createServer(); + server.listen(0, "127.0.0.1", () => { + const address = server.address(); + if (address && typeof address === "object") { + const port = address.port; + server.close(() => resolve(port)); + } else { + server.close(() => reject(new Error("Could not determine port"))); + } + }); + server.on("error", reject); + }); +} + +/** + * Wait for the server to be ready by polling the /app endpoint + */ +async function waitForServer(url: string, timeoutMs: number): Promise { + const startTime = Date.now(); + const pollInterval = 500; + + while (Date.now() - startTime < timeoutMs) { + try { + const response = await fetch(`${url}/app`, { + method: "GET", + signal: AbortSignal.timeout(2000), + }); + if (response.ok) { + return true; + } + } catch { + // Server not ready yet, continue polling + } + await new Promise((resolve) => setTimeout(resolve, pollInterval)); + } + return false; +} + +/** + * Check if an OpenCode server is healthy at the given URL + */ +export async function checkServerHealth(url: string): Promise { + try { + const response = await fetch(`${url}/app`, { + method: "GET", + signal: AbortSignal.timeout(5000), + }); + return response.ok; + } catch { + return false; + } +} + +/** + * List available agents from an OpenCode server + */ +export async function listServerAgents(url: string): Promise { + try { + const response = await fetch(`${url}/agent`, { + method: "GET", + signal: AbortSignal.timeout(10000), + }); + if (!response.ok) { + throw new Error( + `Failed to list agents: ${response.status} ${response.statusText}`, + ); + } + const data = await response.json(); + // The API returns an array of agents + return Array.isArray(data) ? data : []; + } catch (err) { + debug(`Failed to list agents from ${url}: ${err}`); + return []; + } +} + +/** + * Start an OpenCode server process + * + * @example + * ```typescript + * const server = await startOpenCodeServer({ port: 4096 }); + * console.log(`Server running at ${server.url}`); + * + * // Use server.url with --attach flag + * // opencode run --attach http://127.0.0.1:4096 "prompt" + * + * // When done + * await server.stop(); + * ``` + */ +export async function startOpenCodeServer( + options: OpenCodeServerOptions = {}, +): Promise { + const hostname = options.hostname ?? "127.0.0.1"; + const port = options.port ?? (await findAvailablePort()); + const startupTimeout = options.startupTimeout ?? 30000; + + const { command, args } = buildOpenCodeServeCommand({ port, hostname }); + + debug(`Starting OpenCode server: ${command} ${args.join(" ")}`); + + let serverProcess: ChildProcess | null = null; + + try { + serverProcess = spawn(command, args, { + cwd: options.workingDir ?? process.cwd(), + stdio: ["ignore", "pipe", "pipe"], + detached: false, + env: { + ...process.env, + // Ensure opencode doesn't try to use a TTY + NO_COLOR: "1", + }, + }); + + // Capture stderr for debugging + let stderrOutput = ""; + serverProcess.stderr?.on("data", (chunk) => { + stderrOutput += chunk.toString(); + debug(`[opencode serve stderr] ${chunk.toString().trim()}`); + }); + + // Handle process exit + const exitPromise = new Promise((_, reject) => { + serverProcess?.on("exit", (code, signal) => { + reject( + new Error( + `OpenCode server exited unexpectedly (code=${code}, signal=${signal}). stderr: ${stderrOutput}`, + ), + ); + }); + serverProcess?.on("error", (err) => { + reject(new Error(`Failed to start OpenCode server: ${err.message}`)); + }); + }); + + const url = `http://${hostname}:${port}`; + + // Wait for server to be ready or exit + const isReady = await Promise.race([ + waitForServer(url, startupTimeout), + exitPromise, + ]); + + if (!isReady) { + serverProcess.kill("SIGTERM"); + throw new Error( + `OpenCode server failed to start within ${startupTimeout}ms`, + ); + } + + info(`OpenCode server started at ${url}`); + + const server: OpenCodeServer = { + url, + port, + hostname, + stop: async () => { + if (serverProcess && !serverProcess.killed) { + debug(`Stopping OpenCode server at ${url}`); + serverProcess.kill("SIGTERM"); + + // Wait for graceful shutdown + await new Promise((resolve) => { + const timeout = setTimeout(() => { + if (serverProcess && !serverProcess.killed) { + serverProcess.kill("SIGKILL"); + } + resolve(); + }, 5000); + + serverProcess?.on("exit", () => { + clearTimeout(timeout); + resolve(); + }); + }); + + info(`OpenCode server stopped`); + } + }, + isHealthy: async () => checkServerHealth(url), + listAgents: async () => listServerAgents(url), + }; + + return server; + } catch (err) { + // Clean up on error + if (serverProcess && !serverProcess.killed) { + serverProcess.kill("SIGKILL"); + } + throw err; + } +} + +/** + * Create a server handle for an existing external server (no lifecycle management) + */ +export function attachToExternalServer(url: string): OpenCodeServer { + // Parse URL to extract hostname and port + const parsed = new URL(url); + const hostname = parsed.hostname; + const port = parseInt(parsed.port || "4096", 10); + + return { + url: url.replace(/\/$/, ""), // Remove trailing slash + port, + hostname, + stop: async () => { + // External server - we don't manage its lifecycle + debug(`Not stopping external OpenCode server at ${url}`); + }, + isHealthy: async () => checkServerHealth(url), + listAgents: async () => listServerAgents(url), + }; +} diff --git a/src/shared/workflows/template.ts b/src/shared/workflows/template.ts index ea2864f9..cb4aa945 100644 --- a/src/shared/workflows/template.ts +++ b/src/shared/workflows/template.ts @@ -1,62 +1,72 @@ -import { readFile, writeFile } from 'node:fs/promises'; -import { existsSync } from 'node:fs'; -import * as path from 'node:path'; -import { resolvePackageRoot } from '../runtime/pkg.js'; +import { readFile, writeFile } from "node:fs/promises"; +import { existsSync } from "node:fs"; +import * as path from "node:path"; +import { resolvePackageRoot } from "../runtime/pkg.js"; -const TEMPLATE_TRACKING_FILE = 'template.json'; +const TEMPLATE_TRACKING_FILE = "template.json"; -const packageRoot = resolvePackageRoot(import.meta.url, 'workflows template tracking'); +const packageRoot = resolvePackageRoot( + import.meta.url, + "workflows template tracking", +); -const templatesDir = path.resolve(packageRoot, 'templates', 'workflows'); +const templatesDir = path.resolve(packageRoot, "templates", "workflows"); interface TemplateTracking { - activeTemplate: string; - /** - * Timestamp in ISO 8601 format with UTC timezone (e.g., "2025-10-13T14:40:14.123Z"). - * The "Z" suffix explicitly indicates UTC timezone. - * To convert to local time in JavaScript: new Date(lastUpdated).toLocaleString() - */ - lastUpdated: string; - completedSteps?: number[]; - notCompletedSteps?: number[]; - resumeFromLastStep?: boolean; + activeTemplate: string; + /** + * Timestamp in ISO 8601 format with UTC timezone (e.g., "2025-10-13T14:40:14.123Z"). + * The "Z" suffix explicitly indicates UTC timezone. + * To convert to local time in JavaScript: new Date(lastUpdated).toLocaleString() + */ + lastUpdated: string; + completedSteps?: number[]; + notCompletedSteps?: number[]; + resumeFromLastStep?: boolean; } /** * Gets the currently active template name from the tracking file. */ -export async function getActiveTemplate(cmRoot: string): Promise { - const trackingPath = path.join(cmRoot, TEMPLATE_TRACKING_FILE); - - if (!existsSync(trackingPath)) { - return null; - } - - try { - const content = await readFile(trackingPath, 'utf8'); - const data = JSON.parse(content) as TemplateTracking; - return data.activeTemplate ?? null; - } catch (error) { - console.warn(`Failed to read active template from tracking file: ${error instanceof Error ? error.message : String(error)}`); - return null; - } +export async function getActiveTemplate( + cmRoot: string, +): Promise { + const trackingPath = path.join(cmRoot, TEMPLATE_TRACKING_FILE); + + if (!existsSync(trackingPath)) { + return null; + } + + try { + const content = await readFile(trackingPath, "utf8"); + const data = JSON.parse(content) as TemplateTracking; + return data.activeTemplate ?? null; + } catch (error) { + console.warn( + `Failed to read active template from tracking file: ${error instanceof Error ? error.message : String(error)}`, + ); + return null; + } } /** * Sets the active template name in the tracking file. */ -export async function setActiveTemplate(cmRoot: string, templateName: string): Promise { - const trackingPath = path.join(cmRoot, TEMPLATE_TRACKING_FILE); - - const data: TemplateTracking = { - activeTemplate: templateName, - lastUpdated: new Date().toISOString(), // ISO 8601 UTC format (e.g., "2025-10-13T14:40:14.123Z") - completedSteps: [], - notCompletedSteps: [], - resumeFromLastStep: true, - }; - - await writeFile(trackingPath, JSON.stringify(data, null, 2), 'utf8'); +export async function setActiveTemplate( + cmRoot: string, + templateName: string, +): Promise { + const trackingPath = path.join(cmRoot, TEMPLATE_TRACKING_FILE); + + const data: TemplateTracking = { + activeTemplate: templateName, + lastUpdated: new Date().toISOString(), // ISO 8601 UTC format (e.g., "2025-10-13T14:40:14.123Z") + completedSteps: [], + notCompletedSteps: [], + resumeFromLastStep: true, + }; + + await writeFile(trackingPath, JSON.stringify(data, null, 2), "utf8"); } /** @@ -65,30 +75,47 @@ export async function setActiveTemplate(cmRoot: string, templateName: string): P * - There is an active template and it's different from the provided one * - There is no active template but we have a new one (first run with template) */ -export async function hasTemplateChanged(cmRoot: string, templateName: string): Promise { - const activeTemplate = await getActiveTemplate(cmRoot); - - // If no active template, treat it as changed (first run with template should regenerate) - if (activeTemplate === null) { - return true; - } - - // Check if the template is different - return activeTemplate !== templateName; +export async function hasTemplateChanged( + cmRoot: string, + templateName: string, +): Promise { + const activeTemplate = await getActiveTemplate(cmRoot); + + // If no active template, treat it as changed (first run with template should regenerate) + if (activeTemplate === null) { + return true; + } + + // Check if the template is different + return activeTemplate !== templateName; } /** * Gets the full template path from the tracking file. * Returns the default template if no template is tracked. + * Supports both absolute paths and relative paths (relative to CodeMachine templates dir). */ -export async function getTemplatePathFromTracking(cmRoot: string): Promise { - const activeTemplate = await getActiveTemplate(cmRoot); - - if (!activeTemplate) { - // No template tracked, return default - return path.join(templatesDir, 'default.workflow.js'); - } - - // Return full path from template name - return path.join(templatesDir, activeTemplate); +export async function getTemplatePathFromTracking( + cmRoot: string, +): Promise { + const activeTemplate = await getActiveTemplate(cmRoot); + + if (!activeTemplate) { + // No template tracked, return default + return path.join(templatesDir, "default.workflow.js"); + } + + // If absolute path, use it directly + if (path.isAbsolute(activeTemplate)) { + return activeTemplate; + } + + // Check if template exists in project's .codemachine directory first + const projectTemplatePath = path.join(cmRoot, activeTemplate); + if (existsSync(projectTemplatePath)) { + return projectTemplatePath; + } + + // Fall back to CodeMachine's templates directory + return path.join(templatesDir, activeTemplate); } diff --git a/src/workflows/execution/step.ts b/src/workflows/execution/step.ts index 818f7275..82b41292 100644 --- a/src/workflows/execution/step.ts +++ b/src/workflows/execution/step.ts @@ -1,35 +1,35 @@ -import * as path from 'node:path'; -import { readFile, mkdir } from 'node:fs/promises'; -import type { WorkflowStep } from '../templates/index.js'; -import { isModuleStep } from '../templates/types.js'; -import type { EngineType } from '../../infra/engines/index.js'; -import { processPromptString } from '../../shared/prompts/index.js'; -import { executeAgent } from '../../agents/runner/runner.js'; -import type { WorkflowUIManager } from '../../ui/index.js'; +import * as path from "node:path"; +import { readFile, mkdir } from "node:fs/promises"; +import type { WorkflowStep } from "../templates/index.js"; +import { isModuleStep } from "../templates/types.js"; +import type { EngineType } from "../../infra/engines/index.js"; +import { processPromptString } from "../../shared/prompts/index.js"; +import { executeAgent } from "../../agents/runner/runner.js"; +import type { WorkflowUIManager } from "../../ui/index.js"; export interface StepExecutorOptions { - logger: (chunk: string) => void; - stderrLogger: (chunk: string) => void; - timeout?: number; - ui?: WorkflowUIManager; - abortSignal?: AbortSignal; - /** Parent agent ID for tracking relationships */ - parentId?: number; - /** Disable monitoring (for special cases) */ - disableMonitoring?: boolean; - /** Unique agent ID for UI updates (includes step index) */ - uniqueAgentId?: string; + logger: (chunk: string) => void; + stderrLogger: (chunk: string) => void; + timeout?: number; + ui?: WorkflowUIManager; + abortSignal?: AbortSignal; + /** Parent agent ID for tracking relationships */ + parentId?: number; + /** Disable monitoring (for special cases) */ + disableMonitoring?: boolean; + /** Unique agent ID for UI updates (includes step index) */ + uniqueAgentId?: string; } async function ensureProjectScaffold(cwd: string): Promise { - const agentsDir = path.resolve(cwd, '.codemachine', 'agents'); - const planDir = path.resolve(cwd, '.codemachine', 'plan'); - await mkdir(agentsDir, { recursive: true }); - await mkdir(planDir, { recursive: true }); + const agentsDir = path.resolve(cwd, ".codemachine", "agents"); + const planDir = path.resolve(cwd, ".codemachine", "plan"); + await mkdir(agentsDir, { recursive: true }); + await mkdir(planDir, { recursive: true }); } async function runAgentsBuilderStep(cwd: string): Promise { - await ensureProjectScaffold(cwd); + await ensureProjectScaffold(cwd); } /** @@ -39,60 +39,77 @@ async function runAgentsBuilderStep(cwd: string): Promise { * after building the prompt. No duplication with runner.ts anymore. */ export async function executeStep( - step: WorkflowStep, - cwd: string, - options: StepExecutorOptions, + step: WorkflowStep, + cwd: string, + options: StepExecutorOptions, ): Promise { - // Only module steps can be executed - if (!isModuleStep(step)) { - throw new Error('Only module steps can be executed'); - } + // Only module steps can be executed + if (!isModuleStep(step)) { + throw new Error("Only module steps can be executed"); + } - // Load and process the prompt template - const promptPath = path.isAbsolute(step.promptPath) - ? step.promptPath - : path.resolve(cwd, step.promptPath); - const rawPrompt = await readFile(promptPath, 'utf8'); - const prompt = await processPromptString(rawPrompt, cwd); + // Load and process the prompt template + // Support either promptPath (file reference) or prompt (inline string) + let rawPrompt: string; + const stepWithPrompt = step as typeof step & { prompt?: string }; + if (stepWithPrompt.prompt) { + // Use inline prompt directly + rawPrompt = stepWithPrompt.prompt; + } else if (step.promptPath) { + // Load from file + const promptPath = path.isAbsolute(step.promptPath) + ? step.promptPath + : path.resolve(cwd, step.promptPath); + rawPrompt = await readFile(promptPath, "utf8"); + } else { + throw new Error( + `Step ${step.agentId} must have either promptPath or prompt`, + ); + } + const prompt = await processPromptString(rawPrompt, cwd); - // Use environment variable or default to 30 minutes (1800000ms) - const timeout = - options.timeout ?? - (process.env.CODEMACHINE_AGENT_TIMEOUT - ? Number.parseInt(process.env.CODEMACHINE_AGENT_TIMEOUT, 10) - : 1800000); + // Use environment variable or default to 30 minutes (1800000ms) + const timeout = + options.timeout ?? + (process.env.CODEMACHINE_AGENT_TIMEOUT + ? Number.parseInt(process.env.CODEMACHINE_AGENT_TIMEOUT, 10) + : 1800000); - // Determine engine: step override > default - const engineType: EngineType | undefined = step.engine; + // Determine engine: step override > default + const engineType: EngineType | undefined = step.engine; - // Execute via the unified execution runner - // Runner handles: auth, monitoring, engine execution, memory storage - const result = await executeAgent(step.agentId, prompt, { - workingDir: cwd, - engine: engineType, - model: step.model, - logger: options.logger, - stderrLogger: options.stderrLogger, - onTelemetry: options.ui && options.uniqueAgentId - ? (telemetry) => options.ui!.updateAgentTelemetry(options.uniqueAgentId!, telemetry) - : undefined, - parentId: options.parentId, - disableMonitoring: options.disableMonitoring, - abortSignal: options.abortSignal, - timeout, - ui: options.ui, - uniqueAgentId: options.uniqueAgentId, - }); + // Execute via the unified execution runner + // Runner handles: auth, monitoring, engine execution, memory storage + const result = await executeAgent(step.agentId, prompt, { + workingDir: cwd, + engine: engineType, + model: step.model, + agent: step.agent, // Pass OpenCode agent name for --agent flag + attach: step.attach, // Pass server URL for --attach flag (consolidated server mode) + logger: options.logger, + stderrLogger: options.stderrLogger, + onTelemetry: + options.ui && options.uniqueAgentId + ? (telemetry) => + options.ui!.updateAgentTelemetry(options.uniqueAgentId!, telemetry) + : undefined, + parentId: options.parentId, + disableMonitoring: options.disableMonitoring, + abortSignal: options.abortSignal, + timeout, + ui: options.ui, + uniqueAgentId: options.uniqueAgentId, + }); - // Run special post-execution steps - const agentName = step.agentName.toLowerCase(); - if (step.agentId === 'agents-builder' || agentName.includes('builder')) { - await runAgentsBuilderStep(cwd); - } + // Run special post-execution steps + const agentName = step.agentName.toLowerCase(); + if (step.agentId === "agents-builder" || agentName.includes("builder")) { + await runAgentsBuilderStep(cwd); + } - // NOTE: Telemetry is already updated via onTelemetry callback during streaming execution. - // DO NOT parse from final output - it would match the FIRST telemetry line (early/wrong values) - // instead of the LAST telemetry line (final/correct values), causing incorrect UI display. + // NOTE: Telemetry is already updated via onTelemetry callback during streaming execution. + // DO NOT parse from final output - it would match the FIRST telemetry line (early/wrong values) + // instead of the LAST telemetry line (final/correct values), causing incorrect UI display. - return result.output; + return result.output; } diff --git a/src/workflows/execution/trigger.ts b/src/workflows/execution/trigger.ts index 407ca43b..36dcf0b7 100644 --- a/src/workflows/execution/trigger.ts +++ b/src/workflows/execution/trigger.ts @@ -1,188 +1,228 @@ -import * as path from 'node:path'; -import type { EngineType } from '../../infra/engines/index.js'; -import { getEngine, registry } from '../../infra/engines/index.js'; -import { loadAgentConfig, loadAgentTemplate } from '../../agents/runner/index.js'; -import { MemoryAdapter } from '../../infra/fs/memory-adapter.js'; -import { MemoryStore } from '../../agents/index.js'; -import { processPromptString } from '../../shared/prompts/index.js'; -import type { WorkflowUIManager } from '../../ui/index.js'; -import { AgentMonitorService, AgentLoggerService } from '../../agents/monitoring/index.js'; +import * as path from "node:path"; +import type { EngineType } from "../../infra/engines/index.js"; +import { getEngine, registry } from "../../infra/engines/index.js"; +import { + loadAgentConfig, + loadAgentTemplate, +} from "../../agents/runner/index.js"; +import { MemoryAdapter } from "../../infra/fs/memory-adapter.js"; +import { MemoryStore } from "../../agents/index.js"; +import { processPromptString } from "../../shared/prompts/index.js"; +import type { WorkflowUIManager } from "../../ui/index.js"; +import { + AgentMonitorService, + AgentLoggerService, +} from "../../agents/monitoring/index.js"; export interface TriggerExecutionOptions { - triggerAgentId: string; - cwd: string; - engineType: EngineType; - logger: (chunk: string) => void; - stderrLogger: (chunk: string) => void; - sourceAgentId: string; // The agent that triggered this execution - ui?: WorkflowUIManager; - abortSignal?: AbortSignal; - /** Disable monitoring (for special cases) */ - disableMonitoring?: boolean; + triggerAgentId: string; + cwd: string; + engineType: EngineType; + logger: (chunk: string) => void; + stderrLogger: (chunk: string) => void; + sourceAgentId: string; // The agent that triggered this execution + ui?: WorkflowUIManager; + abortSignal?: AbortSignal; + /** Disable monitoring (for special cases) */ + disableMonitoring?: boolean; } /** * Executes a triggered agent by loading it from config/main.agents.js * This bypasses the workflow and allows triggering any agent, even outside the workflow */ -export async function executeTriggerAgent(options: TriggerExecutionOptions): Promise { - const { triggerAgentId, cwd, engineType, logger: _logger, stderrLogger: _stderrLogger, sourceAgentId, ui, abortSignal, disableMonitoring } = options; - - // Initialize monitoring (unless explicitly disabled) - declare outside try for catch block access - const monitor = !disableMonitoring ? AgentMonitorService.getInstance() : null; - const loggerService = !disableMonitoring ? AgentLoggerService.getInstance() : null; - let monitoringAgentId: number | undefined; - - try { - // Load agent config and template from config/main.agents.js - const triggeredAgentConfig = await loadAgentConfig(triggerAgentId, cwd); - const rawTemplate = await loadAgentTemplate(triggerAgentId, cwd); - const triggeredAgentTemplate = await processPromptString(rawTemplate, cwd); - - // Get engine and resolve model/reasoning - const engine = getEngine(engineType); - const engineModule = registry.get(engineType); - const triggeredModel = (triggeredAgentConfig.model as string | undefined) ?? engineModule?.metadata.defaultModel; - const triggeredReasoning = (triggeredAgentConfig.modelReasoningEffort as 'low' | 'medium' | 'high' | undefined) ?? engineModule?.metadata.defaultModelReasoningEffort; - - // Find parent agent in monitoring system by sourceAgentId - let parentMonitoringId: number | undefined; - if (monitor) { - const parentAgents = monitor.queryAgents({ name: sourceAgentId }); - if (parentAgents.length > 0) { - // Get the most recent one (highest ID) - parentMonitoringId = parentAgents.sort((a, b) => b.id - a.id)[0].id; - } - - // Register triggered agent with parent relationship and engine/model info - const promptText = `Triggered by ${sourceAgentId}`; - monitoringAgentId = await monitor.register({ - name: triggerAgentId, - prompt: promptText, - parentId: parentMonitoringId, - engineProvider: engineType, - modelName: triggeredModel, - }); - - // Store full prompt for debug mode logging - if (loggerService && monitoringAgentId !== undefined) { - loggerService.storeFullPrompt(monitoringAgentId, promptText); - } - - // Register monitoring ID with UI immediately so it can load logs - if (ui && monitoringAgentId !== undefined) { - ui.registerMonitoringId(triggerAgentId, monitoringAgentId); - } - } - - // Add triggered agent to UI - if (ui) { - const engineName = engineType; // preserve original engine type, even if unknown - ui.addTriggeredAgent(sourceAgentId, { - id: triggerAgentId, - name: triggeredAgentConfig.name ?? triggerAgentId, - engine: engineName, - status: 'running', - triggeredBy: sourceAgentId, - startTime: Date.now(), - telemetry: { tokensIn: 0, tokensOut: 0 }, - toolCount: 0, - thinkingCount: 0, - }); - } - - if (ui) { - ui.logMessage(sourceAgentId, `Executing triggered agent: ${triggeredAgentConfig.name ?? triggerAgentId}`); - ui.logMessage(triggerAgentId, '═'.repeat(80)); - ui.logMessage(triggerAgentId, `${triggeredAgentConfig.name ?? triggerAgentId} started to work (triggered).`); - } - - // Build prompt for triggered agent (memory write-only, no read) - const memoryDir = path.resolve(cwd, '.codemachine', 'memory'); - const adapter = new MemoryAdapter(memoryDir); - const store = new MemoryStore(adapter); - const compositePrompt = triggeredAgentTemplate; - - // Execute triggered agent - let totalTriggeredStdout = ''; - const triggeredResult = await engine.run({ - prompt: compositePrompt, - workingDir: cwd, - model: triggeredModel, - modelReasoningEffort: triggeredReasoning, - onData: (chunk) => { - totalTriggeredStdout += chunk; - - // Write to log file only (UI reads from log file) - if (loggerService && monitoringAgentId !== undefined) { - loggerService.write(monitoringAgentId, chunk); - } - }, - onErrorData: (chunk) => { - // Write stderr to log file only (UI reads from log file) - if (loggerService && monitoringAgentId !== undefined) { - loggerService.write(monitoringAgentId, `[STDERR] ${chunk}`); - } - }, - onTelemetry: (telemetry) => { - ui?.updateAgentTelemetry(triggerAgentId, telemetry); - - // Update telemetry in monitoring (fire and forget - don't block streaming) - if (monitor && monitoringAgentId !== undefined) { - monitor.updateTelemetry(monitoringAgentId, telemetry).catch(err => - console.error(`Failed to update telemetry: ${err}`) - ); - } - }, - abortSignal, - }); - - // NOTE: Telemetry is already updated via onTelemetry callback during streaming execution. - // DO NOT parse from final output - it would match the FIRST telemetry line (early/wrong values) - // instead of the LAST telemetry line (final/correct values), causing incorrect UI display. - - // Store output in memory - const triggeredStdout = triggeredResult.stdout || totalTriggeredStdout; - const triggeredSlice = triggeredStdout.slice(-2000); - await store.append({ - agentId: triggerAgentId, - content: triggeredSlice, - timestamp: new Date().toISOString(), - }); - - // Update UI status on completion - if (ui) { - // Always mark as completed (no failed status) - ui.updateAgentStatus(triggerAgentId, 'completed'); - } - - if (ui) { - ui.logMessage(triggerAgentId, `${triggeredAgentConfig.name ?? triggerAgentId} (triggered) has completed their work.`); - ui.logMessage(triggerAgentId, '═'.repeat(80)); - } - - // Mark agent as completed in monitoring - if (monitor && monitoringAgentId !== undefined) { - await monitor.complete(monitoringAgentId); - // Note: Don't close stream here - workflow may write more messages - // Streams will be closed by cleanup handlers or monitoring service shutdown - } - } catch (triggerError) { - // Mark agent as failed in monitoring - if (monitor && monitoringAgentId !== undefined) { - await monitor.fail(monitoringAgentId, triggerError as Error); - // Note: Don't close stream here - workflow may write more messages - // Streams will be closed by cleanup handlers or monitoring service shutdown - } - - // Don't update status to failed - let it stay as running/retrying - if (ui) { - ui.logMessage( - sourceAgentId, - `Triggered agent '${triggerAgentId}' failed: ${triggerError instanceof Error ? triggerError.message : String(triggerError)}` - ); - } - // Continue with workflow even if triggered agent fails - throw triggerError; - } +export async function executeTriggerAgent( + options: TriggerExecutionOptions, +): Promise { + const { + triggerAgentId, + cwd, + engineType, + logger: _logger, + stderrLogger: _stderrLogger, + sourceAgentId, + ui, + abortSignal, + disableMonitoring, + } = options; + + // Initialize monitoring (unless explicitly disabled) - declare outside try for catch block access + const monitor = !disableMonitoring ? AgentMonitorService.getInstance() : null; + const loggerService = !disableMonitoring + ? AgentLoggerService.getInstance() + : null; + let monitoringAgentId: number | undefined; + + try { + // Load agent config and template from config/main.agents.js + const triggeredAgentConfig = await loadAgentConfig(triggerAgentId, cwd); + const rawTemplate = await loadAgentTemplate(triggerAgentId, cwd); + const triggeredAgentTemplate = await processPromptString(rawTemplate, cwd); + + // Get engine and resolve model/reasoning + const engine = getEngine(engineType); + const engineModule = registry.get(engineType); + const triggeredModel = + (triggeredAgentConfig.model as string | undefined) ?? + engineModule?.metadata.defaultModel; + const triggeredReasoning = + (triggeredAgentConfig.modelReasoningEffort as + | "low" + | "medium" + | "high" + | undefined) ?? engineModule?.metadata.defaultModelReasoningEffort; + + // Find parent agent in monitoring system by sourceAgentId + let parentMonitoringId: number | undefined; + if (monitor) { + const parentAgents = monitor.queryAgents({ name: sourceAgentId }); + if (parentAgents.length > 0) { + // Get the most recent one (highest ID) + parentMonitoringId = parentAgents.sort((a, b) => b.id - a.id)[0].id; + } + + // Register triggered agent with parent relationship and engine/model info + const promptText = `Triggered by ${sourceAgentId}`; + monitoringAgentId = await monitor.register({ + name: triggerAgentId, + prompt: promptText, + parentId: parentMonitoringId, + engineProvider: engineType, + modelName: triggeredModel, + }); + + // Store full prompt for debug mode logging + if (loggerService && monitoringAgentId !== undefined) { + loggerService.storeFullPrompt(monitoringAgentId, promptText); + } + + // Register monitoring ID with UI immediately so it can load logs + if (ui && monitoringAgentId !== undefined) { + ui.registerMonitoringId(triggerAgentId, monitoringAgentId); + } + } + + // Add triggered agent to UI + if (ui) { + const engineName = engineType; // preserve original engine type, even if unknown + ui.addTriggeredAgent(sourceAgentId, { + id: triggerAgentId, + name: triggeredAgentConfig.name ?? triggerAgentId, + engine: engineName, + status: "running", + triggeredBy: sourceAgentId, + startTime: Date.now(), + telemetry: { tokensIn: 0, tokensOut: 0 }, + toolCount: 0, + thinkingCount: 0, + }); + } + + if (ui) { + ui.logMessage( + sourceAgentId, + `Executing triggered agent: ${triggeredAgentConfig.name ?? triggerAgentId}`, + ); + ui.logMessage(triggerAgentId, "═".repeat(80)); + ui.logMessage( + triggerAgentId, + `${triggeredAgentConfig.name ?? triggerAgentId} started to work (triggered).`, + ); + } + + // Build prompt for triggered agent (memory write-only, no read) + const memoryDir = path.resolve(cwd, ".codemachine", "memory"); + const adapter = new MemoryAdapter(memoryDir); + const store = new MemoryStore(adapter); + const compositePrompt = triggeredAgentTemplate; + + // Execute triggered agent + let totalTriggeredStdout = ""; + const triggeredResult = await engine.run({ + prompt: compositePrompt, + workingDir: cwd, + model: triggeredModel, + modelReasoningEffort: triggeredReasoning, + onData: (chunk) => { + totalTriggeredStdout += chunk; + + // Write to log file only (UI reads from log file) + if (loggerService && monitoringAgentId !== undefined) { + loggerService.write(monitoringAgentId, chunk); + } + }, + onErrorData: (chunk) => { + // Write stderr to log file only (UI reads from log file) + if (loggerService && monitoringAgentId !== undefined) { + loggerService.write(monitoringAgentId, `[STDERR] ${chunk}`); + } + }, + onTelemetry: (telemetry) => { + ui?.updateAgentTelemetry(triggerAgentId, telemetry); + + // Update telemetry in monitoring (fire and forget - don't block streaming) + if (monitor && monitoringAgentId !== undefined) { + monitor + .updateTelemetry(monitoringAgentId, telemetry) + .catch((err) => + console.error(`Failed to update telemetry: ${err}`), + ); + } + }, + abortSignal, + }); + + // NOTE: Telemetry is already updated via onTelemetry callback during streaming execution. + // DO NOT parse from final output - it would match the FIRST telemetry line (early/wrong values) + // instead of the LAST telemetry line (final/correct values), causing incorrect UI display. + + // Store output in memory (skip if empty to avoid validation error) + const triggeredStdout = triggeredResult.stdout || totalTriggeredStdout; + const triggeredSlice = triggeredStdout.slice(-2000).trim(); + if (triggeredSlice) { + await store.append({ + agentId: triggerAgentId, + content: triggeredSlice, + timestamp: new Date().toISOString(), + }); + } + + // Update UI status on completion + if (ui) { + // Always mark as completed (no failed status) + ui.updateAgentStatus(triggerAgentId, "completed"); + } + + if (ui) { + ui.logMessage( + triggerAgentId, + `${triggeredAgentConfig.name ?? triggerAgentId} (triggered) has completed their work.`, + ); + ui.logMessage(triggerAgentId, "═".repeat(80)); + } + + // Mark agent as completed in monitoring + if (monitor && monitoringAgentId !== undefined) { + await monitor.complete(monitoringAgentId); + // Note: Don't close stream here - workflow may write more messages + // Streams will be closed by cleanup handlers or monitoring service shutdown + } + } catch (triggerError) { + // Mark agent as failed in monitoring + if (monitor && monitoringAgentId !== undefined) { + await monitor.fail(monitoringAgentId, triggerError as Error); + // Note: Don't close stream here - workflow may write more messages + // Streams will be closed by cleanup handlers or monitoring service shutdown + } + + // Don't update status to failed - let it stay as running/retrying + if (ui) { + ui.logMessage( + sourceAgentId, + `Triggered agent '${triggerAgentId}' failed: ${triggerError instanceof Error ? triggerError.message : String(triggerError)}`, + ); + } + // Continue with workflow even if triggered agent fails + throw triggerError; + } } diff --git a/src/workflows/execution/workflow.ts b/src/workflows/execution/workflow.ts index c0f4b6ff..729b372d 100644 --- a/src/workflows/execution/workflow.ts +++ b/src/workflows/execution/workflow.ts @@ -1,543 +1,700 @@ -import * as path from 'node:path'; -import * as fs from 'node:fs'; +import * as path from "node:path"; +import * as fs from "node:fs"; -import type { RunWorkflowOptions } from '../templates/index.js'; -import { loadTemplateWithPath } from '../templates/index.js'; -import { formatAgentLog } from '../../shared/logging/index.js'; -import { debug, setDebugLogFile } from '../../shared/logging/logger.js'; +import type { RunWorkflowOptions } from "../templates/index.js"; +import { loadTemplateWithPath } from "../templates/index.js"; +import { formatAgentLog } from "../../shared/logging/index.js"; +import { debug, info, setDebugLogFile } from "../../shared/logging/logger.js"; import { - getTemplatePathFromTracking, - getCompletedSteps, - getNotCompletedSteps, - markStepCompleted, - markStepStarted, - removeFromNotCompleted, - getResumeStartIndex, -} from '../../shared/workflows/index.js'; -import { registry } from '../../infra/engines/index.js'; -import { shouldSkipStep, logSkipDebug, type ActiveLoop } from '../behaviors/skip.js'; -import { handleLoopLogic, createActiveLoop } from '../behaviors/loop/controller.js'; -import { handleTriggerLogic } from '../behaviors/trigger/controller.js'; -import { handleCheckpointLogic } from '../behaviors/checkpoint/controller.js'; -import { executeStep } from './step.js'; -import { executeTriggerAgent } from './trigger.js'; -import { shouldExecuteFallback, executeFallbackStep } from './fallback.js'; -import { WorkflowUIManager } from '../../ui/index.js'; -import { MonitoringCleanup } from '../../agents/monitoring/index.js'; + getTemplatePathFromTracking, + getCompletedSteps, + getNotCompletedSteps, + markStepCompleted, + markStepStarted, + removeFromNotCompleted, + getResumeStartIndex, +} from "../../shared/workflows/index.js"; +import { registry } from "../../infra/engines/index.js"; +import { + shouldSkipStep, + logSkipDebug, + type ActiveLoop, +} from "../behaviors/skip.js"; +import { + handleLoopLogic, + createActiveLoop, +} from "../behaviors/loop/controller.js"; +import { handleTriggerLogic } from "../behaviors/trigger/controller.js"; +import { handleCheckpointLogic } from "../behaviors/checkpoint/controller.js"; +import { executeStep } from "./step.js"; +import { executeTriggerAgent } from "./trigger.js"; +import { shouldExecuteFallback, executeFallbackStep } from "./fallback.js"; +import { WorkflowUIManager } from "../../ui/index.js"; +import { MonitoringCleanup } from "../../agents/monitoring/index.js"; +import { + type OpenCodeServer, + startOpenCodeServer, + attachToExternalServer, + checkServerHealth, +} from "../../infra/engines/providers/opencode/execution/server.js"; /** * Cache for engine authentication status with TTL * Prevents repeated auth checks (which can take 10-30 seconds) */ class EngineAuthCache { - private cache: Map = new Map(); - private ttlMs: number = 5 * 60 * 1000; // 5 minutes TTL - - /** - * Check if engine is authenticated (with caching) - */ - async isAuthenticated(engineId: string, checkFn: () => Promise): Promise { - const cached = this.cache.get(engineId); - const now = Date.now(); - - // Return cached value if still valid - if (cached && (now - cached.timestamp) < this.ttlMs) { - return cached.isAuthenticated; - } - - // Cache miss or expired - perform actual check - const result = await checkFn(); - - // Cache the result - this.cache.set(engineId, { - isAuthenticated: result, - timestamp: now - }); - - return result; - } - - /** - * Invalidate cache for specific engine - */ - invalidate(engineId: string): void { - this.cache.delete(engineId); - } - - /** - * Clear entire cache - */ - clear(): void { - this.cache.clear(); - } + private cache: Map = + new Map(); + private ttlMs: number = 5 * 60 * 1000; // 5 minutes TTL + + /** + * Check if engine is authenticated (with caching) + */ + async isAuthenticated( + engineId: string, + checkFn: () => Promise, + ): Promise { + const cached = this.cache.get(engineId); + const now = Date.now(); + + // Return cached value if still valid + if (cached && now - cached.timestamp < this.ttlMs) { + return cached.isAuthenticated; + } + + // Cache miss or expired - perform actual check + const result = await checkFn(); + + // Cache the result + this.cache.set(engineId, { + isAuthenticated: result, + timestamp: now, + }); + + return result; + } + + /** + * Invalidate cache for specific engine + */ + invalidate(engineId: string): void { + this.cache.delete(engineId); + } + + /** + * Clear entire cache + */ + clear(): void { + this.cache.clear(); + } } // Global auth cache instance const authCache = new EngineAuthCache(); -export async function runWorkflow(options: RunWorkflowOptions = {}): Promise { - const cwd = options.cwd ? path.resolve(options.cwd) : process.cwd(); - - // Redirect debug logs to file whenever LOG_LEVEL=debug (or DEBUG env is truthy) so they don't break Ink layout - const rawLogLevel = (process.env.LOG_LEVEL || '').trim().toLowerCase(); - const debugFlag = (process.env.DEBUG || '').trim().toLowerCase(); - const debugEnabled = rawLogLevel === 'debug' || (debugFlag !== '' && debugFlag !== '0' && debugFlag !== 'false'); - const isDebugLogLevel = debugEnabled; - const debugLogPath = isDebugLogLevel ? path.join(cwd, '.codemachine', 'logs', 'workflow-debug.log') : null; - setDebugLogFile(debugLogPath); - - // Set up cleanup handlers for graceful shutdown - MonitoringCleanup.setup(); - - // Load template from .codemachine/template.json or use provided path - const cmRoot = path.join(cwd, '.codemachine'); - const templatePath = options.templatePath || (await getTemplatePathFromTracking(cmRoot)); - - const { template } = await loadTemplateWithPath(cwd, templatePath); - - debug(`Using workflow template: ${template.name}`); - - // Sync agent configurations before running the workflow - const workflowAgents = Array.from( - template.steps - .filter((step) => step.type === 'module') - .reduce((acc, step) => { - const id = step.agentId?.trim(); - if (!id) return acc; - const existing = acc.get(id) ?? { id }; - acc.set(id, { - ...existing, - id, - model: step.model ?? existing.model, - modelReasoningEffort: step.modelReasoningEffort ?? existing.modelReasoningEffort, - }); - return acc; - }, new Map()).values(), - ); - - // Sync agent configurations for engines that need it - if (workflowAgents.length > 0) { - const engines = registry.getAll(); - for (const engine of engines) { - if (engine.syncConfig) { - await engine.syncConfig({ additionalAgents: workflowAgents }); - } - } - } - - // Load completed steps for executeOnce tracking - const completedSteps = await getCompletedSteps(cmRoot); - - // Load not completed steps for fallback tracking - const notCompletedSteps = await getNotCompletedSteps(cmRoot); - - const loopCounters = new Map(); - let activeLoop: ActiveLoop | null = null; - const workflowStartTime = Date.now(); - - // Initialize Workflow UI Manager - const ui = new WorkflowUIManager(template.name); - if (debugLogPath) { - ui.setDebugLogPath(debugLogPath); - } - - // Pre-populate timeline with all workflow steps BEFORE starting UI - // This prevents duplicate renders at startup - // Set initial status based on completion tracking - template.steps.forEach((step, stepIndex) => { - if (step.type === 'module') { - const defaultEngine = registry.getDefault(); - const engineType = step.engine ?? defaultEngine?.metadata.id ?? 'unknown'; - const engineName = engineType; // preserve original engine type, even if unknown - - // Create a unique identifier for each step instance (agentId + stepIndex) - // This allows multiple instances of the same agent to appear separately in the UI - const uniqueAgentId = `${step.agentId}-step-${stepIndex}`; - - // Determine initial status based on completion tracking - let initialStatus: 'pending' | 'completed' = 'pending'; - if (completedSteps.includes(stepIndex)) { - initialStatus = 'completed'; - } - - const agentId = ui.addMainAgent(step.agentName ?? step.agentId, engineName, stepIndex, initialStatus, uniqueAgentId); - - // Update agent with step information - const state = ui.getState(); - const agent = state.agents.find(a => a.id === agentId); - if (agent) { - agent.stepIndex = stepIndex; - agent.totalSteps = template.steps.filter(s => s.type === 'module').length; - } - } else if (step.type === 'ui') { - // Pre-populate UI elements - ui.addUIElement(step.text, stepIndex); - } - }); - - // Start UI after all agents are pre-populated (single clean render) - ui.start(); - - // Get the starting index based on resume configuration - const startIndex = await getResumeStartIndex(cmRoot); - - if (startIndex > 0) { - console.log(`Resuming workflow from step ${startIndex}...`); - } - - // Workflow stop flag for Ctrl+C handling - let workflowShouldStop = false; - let stoppedByCheckpointQuit = false; - const stopListener = () => { - workflowShouldStop = true; - }; - process.on('workflow:stop', stopListener); - - try { - for (let index = startIndex; index < template.steps.length; index += 1) { - // Check if workflow should stop (Ctrl+C pressed) - if (workflowShouldStop) { - console.log(formatAgentLog('workflow', 'Workflow stopped by user.')); - break; - } - - const step = template.steps[index]; - - // UI elements are pre-populated and don't need execution - if (step.type === 'ui') { - continue; - } - - if (step.type !== 'module') { - continue; - } - - // Create unique agent ID for this step instance (matches UI pre-population) - const uniqueAgentId = `${step.agentId}-step-${index}`; - - const skipResult = shouldSkipStep(step, index, completedSteps, activeLoop, ui, uniqueAgentId); - if (skipResult.skip) { - ui.logMessage(uniqueAgentId, skipResult.reason!); - continue; - } - - logSkipDebug(step, activeLoop); - - // Update UI status to running (this clears the output buffer) - ui.updateAgentStatus(uniqueAgentId, 'running'); - - // Log start message AFTER clearing buffer - ui.logMessage(uniqueAgentId, '═'.repeat(80)); - ui.logMessage(uniqueAgentId, `${step.agentName} started to work.`); - - // Reset behavior file to default "continue" before each agent run - const behaviorFile = path.join(cwd, '.codemachine/memory/behavior.json'); - const behaviorDir = path.dirname(behaviorFile); - if (!fs.existsSync(behaviorDir)) { - fs.mkdirSync(behaviorDir, { recursive: true }); - } - fs.writeFileSync(behaviorFile, JSON.stringify({ action: 'continue' }, null, 2)); - - // Mark step as started (adds to notCompletedSteps) - await markStepStarted(cmRoot, index); - - // Determine engine: step override > first authenticated engine - let engineType: string; - if (step.engine) { - engineType = step.engine; - - // If an override is provided but not authenticated, log and fall back - const overrideEngine = registry.get(engineType); - const isOverrideAuthed = overrideEngine - ? await authCache.isAuthenticated(overrideEngine.metadata.id, () => overrideEngine.auth.isAuthenticated()) - : false; - if (!isOverrideAuthed) { - const pretty = overrideEngine?.metadata.name ?? engineType; - ui.logMessage( - uniqueAgentId, - `${pretty} override is not authenticated; falling back to first authenticated engine by order. Run 'codemachine auth login' to use ${pretty}.` - ); - - // Find first authenticated engine by order (with caching) - const engines = registry.getAll(); - let fallbackEngine = null as typeof overrideEngine | null; - for (const eng of engines) { - const isAuth = await authCache.isAuthenticated( - eng.metadata.id, - () => eng.auth.isAuthenticated() - ); - if (isAuth) { - fallbackEngine = eng; - break; - } - } - - // If none authenticated, fall back to registry default (may still require auth) - if (!fallbackEngine) { - fallbackEngine = registry.getDefault() ?? null; - } - - if (fallbackEngine) { - engineType = fallbackEngine.metadata.id; - ui.logMessage( - uniqueAgentId, - `Falling back to ${fallbackEngine.metadata.name} (${engineType})` - ); - } - } - } else { - // Fallback: find first authenticated engine by order (with caching) - const engines = registry.getAll(); - let foundEngine = null; - - for (const engine of engines) { - const isAuth = await authCache.isAuthenticated( - engine.metadata.id, - () => engine.auth.isAuthenticated() - ); - if (isAuth) { - foundEngine = engine; - break; - } - } - - if (!foundEngine) { - // If no authenticated engine, use default (first by order) - foundEngine = registry.getDefault(); - } - - if (!foundEngine) { - throw new Error('No engines registered. Please install at least one engine.'); - } - - engineType = foundEngine.metadata.id; - ui.logMessage(uniqueAgentId, `No engine specified, using ${foundEngine.metadata.name} (${engineType})`); - } - - // Ensure the selected engine is used during execution - // (executeStep falls back to default engine if step.engine is unset) - // Mutate current step to carry the chosen engine forward - step.engine = engineType; - - // Set up skip listener and abort controller for this step (covers fallback + main + triggers) - const abortController = new AbortController(); - let skipRequested = false; // Prevent duplicate skip requests during async abort handling - const skipListener = () => { - if (skipRequested) { - // Ignore duplicate skip events (user pressing Ctrl+S rapidly) - // This prevents multiple "Skip requested" messages during Bun.spawn's async termination - return; - } - skipRequested = true; - ui.logMessage(uniqueAgentId, '⏭️ Skip requested by user...'); - abortController.abort(); - }; - process.once('workflow:skip', skipListener); - - try { - // Check if fallback should be executed before the original step - if (shouldExecuteFallback(step, index, notCompletedSteps)) { - ui.logMessage(uniqueAgentId, `Detected incomplete step. Running fallback agent first.`); - try { - await executeFallbackStep(step, cwd, workflowStartTime, engineType, ui, uniqueAgentId, abortController.signal); - } catch (error) { - // Fallback failed, step remains in notCompletedSteps - ui.logMessage(uniqueAgentId, `Fallback failed. Skipping original step retry.`); - // Don't update status to failed - just let it stay as running or retrying - throw error; - } - } - - const output = await executeStep(step, cwd, { - logger: () => {}, // No-op: UI reads from log files - stderrLogger: () => {}, // No-op: UI reads from log files - ui, - abortSignal: abortController.signal, - uniqueAgentId, - }); - - // Check for trigger behavior first - const triggerResult = await handleTriggerLogic(step, output, cwd, ui); - if (triggerResult?.shouldTrigger && triggerResult.triggerAgentId) { - const triggeredAgentId = triggerResult.triggerAgentId; // Capture for use in callbacks - try { - await executeTriggerAgent({ - triggerAgentId: triggeredAgentId, - cwd, - engineType, - logger: () => {}, // No-op: UI reads from log files - stderrLogger: () => {}, // No-op: UI reads from log files - sourceAgentId: uniqueAgentId, - ui, - abortSignal: abortController.signal, - }); - } catch (triggerError) { - // Check if this was a user-requested skip (abort) - if (triggerError instanceof Error && triggerError.name === 'AbortError') { - ui.updateAgentStatus(triggeredAgentId, 'skipped'); - ui.logMessage(triggeredAgentId, `Triggered agent was skipped by user.`); - } - // Continue with workflow even if triggered agent fails or is skipped - } - } - - // Remove from notCompletedSteps immediately after successful execution - // This must happen BEFORE loop logic to ensure cleanup even when loops trigger - await removeFromNotCompleted(cmRoot, index); - - // Mark step as completed if executeOnce is true - if (step.executeOnce) { - await markStepCompleted(cmRoot, index); - } - - // Update UI status to completed - // This must happen BEFORE loop logic to ensure UI updates even when loops trigger - ui.updateAgentStatus(uniqueAgentId, 'completed'); - - // Log completion messages BEFORE loop check (so they're part of current agent's output) - ui.logMessage(uniqueAgentId, `${step.agentName} has completed their work.`); - ui.logMessage(uniqueAgentId, '\n' + '═'.repeat(80) + '\n'); - - // Check for checkpoint behavior first (to pause workflow for manual review) - const checkpointResult = await handleCheckpointLogic(step, output, cwd, ui); - if (checkpointResult?.shouldStopWorkflow) { - // Wait for user action via events (Continue or Quit) - await new Promise((resolve) => { - const continueHandler = () => { - cleanup(); - resolve(); - }; - const quitHandler = () => { - cleanup(); - workflowShouldStop = true; - stoppedByCheckpointQuit = true; - resolve(); - }; - const cleanup = () => { - process.removeListener('checkpoint:continue', continueHandler); - process.removeListener('checkpoint:quit', quitHandler); - }; - - process.once('checkpoint:continue', continueHandler); - process.once('checkpoint:quit', quitHandler); - }); - - // Clear checkpoint state and resume - ui.clearCheckpointState(); - - if (workflowShouldStop) { - // User chose to quit from checkpoint - set status to stopped - ui.setWorkflowStatus('stopped'); - break; // User chose to quit - } - // Otherwise continue to next step (current step already marked complete via executeOnce) - } - - const loopResult = await handleLoopLogic(step, index, output, loopCounters, cwd, ui); - - if (loopResult.decision?.shouldRepeat) { - // Set active loop with skip list - activeLoop = createActiveLoop(loopResult.decision); - - // Update UI loop state - const loopKey = `${step.module?.id ?? step.agentId}:${index}`; - const iteration = (loopCounters.get(loopKey) || 0) + 1; - ui.setLoopState({ - active: true, - sourceAgent: uniqueAgentId, - backSteps: loopResult.decision.stepsBack, - iteration, - maxIterations: step.module?.behavior?.type === 'loop' ? step.module.behavior.maxIterations ?? Infinity : Infinity, - skipList: loopResult.decision.skipList || [], - reason: loopResult.decision.reason, - }); - - // Reset all agents that will be re-executed in the loop - // Clear their UI data (telemetry, tool counts, subagents) and monitoring registry data - // Save their current state to execution history with cycle number - for (let resetIndex = loopResult.newIndex; resetIndex <= index; resetIndex += 1) { - const resetStep = template.steps[resetIndex]; - if (resetStep && resetStep.type === 'module') { - const resetUniqueAgentId = `${resetStep.agentId}-step-${resetIndex}`; - await ui.resetAgentForLoop(resetUniqueAgentId, iteration); - } - } - - index = loopResult.newIndex; - continue; - } - - // Clear active loop only when a loop step explicitly terminates - const newActiveLoop = createActiveLoop(loopResult.decision); - if (newActiveLoop !== (undefined as unknown as ActiveLoop | null)) { - activeLoop = newActiveLoop; - if (!newActiveLoop) { - ui.setLoopState(null); - ui.clearLoopRound(uniqueAgentId); - } - } - } catch (error) { - // Check if this was a user-requested skip (abort) - if (error instanceof Error && error.name === 'AbortError') { - ui.updateAgentStatus(uniqueAgentId, 'skipped'); - ui.logMessage(uniqueAgentId, `${step.agentName} was skipped by user.`); - ui.logMessage(uniqueAgentId, '\n' + '═'.repeat(80) + '\n'); - // Continue to next step - don't throw - } else { - // Don't update status to failed - let it stay as running/retrying - ui.logMessage( - uniqueAgentId, - `${step.agentName} failed: ${error instanceof Error ? error.message : String(error)}` - ); - throw error; - } - } finally { - // Always clean up the skip listener - process.removeListener('workflow:skip', skipListener); - } - } - - // Check if workflow was stopped by user (Ctrl+C or checkpoint quit) - if (workflowShouldStop) { - if (stoppedByCheckpointQuit) { - // Workflow was stopped by checkpoint quit - status already set to 'stopped' - // UI will stay running showing the stopped status - // Wait indefinitely - user can press Ctrl+C to exit - await new Promise(() => { - // Never resolves - keeps event loop alive until Ctrl+C exits process - }); - } else { - // Workflow was stopped by Ctrl+C - status already updated by MonitoringCleanup handler - // Keep UI alive to show "Press Ctrl+C again to exit" message - // The second Ctrl+C will be handled by MonitoringCleanup's SIGINT handler - // Wait indefinitely - the SIGINT handler will call process.exit() - await new Promise(() => { - // Never resolves - keeps event loop alive until second Ctrl+C exits process - }); - } - } - - // Workflow completed successfully - MonitoringCleanup.clearWorkflowHandlers(); - - // Set status to completed and keep UI alive - ui.setWorkflowStatus('completed'); - // UI will stay running - user presses Ctrl+C to exit with two-stage behavior - // Wait indefinitely - the SIGINT handler will call process.exit() - await new Promise(() => { - // Never resolves - keeps event loop alive until Ctrl+C exits process - }); - } catch (error) { - // On workflow error, set status, stop UI, then exit - ui.setWorkflowStatus('stopped'); - - // Stop UI to restore console before logging error - ui.stop(); - - // Re-throw error to be handled by caller (will now print after UI is stopped) - throw error; - } finally { - // Clean up workflow stop listener - process.removeListener('workflow:stop', stopListener); - } +export async function runWorkflow( + options: RunWorkflowOptions = {}, +): Promise { + const cwd = options.cwd ? path.resolve(options.cwd) : process.cwd(); + + // Redirect debug logs to file whenever LOG_LEVEL=debug (or DEBUG env is truthy) so they don't break Ink layout + const rawLogLevel = (process.env.LOG_LEVEL || "").trim().toLowerCase(); + const debugFlag = (process.env.DEBUG || "").trim().toLowerCase(); + const debugEnabled = + rawLogLevel === "debug" || + (debugFlag !== "" && debugFlag !== "0" && debugFlag !== "false"); + const isDebugLogLevel = debugEnabled; + const debugLogPath = isDebugLogLevel + ? path.join(cwd, ".codemachine", "logs", "workflow-debug.log") + : null; + setDebugLogFile(debugLogPath); + + // Set up cleanup handlers for graceful shutdown + MonitoringCleanup.setup(); + + // Load template from .codemachine/template.json or use provided path + const cmRoot = path.join(cwd, ".codemachine"); + const templatePath = + options.templatePath || (await getTemplatePathFromTracking(cmRoot)); + + const { template } = await loadTemplateWithPath(cwd, templatePath); + + debug(`Using workflow template: ${template.name}`); + + // Check for OpenCode server options (from CLI flags or environment variables) + const useOpencodeServer = + options.opencodeServer || process.env.CODEMACHINE_OPENCODE_SERVER === "1"; + const opencodeAttachUrl = + options.opencodeAttach || process.env.CODEMACHINE_OPENCODE_ATTACH; + + // Detect if any steps use OpenCode engine + const hasOpenCodeSteps = template.steps.some( + (step) => step.type === "module" && step.engine === "opencode", + ); + + // Set up OpenCode server if requested and workflow has OpenCode steps + let opencodeServer: OpenCodeServer | null = null; + if (hasOpenCodeSteps && (useOpencodeServer || opencodeAttachUrl)) { + if (opencodeAttachUrl) { + // Attach to existing external server + info(`Attaching to external OpenCode server: ${opencodeAttachUrl}`); + opencodeServer = attachToExternalServer(opencodeAttachUrl); + + // Verify server is healthy + const healthy = await checkServerHealth(opencodeAttachUrl); + if (!healthy) { + throw new Error( + `Cannot connect to OpenCode server at ${opencodeAttachUrl}. Is it running?`, + ); + } + info(`Connected to OpenCode server at ${opencodeServer.url}`); + } else if (useOpencodeServer) { + // Start consolidated server (experimental) + info( + "[Experimental] Starting consolidated OpenCode server for workflow...", + ); + try { + opencodeServer = await startOpenCodeServer({ workingDir: cwd }); + info(`OpenCode server started at ${opencodeServer.url}`); + } catch (err) { + throw new Error( + `Failed to start OpenCode server: ${err instanceof Error ? err.message : String(err)}`, + ); + } + } + + // Inject attach URL into all OpenCode steps + if (opencodeServer) { + for (const step of template.steps) { + if (step.type === "module" && step.engine === "opencode") { + step.attach = opencodeServer.url; + } + } + } + } + + // Sync agent configurations before running the workflow + const workflowAgents = Array.from( + template.steps + .filter((step) => step.type === "module") + .reduce((acc, step) => { + const id = step.agentId?.trim(); + if (!id) return acc; + const existing = acc.get(id) ?? { id }; + acc.set(id, { + ...existing, + id, + model: step.model ?? existing.model, + modelReasoningEffort: + step.modelReasoningEffort ?? existing.modelReasoningEffort, + }); + return acc; + }, new Map< + string, + { id: string; model?: unknown; modelReasoningEffort?: unknown } + >()) + .values(), + ); + + // Sync agent configurations for engines that need it + if (workflowAgents.length > 0) { + const engines = registry.getAll(); + for (const engine of engines) { + if (engine.syncConfig) { + await engine.syncConfig({ additionalAgents: workflowAgents }); + } + } + } + + // Load completed steps for executeOnce tracking + const completedSteps = await getCompletedSteps(cmRoot); + + // Load not completed steps for fallback tracking + const notCompletedSteps = await getNotCompletedSteps(cmRoot); + + const loopCounters = new Map(); + let activeLoop: ActiveLoop | null = null; + const workflowStartTime = Date.now(); + + // Initialize Workflow UI Manager + const ui = new WorkflowUIManager(template.name); + if (debugLogPath) { + ui.setDebugLogPath(debugLogPath); + } + + // Pre-populate timeline with all workflow steps BEFORE starting UI + // This prevents duplicate renders at startup + // Set initial status based on completion tracking + template.steps.forEach((step, stepIndex) => { + if (step.type === "module") { + const defaultEngine = registry.getDefault(); + const engineType = step.engine ?? defaultEngine?.metadata.id ?? "unknown"; + const engineName = engineType; // preserve original engine type, even if unknown + + // Create a unique identifier for each step instance (agentId + stepIndex) + // This allows multiple instances of the same agent to appear separately in the UI + const uniqueAgentId = `${step.agentId}-step-${stepIndex}`; + + // Determine initial status based on completion tracking + let initialStatus: "pending" | "completed" = "pending"; + if (completedSteps.includes(stepIndex)) { + initialStatus = "completed"; + } + + const agentId = ui.addMainAgent( + step.agentName ?? step.agentId, + engineName, + stepIndex, + initialStatus, + uniqueAgentId, + ); + + // Update agent with step information + const state = ui.getState(); + const agent = state.agents.find((a) => a.id === agentId); + if (agent) { + agent.stepIndex = stepIndex; + agent.totalSteps = template.steps.filter( + (s) => s.type === "module", + ).length; + } + } else if (step.type === "ui") { + // Pre-populate UI elements + ui.addUIElement(step.text, stepIndex); + } + }); + + // Start UI after all agents are pre-populated (single clean render) + ui.start(); + + // Get the starting index based on resume configuration + const startIndex = await getResumeStartIndex(cmRoot); + + if (startIndex > 0) { + console.log(`Resuming workflow from step ${startIndex}...`); + } + + // Workflow stop flag for Ctrl+C handling + let workflowShouldStop = false; + let stoppedByCheckpointQuit = false; + const stopListener = () => { + workflowShouldStop = true; + }; + process.on("workflow:stop", stopListener); + + try { + for (let index = startIndex; index < template.steps.length; index += 1) { + // Check if workflow should stop (Ctrl+C pressed) + if (workflowShouldStop) { + console.log(formatAgentLog("workflow", "Workflow stopped by user.")); + break; + } + + const step = template.steps[index]; + + // UI elements are pre-populated and don't need execution + if (step.type === "ui") { + continue; + } + + if (step.type !== "module") { + continue; + } + + // Create unique agent ID for this step instance (matches UI pre-population) + const uniqueAgentId = `${step.agentId}-step-${index}`; + + const skipResult = shouldSkipStep( + step, + index, + completedSteps, + activeLoop, + ui, + uniqueAgentId, + ); + if (skipResult.skip) { + ui.logMessage(uniqueAgentId, skipResult.reason!); + continue; + } + + logSkipDebug(step, activeLoop); + + // Update UI status to running (this clears the output buffer) + ui.updateAgentStatus(uniqueAgentId, "running"); + + // Log start message AFTER clearing buffer + ui.logMessage(uniqueAgentId, "═".repeat(80)); + ui.logMessage(uniqueAgentId, `${step.agentName} started to work.`); + + // Reset behavior file to default "continue" before each agent run + const behaviorFile = path.join(cwd, ".codemachine/memory/behavior.json"); + const behaviorDir = path.dirname(behaviorFile); + if (!fs.existsSync(behaviorDir)) { + fs.mkdirSync(behaviorDir, { recursive: true }); + } + fs.writeFileSync( + behaviorFile, + JSON.stringify({ action: "continue" }, null, 2), + ); + + // Mark step as started (adds to notCompletedSteps) + await markStepStarted(cmRoot, index); + + // Determine engine: step override > first authenticated engine + let engineType: string; + if (step.engine) { + engineType = step.engine; + + // If an override is provided but not authenticated, log and fall back + const overrideEngine = registry.get(engineType); + const isOverrideAuthed = overrideEngine + ? await authCache.isAuthenticated(overrideEngine.metadata.id, () => + overrideEngine.auth.isAuthenticated(), + ) + : false; + if (!isOverrideAuthed) { + const pretty = overrideEngine?.metadata.name ?? engineType; + ui.logMessage( + uniqueAgentId, + `${pretty} override is not authenticated; falling back to first authenticated engine by order. Run 'codemachine auth login' to use ${pretty}.`, + ); + + // Find first authenticated engine by order (with caching) + const engines = registry.getAll(); + let fallbackEngine = null as typeof overrideEngine | null; + for (const eng of engines) { + const isAuth = await authCache.isAuthenticated( + eng.metadata.id, + () => eng.auth.isAuthenticated(), + ); + if (isAuth) { + fallbackEngine = eng; + break; + } + } + + // If none authenticated, fall back to registry default (may still require auth) + if (!fallbackEngine) { + fallbackEngine = registry.getDefault() ?? null; + } + + if (fallbackEngine) { + engineType = fallbackEngine.metadata.id; + ui.logMessage( + uniqueAgentId, + `Falling back to ${fallbackEngine.metadata.name} (${engineType})`, + ); + } + } + } else { + // Fallback: find first authenticated engine by order (with caching) + const engines = registry.getAll(); + let foundEngine = null; + + for (const engine of engines) { + const isAuth = await authCache.isAuthenticated( + engine.metadata.id, + () => engine.auth.isAuthenticated(), + ); + if (isAuth) { + foundEngine = engine; + break; + } + } + + if (!foundEngine) { + // If no authenticated engine, use default (first by order) + foundEngine = registry.getDefault(); + } + + if (!foundEngine) { + throw new Error( + "No engines registered. Please install at least one engine.", + ); + } + + engineType = foundEngine.metadata.id; + ui.logMessage( + uniqueAgentId, + `No engine specified, using ${foundEngine.metadata.name} (${engineType})`, + ); + } + + // Ensure the selected engine is used during execution + // (executeStep falls back to default engine if step.engine is unset) + // Mutate current step to carry the chosen engine forward + step.engine = engineType; + + // Set up skip listener and abort controller for this step (covers fallback + main + triggers) + const abortController = new AbortController(); + let skipRequested = false; // Prevent duplicate skip requests during async abort handling + const skipListener = () => { + if (skipRequested) { + // Ignore duplicate skip events (user pressing Ctrl+S rapidly) + // This prevents multiple "Skip requested" messages during Bun.spawn's async termination + return; + } + skipRequested = true; + ui.logMessage(uniqueAgentId, "⏭️ Skip requested by user..."); + abortController.abort(); + }; + process.once("workflow:skip", skipListener); + + try { + // Check if fallback should be executed before the original step + if (shouldExecuteFallback(step, index, notCompletedSteps)) { + ui.logMessage( + uniqueAgentId, + `Detected incomplete step. Running fallback agent first.`, + ); + try { + await executeFallbackStep( + step, + cwd, + workflowStartTime, + engineType, + ui, + uniqueAgentId, + abortController.signal, + ); + } catch (error) { + // Fallback failed, step remains in notCompletedSteps + ui.logMessage( + uniqueAgentId, + `Fallback failed. Skipping original step retry.`, + ); + // Don't update status to failed - just let it stay as running or retrying + throw error; + } + } + + const output = await executeStep(step, cwd, { + logger: () => {}, // No-op: UI reads from log files + stderrLogger: () => {}, // No-op: UI reads from log files + ui, + abortSignal: abortController.signal, + uniqueAgentId, + }); + + // Check for trigger behavior first + const triggerResult = await handleTriggerLogic(step, output, cwd, ui); + if (triggerResult?.shouldTrigger && triggerResult.triggerAgentId) { + const triggeredAgentId = triggerResult.triggerAgentId; // Capture for use in callbacks + try { + await executeTriggerAgent({ + triggerAgentId: triggeredAgentId, + cwd, + engineType, + logger: () => {}, // No-op: UI reads from log files + stderrLogger: () => {}, // No-op: UI reads from log files + sourceAgentId: uniqueAgentId, + ui, + abortSignal: abortController.signal, + }); + } catch (triggerError) { + // Check if this was a user-requested skip (abort) + if ( + triggerError instanceof Error && + triggerError.name === "AbortError" + ) { + ui.updateAgentStatus(triggeredAgentId, "skipped"); + ui.logMessage( + triggeredAgentId, + `Triggered agent was skipped by user.`, + ); + } + // Continue with workflow even if triggered agent fails or is skipped + } + } + + // Remove from notCompletedSteps immediately after successful execution + // This must happen BEFORE loop logic to ensure cleanup even when loops trigger + await removeFromNotCompleted(cmRoot, index); + + // Mark step as completed if executeOnce is true + if (step.executeOnce) { + await markStepCompleted(cmRoot, index); + } + + // Update UI status to completed + // This must happen BEFORE loop logic to ensure UI updates even when loops trigger + ui.updateAgentStatus(uniqueAgentId, "completed"); + + // Log completion messages BEFORE loop check (so they're part of current agent's output) + ui.logMessage( + uniqueAgentId, + `${step.agentName} has completed their work.`, + ); + ui.logMessage(uniqueAgentId, "\n" + "═".repeat(80) + "\n"); + + // Check for checkpoint behavior first (to pause workflow for manual review) + const checkpointResult = await handleCheckpointLogic( + step, + output, + cwd, + ui, + ); + if (checkpointResult?.shouldStopWorkflow) { + // Wait for user action via events (Continue or Quit) + await new Promise((resolve) => { + const continueHandler = () => { + cleanup(); + resolve(); + }; + const quitHandler = () => { + cleanup(); + workflowShouldStop = true; + stoppedByCheckpointQuit = true; + resolve(); + }; + const cleanup = () => { + process.removeListener("checkpoint:continue", continueHandler); + process.removeListener("checkpoint:quit", quitHandler); + }; + + process.once("checkpoint:continue", continueHandler); + process.once("checkpoint:quit", quitHandler); + }); + + // Clear checkpoint state and resume + ui.clearCheckpointState(); + + if (workflowShouldStop) { + // User chose to quit from checkpoint - set status to stopped + ui.setWorkflowStatus("stopped"); + break; // User chose to quit + } + // Otherwise continue to next step (current step already marked complete via executeOnce) + } + + const loopResult = await handleLoopLogic( + step, + index, + output, + loopCounters, + cwd, + ui, + ); + + if (loopResult.decision?.shouldRepeat) { + // Set active loop with skip list + activeLoop = createActiveLoop(loopResult.decision); + + // Update UI loop state + const loopKey = `${step.module?.id ?? step.agentId}:${index}`; + const iteration = (loopCounters.get(loopKey) || 0) + 1; + ui.setLoopState({ + active: true, + sourceAgent: uniqueAgentId, + backSteps: loopResult.decision.stepsBack, + iteration, + maxIterations: + step.module?.behavior?.type === "loop" + ? (step.module.behavior.maxIterations ?? Infinity) + : Infinity, + skipList: loopResult.decision.skipList || [], + reason: loopResult.decision.reason, + }); + + // Reset all agents that will be re-executed in the loop + // Clear their UI data (telemetry, tool counts, subagents) and monitoring registry data + // Save their current state to execution history with cycle number + for ( + let resetIndex = loopResult.newIndex; + resetIndex <= index; + resetIndex += 1 + ) { + const resetStep = template.steps[resetIndex]; + if (resetStep && resetStep.type === "module") { + const resetUniqueAgentId = `${resetStep.agentId}-step-${resetIndex}`; + await ui.resetAgentForLoop(resetUniqueAgentId, iteration); + } + } + + index = loopResult.newIndex; + continue; + } + + // Clear active loop only when a loop step explicitly terminates + const newActiveLoop = createActiveLoop(loopResult.decision); + if (newActiveLoop !== (undefined as unknown as ActiveLoop | null)) { + activeLoop = newActiveLoop; + if (!newActiveLoop) { + ui.setLoopState(null); + ui.clearLoopRound(uniqueAgentId); + } + } + } catch (error) { + // Check if this was a user-requested skip (abort) + if (error instanceof Error && error.name === "AbortError") { + ui.updateAgentStatus(uniqueAgentId, "skipped"); + ui.logMessage( + uniqueAgentId, + `${step.agentName} was skipped by user.`, + ); + ui.logMessage(uniqueAgentId, "\n" + "═".repeat(80) + "\n"); + // Continue to next step - don't throw + } else { + // Don't update status to failed - let it stay as running/retrying + ui.logMessage( + uniqueAgentId, + `${step.agentName} failed: ${error instanceof Error ? error.message : String(error)}`, + ); + throw error; + } + } finally { + // Always clean up the skip listener + process.removeListener("workflow:skip", skipListener); + } + } + + // Check if workflow was stopped by user (Ctrl+C or checkpoint quit) + if (workflowShouldStop) { + if (stoppedByCheckpointQuit) { + // Workflow was stopped by checkpoint quit - status already set to 'stopped' + // UI will stay running showing the stopped status + // Wait indefinitely - user can press Ctrl+C to exit + await new Promise(() => { + // Never resolves - keeps event loop alive until Ctrl+C exits process + }); + } else { + // Workflow was stopped by Ctrl+C - status already updated by MonitoringCleanup handler + // Keep UI alive to show "Press Ctrl+C again to exit" message + // The second Ctrl+C will be handled by MonitoringCleanup's SIGINT handler + // Wait indefinitely - the SIGINT handler will call process.exit() + await new Promise(() => { + // Never resolves - keeps event loop alive until second Ctrl+C exits process + }); + } + } + + // Workflow completed successfully + MonitoringCleanup.clearWorkflowHandlers(); + + // Set status to completed and keep UI alive + ui.setWorkflowStatus("completed"); + // UI will stay running - user presses Ctrl+C to exit with two-stage behavior + // Wait indefinitely - the SIGINT handler will call process.exit() + await new Promise(() => { + // Never resolves - keeps event loop alive until Ctrl+C exits process + }); + } catch (error) { + // On workflow error, set status, stop UI, then exit + ui.setWorkflowStatus("stopped"); + + // Stop UI to restore console before logging error + ui.stop(); + + // Re-throw error to be handled by caller (will now print after UI is stopped) + throw error; + } finally { + // Clean up workflow stop listener + process.removeListener("workflow:stop", stopListener); + + // Stop OpenCode server if we started one + if (opencodeServer) { + debug("Stopping OpenCode server..."); + await opencodeServer.stop(); + } + } } diff --git a/src/workflows/templates/loader.ts b/src/workflows/templates/loader.ts index 44ce7d88..b348d972 100644 --- a/src/workflows/templates/loader.ts +++ b/src/workflows/templates/loader.ts @@ -1,88 +1,136 @@ -import * as path from 'node:path'; -import { createRequire } from 'node:module'; -import { pathToFileURL } from 'node:url'; -import type { WorkflowTemplate } from './types.js'; -import { validateWorkflowTemplate } from './validator.js'; -import { ensureTemplateGlobals } from './globals.js'; -import { resolvePackageRoot } from '../../shared/runtime/pkg.js'; +import * as path from "node:path"; +import { createRequire } from "node:module"; +import { pathToFileURL } from "node:url"; +import { existsSync } from "node:fs"; +import type { WorkflowTemplate } from "./types.js"; +import { validateWorkflowTemplate } from "./validator.js"; +import { ensureTemplateGlobals } from "./globals.js"; +import { resolvePackageRoot } from "../../shared/runtime/pkg.js"; // Package root resolution -export const packageRoot = resolvePackageRoot(import.meta.url, 'workflow templates loader'); +export const packageRoot = resolvePackageRoot( + import.meta.url, + "workflow templates loader", +); -export const templatesDir = path.resolve(packageRoot, 'templates', 'workflows'); +export const templatesDir = path.resolve(packageRoot, "templates", "workflows"); // Module loading export async function loadWorkflowModule(modPath: string): Promise { - ensureTemplateGlobals(); - const ext = path.extname(modPath).toLowerCase(); - if (ext === '.cjs' || ext === '.cts') { - const require = createRequire(import.meta.url); - try { - delete require.cache[require.resolve(modPath)]; - } catch { - // Ignore cache deletion errors - } - return require(modPath); - } + ensureTemplateGlobals(); + const ext = path.extname(modPath).toLowerCase(); + if (ext === ".cjs" || ext === ".cts") { + const require = createRequire(import.meta.url); + try { + delete require.cache[require.resolve(modPath)]; + } catch { + // Ignore cache deletion errors + } + return require(modPath); + } - const fileUrl = pathToFileURL(modPath); - const cacheBustingUrl = new URL(fileUrl.href); - cacheBustingUrl.searchParams.set('ts', Date.now().toString()); - const mod = await import(cacheBustingUrl.href); - return mod?.default ?? mod; + const fileUrl = pathToFileURL(modPath); + const cacheBustingUrl = new URL(fileUrl.href); + cacheBustingUrl.searchParams.set("ts", Date.now().toString()); + const mod = await import(cacheBustingUrl.href); + return mod?.default ?? mod; } // Template loading -export async function loadTemplate(cwd: string, templatePath?: string): Promise { - const resolvedTemplateOverride = templatePath - ? path.isAbsolute(templatePath) - ? templatePath - : path.resolve(packageRoot, templatePath) - : undefined; - const codemachineTemplate = path.resolve(templatesDir, 'codemachine.workflow.js'); - const candidates = [resolvedTemplateOverride, codemachineTemplate].filter(Boolean) as string[]; +export async function loadTemplate( + cwd: string, + templatePath?: string, +): Promise { + const resolvedTemplateOverride = templatePath + ? resolveTemplatePath(cwd, templatePath) + : undefined; + const codemachineTemplate = path.resolve( + templatesDir, + "codemachine.workflow.js", + ); + const candidates = [resolvedTemplateOverride, codemachineTemplate].filter( + Boolean, + ) as string[]; - const errors: string[] = []; - for (const modPath of candidates) { - try { - const tpl = (await loadWorkflowModule(modPath)) as unknown; - const result = validateWorkflowTemplate(tpl); - if (result.valid) return tpl as WorkflowTemplate; - const rel = path.relative(cwd, modPath); - errors.push(`${rel}: ${result.errors.join('; ')}`); - } catch (e) { - const rel = path.relative(cwd, modPath); - errors.push(`${rel}: ${e instanceof Error ? e.message : String(e)}`); - } - } - const looked = candidates.map((p) => path.relative(cwd, p)).join(', '); - const details = errors.length ? `\nValidation errors:\n- ${errors.join('\n- ')}` : ''; - throw new Error(`No workflow template found. Looked for: ${looked}${details}`); + const errors: string[] = []; + for (const modPath of candidates) { + try { + const tpl = (await loadWorkflowModule(modPath)) as unknown; + const result = validateWorkflowTemplate(tpl); + if (result.valid) return tpl as WorkflowTemplate; + const rel = path.relative(cwd, modPath); + errors.push(`${rel}: ${result.errors.join("; ")}`); + } catch (e) { + const rel = path.relative(cwd, modPath); + errors.push(`${rel}: ${e instanceof Error ? e.message : String(e)}`); + } + } + const looked = candidates.map((p) => path.relative(cwd, p)).join(", "); + const details = errors.length + ? `\nValidation errors:\n- ${errors.join("\n- ")}` + : ""; + throw new Error( + `No workflow template found. Looked for: ${looked}${details}`, + ); } -export async function loadTemplateWithPath(cwd: string, templatePath?: string): Promise<{ template: WorkflowTemplate; resolvedPath: string }> { - const resolvedTemplateOverride = templatePath - ? path.isAbsolute(templatePath) - ? templatePath - : path.resolve(packageRoot, templatePath) - : undefined; - const codemachineTemplate = path.resolve(templatesDir, 'codemachine.workflow.js'); - const candidates = [resolvedTemplateOverride, codemachineTemplate].filter(Boolean) as string[]; +/** + * Resolves a template path, checking the project's .codemachine directory first. + * For relative paths like "templates/workflows/foo.workflow.js": + * 1. First check if it exists at {cwd}/.codemachine/{templatePath} + * 2. Fall back to CodeMachine's package templates directory + * For absolute paths, use them directly. + */ +function resolveTemplatePath(cwd: string, templatePath: string): string { + // Absolute paths are used directly + if (path.isAbsolute(templatePath)) { + return templatePath; + } - const errors: string[] = []; - for (const modPath of candidates) { - try { - const tpl = (await loadWorkflowModule(modPath)) as unknown; - const result = validateWorkflowTemplate(tpl); - if (result.valid) return { template: tpl as WorkflowTemplate, resolvedPath: modPath }; - const rel = path.relative(cwd, modPath); - errors.push(`${rel}: ${result.errors.join('; ')}`); - } catch (e) { - const rel = path.relative(cwd, modPath); - errors.push(`${rel}: ${e instanceof Error ? e.message : String(e)}`); - } - } - const looked = candidates.map((p) => path.relative(cwd, p)).join(', '); - const details = errors.length ? `\nValidation errors:\n- ${errors.join('\n- ')}` : ''; - throw new Error(`No workflow template found. Looked for: ${looked}${details}`); + // Check project's .codemachine directory first + const projectPath = path.join(cwd, ".codemachine", templatePath); + if (existsSync(projectPath)) { + return projectPath; + } + + // Fall back to CodeMachine's templates directory + return path.resolve(packageRoot, templatePath); +} + +export async function loadTemplateWithPath( + cwd: string, + templatePath?: string, +): Promise<{ template: WorkflowTemplate; resolvedPath: string }> { + const resolvedTemplateOverride = templatePath + ? resolveTemplatePath(cwd, templatePath) + : undefined; + const codemachineTemplate = path.resolve( + templatesDir, + "codemachine.workflow.js", + ); + const candidates = [resolvedTemplateOverride, codemachineTemplate].filter( + Boolean, + ) as string[]; + + const errors: string[] = []; + for (const modPath of candidates) { + try { + const tpl = (await loadWorkflowModule(modPath)) as unknown; + const result = validateWorkflowTemplate(tpl); + if (result.valid) + return { template: tpl as WorkflowTemplate, resolvedPath: modPath }; + const rel = path.relative(cwd, modPath); + errors.push(`${rel}: ${result.errors.join("; ")}`); + } catch (e) { + const rel = path.relative(cwd, modPath); + errors.push(`${rel}: ${e instanceof Error ? e.message : String(e)}`); + } + } + const looked = candidates.map((p) => path.relative(cwd, p)).join(", "); + const details = errors.length + ? `\nValidation errors:\n- ${errors.join("\n- ")}` + : ""; + throw new Error( + `No workflow template found. Looked for: ${looked}${details}`, + ); } diff --git a/src/workflows/templates/types.ts b/src/workflows/templates/types.ts index 74b68925..3066fca2 100644 --- a/src/workflows/templates/types.ts +++ b/src/workflows/templates/types.ts @@ -1,48 +1,67 @@ export type UnknownRecord = Record; export interface LoopModuleBehavior { - type: 'loop'; - action: 'stepBack'; - steps: number; - trigger?: string; // Optional: behavior now controlled via .codemachine/memory/behavior.json - maxIterations?: number; - skip?: string[]; + type: "loop"; + action: "stepBack"; + steps: number; + trigger?: string; // Optional: behavior now controlled via .codemachine/memory/behavior.json + maxIterations?: number; + skip?: string[]; } export interface TriggerModuleBehavior { - type: 'trigger'; - action: 'mainAgentCall'; - triggerAgentId: string; // Agent ID to trigger + type: "trigger"; + action: "mainAgentCall"; + triggerAgentId: string; // Agent ID to trigger } export interface CheckpointModuleBehavior { - type: 'checkpoint'; - action: 'evaluate'; + type: "checkpoint"; + action: "evaluate"; } -export type ModuleBehavior = LoopModuleBehavior | TriggerModuleBehavior | CheckpointModuleBehavior; +export type ModuleBehavior = + | LoopModuleBehavior + | TriggerModuleBehavior + | CheckpointModuleBehavior; export interface ModuleMetadata { - id: string; - behavior?: ModuleBehavior; + id: string; + behavior?: ModuleBehavior; } export interface ModuleStep { - type: 'module'; - agentId: string; - agentName: string; - promptPath: string; - model?: string; - modelReasoningEffort?: 'low' | 'medium' | 'high'; - engine?: string; // Dynamic engine type from registry - module?: ModuleMetadata; - executeOnce?: boolean; - notCompletedFallback?: string; // Agent ID to run if step is in notCompletedSteps + type: "module"; + agentId: string; + agentName: string; + /** Path to prompt file (either promptPath or prompt is required) */ + promptPath?: string; + /** Inline prompt string (either promptPath or prompt is required) */ + prompt?: string; + model?: string; + modelReasoningEffort?: "low" | "medium" | "high"; + engine?: string; // Dynamic engine type from registry + /** + * OpenCode agent name to use (passed as --agent flag) + * This selects which OpenCode agent configuration to use for execution. + * If not specified, uses the engine's default agent. + */ + agent?: string; + /** + * URL of running OpenCode server to attach to (e.g., http://localhost:4096) + * When provided, uses --attach flag to connect to existing server. + * This is typically set automatically by the workflow runner when using + * consolidated server mode (--opencode-server flag). + */ + attach?: string; + module?: ModuleMetadata; + executeOnce?: boolean; + notCompletedFallback?: string; // Agent ID to run if step is in notCompletedSteps } export interface UIStep { - type: 'ui'; - text: string; + type: "ui"; + text: string; } export type WorkflowStep = ModuleStep | UIStep; @@ -51,28 +70,40 @@ export type WorkflowStep = ModuleStep | UIStep; * Type guard to check if a step is a ModuleStep */ export function isModuleStep(step: WorkflowStep): step is ModuleStep { - return step.type === 'module'; + return step.type === "module"; } export interface WorkflowTemplate { - name: string; - steps: WorkflowStep[]; - subAgentIds?: string[]; + name: string; + steps: WorkflowStep[]; + subAgentIds?: string[]; } -export type ModuleName = ModuleStep['agentId']; +export type ModuleName = ModuleStep["agentId"]; export interface RunWorkflowOptions { - cwd?: string; - templatePath?: string; - specificationPath?: string; + cwd?: string; + templatePath?: string; + specificationPath?: string; + /** + * [Experimental] Start a consolidated OpenCode server for all workflow steps. + * Reduces MCP cold boot times by reusing a single server instance. + * The server is started before the first OpenCode step and stopped when the workflow completes. + */ + opencodeServer?: boolean; + /** + * Attach to an existing OpenCode server instead of starting a new one. + * Mutually exclusive with opencodeServer. + * Example: "http://localhost:4096" + */ + opencodeAttach?: string; } export interface TaskManagerOptions { - cwd: string; - tasksPath?: string; - logsPath?: string; - parallel?: boolean; - abortSignal?: AbortSignal; - execute?: (agentId: string, prompt: string) => Promise; + cwd: string; + tasksPath?: string; + logsPath?: string; + parallel?: boolean; + abortSignal?: AbortSignal; + execute?: (agentId: string, prompt: string) => Promise; } diff --git a/src/workflows/templates/validator.ts b/src/workflows/templates/validator.ts index 5bc282dc..62d50bb7 100644 --- a/src/workflows/templates/validator.ts +++ b/src/workflows/templates/validator.ts @@ -1,125 +1,164 @@ -import type { WorkflowTemplate } from './types.js'; +import type { WorkflowTemplate } from "./types.js"; export interface ValidationResult { - valid: boolean; - errors: string[]; + valid: boolean; + errors: string[]; } export function validateWorkflowTemplate(value: unknown): ValidationResult { - const errors: string[] = []; - if (!value || typeof value !== 'object') { - return { valid: false, errors: ['Template is not an object'] }; - } + const errors: string[] = []; + if (!value || typeof value !== "object") { + return { valid: false, errors: ["Template is not an object"] }; + } - const obj = value as { name?: unknown; steps?: unknown }; - if (typeof obj.name !== 'string' || obj.name.trim().length === 0) { - errors.push('Template.name must be a non-empty string'); - } - if (!Array.isArray(obj.steps)) { - errors.push('Template.steps must be an array'); - } else { - obj.steps.forEach((step, index) => { - if (!step || typeof step !== 'object') { - errors.push(`Step[${index}] must be an object`); - return; - } - const candidate = step as { - type?: unknown; - agentId?: unknown; - agentName?: unknown; - promptPath?: unknown; - model?: unknown; - modelReasoningEffort?: unknown; - module?: unknown; - executeOnce?: unknown; - text?: unknown; - }; + const obj = value as { name?: unknown; steps?: unknown }; + if (typeof obj.name !== "string" || obj.name.trim().length === 0) { + errors.push("Template.name must be a non-empty string"); + } + if (!Array.isArray(obj.steps)) { + errors.push("Template.steps must be an array"); + } else { + obj.steps.forEach((step, index) => { + if (!step || typeof step !== "object") { + errors.push(`Step[${index}] must be an object`); + return; + } + const candidate = step as { + type?: unknown; + agentId?: unknown; + agentName?: unknown; + promptPath?: unknown; + prompt?: unknown; + model?: unknown; + modelReasoningEffort?: unknown; + module?: unknown; + executeOnce?: unknown; + text?: unknown; + }; - // Validate step type - if (candidate.type !== 'module' && candidate.type !== 'ui') { - errors.push(`Step[${index}].type must be 'module' or 'ui'`); - } + // Validate step type + if (candidate.type !== "module" && candidate.type !== "ui") { + errors.push(`Step[${index}].type must be 'module' or 'ui'`); + } - // Validate UI step - if (candidate.type === 'ui') { - if (typeof candidate.text !== 'string' || (candidate.text as string).trim().length === 0) { - errors.push(`Step[${index}].text must be a non-empty string`); - } - // UI steps don't need other validation - return; - } + // Validate UI step + if (candidate.type === "ui") { + if ( + typeof candidate.text !== "string" || + (candidate.text as string).trim().length === 0 + ) { + errors.push(`Step[${index}].text must be a non-empty string`); + } + // UI steps don't need other validation + return; + } - // Validate module step - if (candidate.type === 'module') { - if (typeof candidate.agentId !== 'string') { - errors.push(`Step[${index}].agentId must be a string`); - } - if (typeof candidate.agentName !== 'string') { - errors.push(`Step[${index}].agentName must be a string`); - } - if (typeof candidate.promptPath !== 'string') { - errors.push(`Step[${index}].promptPath must be a string`); - } + // Validate module step + if (candidate.type === "module") { + if (typeof candidate.agentId !== "string") { + errors.push(`Step[${index}].agentId must be a string`); + } + if (typeof candidate.agentName !== "string") { + errors.push(`Step[${index}].agentName must be a string`); + } + // Allow either promptPath (file reference) or prompt (inline string) + const hasPromptPath = typeof candidate.promptPath === "string"; + const hasInlinePrompt = + typeof (candidate as { prompt?: unknown }).prompt === "string"; + if (!hasPromptPath && !hasInlinePrompt) { + errors.push( + `Step[${index}] must have either promptPath or prompt as a string`, + ); + } - if (candidate.model !== undefined && typeof candidate.model !== 'string') { - errors.push(`Step[${index}].model must be a string`); - } + if ( + candidate.model !== undefined && + typeof candidate.model !== "string" + ) { + errors.push(`Step[${index}].model must be a string`); + } - if (candidate.modelReasoningEffort !== undefined) { - const mre = candidate.modelReasoningEffort; - if (mre !== 'low' && mre !== 'medium' && mre !== 'high') { - errors.push( - `Step[${index}].modelReasoningEffort must be one of 'low'|'medium'|'high' (got '${String(mre)}')`, - ); - } - } + if (candidate.modelReasoningEffort !== undefined) { + const mre = candidate.modelReasoningEffort; + if (mre !== "low" && mre !== "medium" && mre !== "high") { + errors.push( + `Step[${index}].modelReasoningEffort must be one of 'low'|'medium'|'high' (got '${String(mre)}')`, + ); + } + } - if (candidate.executeOnce !== undefined && typeof candidate.executeOnce !== 'boolean') { - errors.push(`Step[${index}].executeOnce must be a boolean`); - } + if ( + candidate.executeOnce !== undefined && + typeof candidate.executeOnce !== "boolean" + ) { + errors.push(`Step[${index}].executeOnce must be a boolean`); + } - if (candidate.module !== undefined) { - if (!candidate.module || typeof candidate.module !== 'object') { - errors.push(`Step[${index}].module must be an object`); - } else { - const moduleMeta = candidate.module as { id?: unknown; behavior?: unknown }; - if (typeof moduleMeta.id !== 'string') { - errors.push(`Step[${index}].module.id must be a string`); - } - if (moduleMeta.behavior !== undefined) { - if (!moduleMeta.behavior || typeof moduleMeta.behavior !== 'object') { - errors.push(`Step[${index}].module.behavior must be an object`); - } else { - const behavior = moduleMeta.behavior as { - type?: unknown; - action?: unknown; - steps?: unknown; - trigger?: unknown; - maxIterations?: unknown; - }; - if (behavior.type !== 'loop' || behavior.action !== 'stepBack') { - errors.push(`Step[${index}].module.behavior must be { type: 'loop', action: 'stepBack', ... }`); - } - if (typeof behavior.steps !== 'number' || behavior.steps <= 0) { - errors.push(`Step[${index}].module.behavior.steps must be a positive number`); - } - if (behavior.trigger !== undefined && typeof behavior.trigger !== 'string') { - errors.push(`Step[${index}].module.behavior.trigger must be a string if provided`); - } - if (behavior.maxIterations !== undefined && typeof behavior.maxIterations !== 'number') { - errors.push(`Step[${index}].module.behavior.maxIterations must be a number`); - } - } - } - } - } - } - }); - } + if (candidate.module !== undefined) { + if (!candidate.module || typeof candidate.module !== "object") { + errors.push(`Step[${index}].module must be an object`); + } else { + const moduleMeta = candidate.module as { + id?: unknown; + behavior?: unknown; + }; + if (typeof moduleMeta.id !== "string") { + errors.push(`Step[${index}].module.id must be a string`); + } + if (moduleMeta.behavior !== undefined) { + if ( + !moduleMeta.behavior || + typeof moduleMeta.behavior !== "object" + ) { + errors.push(`Step[${index}].module.behavior must be an object`); + } else { + const behavior = moduleMeta.behavior as { + type?: unknown; + action?: unknown; + steps?: unknown; + trigger?: unknown; + maxIterations?: unknown; + }; + if ( + behavior.type !== "loop" || + behavior.action !== "stepBack" + ) { + errors.push( + `Step[${index}].module.behavior must be { type: 'loop', action: 'stepBack', ... }`, + ); + } + if (typeof behavior.steps !== "number" || behavior.steps <= 0) { + errors.push( + `Step[${index}].module.behavior.steps must be a positive number`, + ); + } + if ( + behavior.trigger !== undefined && + typeof behavior.trigger !== "string" + ) { + errors.push( + `Step[${index}].module.behavior.trigger must be a string if provided`, + ); + } + if ( + behavior.maxIterations !== undefined && + typeof behavior.maxIterations !== "number" + ) { + errors.push( + `Step[${index}].module.behavior.maxIterations must be a number`, + ); + } + } + } + } + } + } + }); + } - return { valid: errors.length === 0, errors }; + return { valid: errors.length === 0, errors }; } export function isWorkflowTemplate(value: unknown): value is WorkflowTemplate { - return validateWorkflowTemplate(value).valid; + return validateWorkflowTemplate(value).valid; } diff --git a/src/workflows/utils/resolvers/module.ts b/src/workflows/utils/resolvers/module.ts index ff48ec01..99698ddb 100644 --- a/src/workflows/utils/resolvers/module.ts +++ b/src/workflows/utils/resolvers/module.ts @@ -42,6 +42,7 @@ export function resolveModule(id: string, overrides: ModuleOverrides = {}): Work const promptPath = overrides.promptPath ?? moduleEntry.promptPath; const model = overrides.model ?? moduleEntry.model; const modelReasoningEffort = overrides.modelReasoningEffort ?? moduleEntry.modelReasoningEffort; + const engine = overrides.engine ?? moduleEntry.engine; if (typeof promptPath !== 'string' || !promptPath.trim()) { throw new Error(`Module ${id} is missing a promptPath configuration.`); @@ -56,6 +57,7 @@ export function resolveModule(id: string, overrides: ModuleOverrides = {}): Work promptPath, model, modelReasoningEffort, + engine, module: { id: moduleEntry.id, behavior, diff --git a/tests/integration/opencode/server.test.ts b/tests/integration/opencode/server.test.ts new file mode 100644 index 00000000..08dedb87 --- /dev/null +++ b/tests/integration/opencode/server.test.ts @@ -0,0 +1,189 @@ +/** + * OpenCode Server Integration Tests + * + * Tests the OpenCode server lifecycle management, including: + * - Starting a server on a random port + * - Health checks + * - Listing available agents + * - Attaching to an external server + * - Server cleanup + * + * Note: These tests require OpenCode to be installed (`opencode` CLI available). + * Tests are skipped if OpenCode is not installed. + */ + +import { describe, test, expect } from "bun:test"; +import { execSync } from "node:child_process"; +import { + startOpenCodeServer, + attachToExternalServer, + checkServerHealth, + type OpenCodeServer, +} from "../../../src/infra/engines/providers/opencode/execution/server.js"; +import { + buildOpenCodeRunCommand, + buildOpenCodeServeCommand, +} from "../../../src/infra/engines/providers/opencode/execution/commands.js"; + +// Check if opencode is installed +let opencodeInstalled = false; +try { + execSync("opencode --version", { stdio: "ignore" }); + opencodeInstalled = true; +} catch { + opencodeInstalled = false; +} + +const describeIfOpencode = opencodeInstalled ? describe : describe.skip; + +describeIfOpencode("OpenCode Server", () => { + describe("Command Building", () => { + test("buildOpenCodeRunCommand includes --attach when provided", () => { + const cmd = buildOpenCodeRunCommand({ + model: "anthropic/claude-3-5-sonnet", + agent: "task-implementer", + attach: "http://localhost:4096", + }); + + expect(cmd.command).toBe("opencode"); + expect(cmd.args).toContain("run"); + expect(cmd.args).toContain("--format"); + expect(cmd.args).toContain("json"); + expect(cmd.args).toContain("--attach"); + expect(cmd.args).toContain("http://localhost:4096"); + expect(cmd.args).toContain("--agent"); + expect(cmd.args).toContain("task-implementer"); + expect(cmd.args).toContain("--model"); + expect(cmd.args).toContain("anthropic/claude-3-5-sonnet"); + }); + + test("buildOpenCodeRunCommand omits --attach when not provided", () => { + const cmd = buildOpenCodeRunCommand({ + model: "anthropic/claude-3-5-sonnet", + agent: "build", + }); + + expect(cmd.args).not.toContain("--attach"); + }); + + test("buildOpenCodeServeCommand with port and hostname", () => { + const cmd = buildOpenCodeServeCommand({ + port: 5000, + hostname: "0.0.0.0", + }); + + expect(cmd.command).toBe("opencode"); + expect(cmd.args).toContain("serve"); + expect(cmd.args).toContain("--port"); + expect(cmd.args).toContain("5000"); + expect(cmd.args).toContain("--hostname"); + expect(cmd.args).toContain("0.0.0.0"); + }); + }); + + describe("Server Lifecycle", () => { + let server: OpenCodeServer | null = null; + + test("startOpenCodeServer starts a server on random port", async () => { + server = await startOpenCodeServer({ + startupTimeout: 60000, // 60s timeout for CI + }); + + expect(server).toBeDefined(); + expect(server.url).toMatch(/^http:\/\/127\.0\.0\.1:\d+$/); + expect(server.port).toBeGreaterThan(0); + expect(server.hostname).toBe("127.0.0.1"); + expect(typeof server.stop).toBe("function"); + expect(typeof server.isHealthy).toBe("function"); + expect(typeof server.listAgents).toBe("function"); + }, 90000); // 90s test timeout + + test("server health check returns true for running server", async () => { + if (!server) { + throw new Error("Server not started"); + } + + const healthy = await server.isHealthy(); + expect(healthy).toBe(true); + }); + + test("listAgents returns array of agents", async () => { + if (!server) { + throw new Error("Server not started"); + } + + const agents = await server.listAgents(); + expect(Array.isArray(agents)).toBe(true); + + // OpenCode should have at least the default 'build' agent + // The exact agents depend on the project config + if (agents.length > 0) { + expect(agents[0]).toHaveProperty("id"); + } + }); + + test("server stops cleanly", async () => { + if (!server) { + throw new Error("Server not started"); + } + + await server.stop(); + + // Health check should fail after stop + const healthy = await checkServerHealth(server.url); + expect(healthy).toBe(false); + + server = null; + }); + }); + + describe("External Server", () => { + test("attachToExternalServer creates handle without lifecycle management", () => { + const server = attachToExternalServer("http://localhost:4096"); + + expect(server.url).toBe("http://localhost:4096"); + expect(server.port).toBe(4096); + expect(server.hostname).toBe("localhost"); + expect(typeof server.stop).toBe("function"); + expect(typeof server.isHealthy).toBe("function"); + expect(typeof server.listAgents).toBe("function"); + }); + + test("attachToExternalServer handles URL with trailing slash", () => { + const server = attachToExternalServer("http://localhost:4096/"); + + expect(server.url).toBe("http://localhost:4096"); + }); + + test("attachToExternalServer parses custom port", () => { + const server = attachToExternalServer("http://127.0.0.1:5555"); + + expect(server.port).toBe(5555); + expect(server.hostname).toBe("127.0.0.1"); + }); + }); + + describe("Health Check", () => { + test("checkServerHealth returns false for non-existent server", async () => { + const healthy = await checkServerHealth("http://localhost:59999"); + expect(healthy).toBe(false); + }); + }); +}); + +// Tests that run regardless of OpenCode installation +describe("OpenCode Commands (no runtime)", () => { + test("buildOpenCodeRunCommand defaults to build agent", () => { + const cmd = buildOpenCodeRunCommand({}); + + expect(cmd.args).toContain("--agent"); + expect(cmd.args).toContain("build"); + }); + + test("buildOpenCodeRunCommand includes --format json", () => { + const cmd = buildOpenCodeRunCommand({}); + + expect(cmd.args).toContain("--format"); + expect(cmd.args).toContain("json"); + }); +});