From 4d1f6c2c577485848f848e7a9185b8d61d21a25c Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Sun, 12 Apr 2026 19:34:22 -0700 Subject: [PATCH] fix: add projectId to remaining unscoped API functions - GitHub import functions: apiImportGitHubIssue, apiBatchImportGitHubIssues, apiImportGitHubPull - Task file operations: fetchFileList, fetchFileContent, saveFileContent - AI title summarization: summarizeTitle - Terminal command execution: execTerminalCommand - Wire projectId through GitHubImportModal, AppModals, useFileBrowser, useFileEditor hooks Co-authored-by: factory-droid[bot] <138933559+factory-droid[bot]@users.noreply.github.com> --- packages/dashboard/app/api.ts | 38 +- .../dashboard/app/components/AppModals.tsx | 1 + .../app/components/GitHubImportModal.tsx | 7 +- .../dashboard/app/hooks/useFileBrowser.ts | 8 +- packages/dashboard/app/hooks/useFileEditor.ts | 12 +- packages/engine/src/index.ts | 1 + packages/engine/src/project-engine.ts | 551 ++++++++++++++++++ .../src/runtimes/child-process-worker.ts | 60 +- .../engine/src/runtimes/in-process-runtime.ts | 45 +- 9 files changed, 664 insertions(+), 59 deletions(-) create mode 100644 packages/engine/src/project-engine.ts diff --git a/packages/dashboard/app/api.ts b/packages/dashboard/app/api.ts index 8962fc629..eb46b0aa5 100644 --- a/packages/dashboard/app/api.ts +++ b/packages/dashboard/app/api.ts @@ -577,8 +577,8 @@ export function apiFetchGitHubIssues( } /** Import a specific GitHub issue as a fn task */ -export function apiImportGitHubIssue(owner: string, repo: string, issueNumber: number): Promise { - return api("/github/issues/import", { +export function apiImportGitHubIssue(owner: string, repo: string, issueNumber: number, projectId?: string): Promise { + return api(withProjectId("/github/issues/import", projectId), { method: "POST", body: JSON.stringify({ owner, repo, issueNumber }), }); @@ -599,9 +599,10 @@ export function apiBatchImportGitHubIssues( owner: string, repo: string, issueNumbers: number[], - delayMs?: number + delayMs?: number, + projectId?: string ): Promise<{ results: BatchImportResult[] }> { - return api<{ results: BatchImportResult[] }>("/github/issues/batch-import", { + return api<{ results: BatchImportResult[] }>(withProjectId("/github/issues/batch-import", projectId), { method: "POST", body: JSON.stringify({ owner, repo, issueNumbers, delayMs }), }); @@ -632,8 +633,8 @@ export function apiFetchGitHubPulls( } /** Import a specific GitHub pull request as a fn review task */ -export function apiImportGitHubPull(owner: string, repo: string, prNumber: number): Promise { - return api("/github/pulls/import", { +export function apiImportGitHubPull(owner: string, repo: string, prNumber: number, projectId?: string): Promise { + return api(withProjectId("/github/pulls/import", projectId), { method: "POST", body: JSON.stringify({ owner, repo, prNumber }), }); @@ -814,8 +815,8 @@ export interface TerminalExitEvent { } /** Execute a shell command and get a session ID for streaming output */ -export function execTerminalCommand(command: string): Promise { - return api("/terminal/exec", { +export function execTerminalCommand(command: string, projectId?: string): Promise { + return api(withProjectId("/terminal/exec", projectId), { method: "POST", body: JSON.stringify({ command }), }); @@ -1146,19 +1147,19 @@ export interface SaveFileResponse { } /** List files in task directory */ -export function fetchFileList(taskId: string, path?: string): Promise { +export function fetchFileList(taskId: string, path?: string, projectId?: string): Promise { const query = path ? `?path=${encodeURIComponent(path)}` : ""; - return api(`/tasks/${taskId}/files${query}`); + return api(withProjectId(`/tasks/${taskId}/files${query}`, projectId)); } /** Fetch file content */ -export function fetchFileContent(taskId: string, filePath: string): Promise { - return api(`/tasks/${taskId}/files/${encodeURIComponent(filePath)}`); +export function fetchFileContent(taskId: string, filePath: string, projectId?: string): Promise { + return api(withProjectId(`/tasks/${taskId}/files/${encodeURIComponent(filePath)}`, projectId)); } /** Save file content */ -export function saveFileContent(taskId: string, filePath: string, content: string): Promise { - return api(`/tasks/${taskId}/files/${encodeURIComponent(filePath)}`, { +export function saveFileContent(taskId: string, filePath: string, content: string, projectId?: string): Promise { + return api(withProjectId(`/tasks/${taskId}/files/${encodeURIComponent(filePath)}`, projectId), { method: "POST", body: JSON.stringify({ content }), }); @@ -2734,15 +2735,20 @@ export interface SummarizeTitleResponse { * @param description - The task description to summarize (must be 201-2000 chars) * @param provider - Optional AI model provider (e.g., "anthropic") * @param modelId - Optional AI model ID (e.g., "claude-sonnet-4-5") + * @param projectId - Optional project ID for scoped settings resolution * @returns The generated title (guaranteed ≤60 characters) * @throws Error with descriptive message for 400/429/503 errors */ export async function summarizeTitle( description: string, provider?: string, - modelId?: string + modelId?: string, + projectId?: string ): Promise { - const res = await fetch("/api/ai/summarize-title", { + const url = projectId + ? `/api/ai/summarize-title?projectId=${encodeURIComponent(projectId)}` + : "/api/ai/summarize-title"; + const res = await fetch(url, { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ description, provider, modelId }), diff --git a/packages/dashboard/app/components/AppModals.tsx b/packages/dashboard/app/components/AppModals.tsx index da63fb7c5..c872938e0 100644 --- a/packages/dashboard/app/components/AppModals.tsx +++ b/packages/dashboard/app/components/AppModals.tsx @@ -119,6 +119,7 @@ export function AppModals({ onClose={modalManager.closeGitHubImport} onImport={taskHandlers.handleGitHubImport} tasks={tasks} + projectId={projectId} /> diff --git a/packages/dashboard/app/components/GitHubImportModal.tsx b/packages/dashboard/app/components/GitHubImportModal.tsx index a3ac72b2c..6f3ddb707 100644 --- a/packages/dashboard/app/components/GitHubImportModal.tsx +++ b/packages/dashboard/app/components/GitHubImportModal.tsx @@ -17,6 +17,7 @@ interface GitHubImportModalProps { onClose: () => void; onImport: (task: Task) => void; tasks: Task[]; + projectId?: string; } // Mobile breakpoint in pixels @@ -24,7 +25,7 @@ const MOBILE_BREAKPOINT = 640; type TabType = "issues" | "pulls"; -export function GitHubImportModal({ isOpen, onClose, onImport, tasks }: GitHubImportModalProps) { +export function GitHubImportModal({ isOpen, onClose, onImport, tasks, projectId }: GitHubImportModalProps) { const [owner, setOwner] = useState(""); const [repo, setRepo] = useState(""); const [labels, setLabels] = useState(""); @@ -276,7 +277,7 @@ export function GitHubImportModal({ isOpen, onClose, onImport, tasks }: GitHubIm setError(null); try { - const task = await apiImportGitHubIssue(owner.trim(), repo.trim(), selectedIssueNumber); + const task = await apiImportGitHubIssue(owner.trim(), repo.trim(), selectedIssueNumber, projectId); onImport(task); onClose(); } catch (err: any) { @@ -295,7 +296,7 @@ export function GitHubImportModal({ isOpen, onClose, onImport, tasks }: GitHubIm setError(null); try { - const task = await apiImportGitHubPull(owner.trim(), repo.trim(), selectedPullNumber); + const task = await apiImportGitHubPull(owner.trim(), repo.trim(), selectedPullNumber, projectId); onImport(task); onClose(); } catch (err: any) { diff --git a/packages/dashboard/app/hooks/useFileBrowser.ts b/packages/dashboard/app/hooks/useFileBrowser.ts index 07e87292e..30424c279 100644 --- a/packages/dashboard/app/hooks/useFileBrowser.ts +++ b/packages/dashboard/app/hooks/useFileBrowser.ts @@ -16,9 +16,10 @@ interface UseFileBrowserReturn { * * @param taskId - The task ID to browse * @param enabled - Whether to enable fetching (e.g., when tab is active) + * @param projectId - Optional project ID for scoped store resolution * @returns File browser state and controls */ -export function useFileBrowser(taskId: string, enabled: boolean): UseFileBrowserReturn { +export function useFileBrowser(taskId: string, enabled: boolean, projectId?: string): UseFileBrowserReturn { const [entries, setEntries] = useState([]); const [currentPath, setCurrentPath] = useState("."); const [loading, setLoading] = useState(false); @@ -48,7 +49,8 @@ export function useFileBrowser(taskId: string, enabled: boolean): UseFileBrowser try { const response: FileListResponse = await fetchFileList( taskId, - currentPath === "." ? undefined : currentPath + currentPath === "." ? undefined : currentPath, + projectId ); if (!cancelled) { @@ -71,7 +73,7 @@ export function useFileBrowser(taskId: string, enabled: boolean): UseFileBrowser return () => { cancelled = true; }; - }, [taskId, currentPath, enabled, refreshKey]); + }, [taskId, currentPath, enabled, refreshKey, projectId]); return { entries, diff --git a/packages/dashboard/app/hooks/useFileEditor.ts b/packages/dashboard/app/hooks/useFileEditor.ts index c05edda3e..768435e59 100644 --- a/packages/dashboard/app/hooks/useFileEditor.ts +++ b/packages/dashboard/app/hooks/useFileEditor.ts @@ -20,12 +20,14 @@ interface UseFileEditorReturn { * @param taskId - The task ID * @param filePath - The file path to edit (null if no file selected) * @param enabled - Whether to enable loading (e.g., when editor is visible) + * @param projectId - Optional project ID for scoped store resolution * @returns File editor state and controls */ export function useFileEditor( taskId: string, filePath: string | null, - enabled: boolean + enabled: boolean, + projectId?: string ): UseFileEditorReturn { const [content, setContentState] = useState(""); const [originalContent, setOriginalContent] = useState(""); @@ -56,7 +58,7 @@ export function useFileEditor( setError(null); try { - const response: FileContentResponse = await fetchFileContent(taskId, filePath!); + const response: FileContentResponse = await fetchFileContent(taskId, filePath!, projectId); if (!cancelled) { setContentState(response.content); @@ -82,7 +84,7 @@ export function useFileEditor( return () => { cancelled = true; }; - }, [taskId, filePath, enabled]); + }, [taskId, filePath, enabled, projectId]); const hasChanges = content !== originalContent; @@ -95,7 +97,7 @@ export function useFileEditor( setError(null); try { - const response: SaveFileResponse = await saveFileContent(taskId, filePath, content); + const response: SaveFileResponse = await saveFileContent(taskId, filePath, content, projectId); setOriginalContent(content); setMtime(response.mtime); } catch (err: any) { @@ -104,7 +106,7 @@ export function useFileEditor( } finally { setSaving(false); } - }, [taskId, filePath, content, hasChanges]); + }, [taskId, filePath, content, hasChanges, projectId]); return { content, diff --git a/packages/engine/src/index.ts b/packages/engine/src/index.ts index 521cc8adc..e06acaf75 100644 --- a/packages/engine/src/index.ts +++ b/packages/engine/src/index.ts @@ -35,6 +35,7 @@ export { TokenCapDetector, type TokenCapCheckResult } from "./token-cap-detector export { SelfHealingManager, type SelfHealingOptions } from "./self-healing.js"; export { PluginRunner, type PluginRunnerOptions } from "./plugin-runner.js"; export { ProjectManager } from "./project-manager.js"; +export { ProjectEngine, type ProjectEngineOptions } from "./project-engine.js"; export { NodeHealthMonitor } from "./node-health-monitor.js"; export { PeerExchangeService, type PeerExchangeServiceOptions, type SyncResult } from "./peer-exchange-service.js"; export { RemoteNodeClient } from "./runtimes/remote-node-client.js"; diff --git a/packages/engine/src/project-engine.ts b/packages/engine/src/project-engine.ts new file mode 100644 index 000000000..ec8819602 --- /dev/null +++ b/packages/engine/src/project-engine.ts @@ -0,0 +1,551 @@ +import type { + TaskStore, + Task, + CentralCore, + Settings, + AutomationStore as AutomationStoreType, +} from "@fusion/core"; +import { InProcessRuntime } from "./runtimes/in-process-runtime.js"; +import type { ProjectRuntimeConfig } from "./project-runtime.js"; +import { PrMonitor } from "./pr-monitor.js"; +import { PrCommentHandler } from "./pr-comment-handler.js"; +import { NtfyNotifier } from "./notifier.js"; +import { CronRunner, createAiPromptExecutor } from "./cron-runner.js"; +import { aiMergeTask } from "./merger.js"; +import { PRIORITY_MERGE } from "./concurrency.js"; +import { runtimeLog } from "./logger.js"; + +/** + * Callback for processing pull-request merge strategy. + * Injected from the CLI layer since it depends on GitHubClient. + */ +export type ProcessPullRequestMergeFn = ( + store: TaskStore, + cwd: string, + taskId: string, +) => Promise<"merged" | "waiting" | "skipped">; + +export interface ProjectEngineOptions { + /** Project identifier for notification deep links */ + projectId?: string; + /** Base URL for ntfy.sh notifications */ + ntfyBaseUrl?: string; + /** + * Returns the merge strategy for the current settings. + * If not provided, defaults to "direct". + */ + getMergeStrategy?: (settings: Settings) => "direct" | "pull-request"; + /** + * Processes a pull-request merge flow. Required when merge strategy + * can be "pull-request". Injected from CLI layer. + */ + processPullRequestMerge?: ProcessPullRequestMergeFn; + /** + * Returns the merge blocker reason for a task, or null/undefined if + * the task is eligible for merge. Imported from @fusion/core. + */ + getTaskMergeBlocker?: (task: Task) => string | null | undefined; + /** + * Callback for insight extraction run processing. + * Invoked after CronRunner completes a memory insight extraction schedule. + */ + onInsightRunProcessed?: (schedule: unknown, result: unknown) => void | Promise; +} + +/** + * ProjectEngine composes an InProcessRuntime with the higher-level + * subsystems that were previously wired inline in serve.ts / dashboard.ts: + * + * - **Auto-merge queue** — serialized merge with conflict retry, semaphore gating + * - **PrMonitor + PrCommentHandler** — GitHub PR feedback loop + * - **NtfyNotifier** — push notifications + * - **CronRunner + AutomationStore** — scheduled automations + * - **Settings event listeners** — dynamic reconfiguration + * + * This ensures every InProcessRuntime (single-project CLI or multi-project + * via ProjectManager) gets the full subsystem set, eliminating the class of + * bugs where a subsystem is forgotten in one code path. + */ +export class ProjectEngine { + private runtime: InProcessRuntime; + private prMonitor?: PrMonitor; + private prCommentHandler?: PrCommentHandler; + private notifier?: NtfyNotifier; + private cronRunner?: CronRunner; + private automationStore?: AutomationStoreType; + + // ── Auto-merge state ── + private mergeQueue: string[] = []; + private mergeActive = new Set(); + private mergeRunning = false; + private activeMergeSession: { dispose: () => void } | null = null; + private mergeRetryTimer: ReturnType | null = null; + private shuttingDown = false; + + private static readonly MAX_AUTO_MERGE_RETRIES = 3; + + // Event handler references for cleanup + private settingsHandlers: Array<(...args: any[]) => void> = []; + private taskMovedHandler?: (...args: any[]) => void; + + constructor( + private config: ProjectRuntimeConfig, + centralCore: CentralCore, + private options: ProjectEngineOptions = {}, + ) { + this.runtime = new InProcessRuntime(config, centralCore); + } + + /** + * Start the engine: initialize the runtime and all auxiliary subsystems. + */ + async start(): Promise { + // 1. Start the core runtime (TaskStore, Scheduler, Executor, Triage, etc.) + await this.runtime.start(); + + const store = this.runtime.getTaskStore(); + const cwd = this.config.workingDirectory; + + // 2. Initialize PrMonitor + PrCommentHandler + this.prMonitor = new PrMonitor(); + this.prCommentHandler = new PrCommentHandler(store); + this.prMonitor.onNewComments((taskId, prInfo, comments) => + this.prCommentHandler!.handleNewComments(taskId, prInfo, comments), + ); + + // 3. Initialize NtfyNotifier + this.notifier = new NtfyNotifier(store, { + projectId: this.options.projectId, + ntfyBaseUrl: this.options.ntfyBaseUrl, + }); + await this.notifier.start(); + + // 4. Initialize AutomationStore + CronRunner + try { + const { AutomationStore } = await import("@fusion/core"); + this.automationStore = new AutomationStore(cwd); + await this.automationStore.init(); + + const aiPromptExecutor = await createAiPromptExecutor(cwd); + this.cronRunner = new CronRunner(store, this.automationStore, { + aiPromptExecutor, + onScheduleRunProcessed: this.options.onInsightRunProcessed as any, + }); + + // Sync insight extraction automation on startup + try { + const { syncInsightExtractionAutomation } = await import("@fusion/core"); + if (typeof syncInsightExtractionAutomation === "function") { + const settings = await store.getSettings(); + await syncInsightExtractionAutomation(this.automationStore, settings); + } + } catch { + // syncInsightExtractionAutomation may not be exported yet + } + + this.cronRunner.start(); + runtimeLog.log("CronRunner initialized and started"); + } catch (err) { + // Non-fatal — automations are optional + runtimeLog.warn( + "AutomationStore/CronRunner initialization failed (continuing without automations):", + err instanceof Error ? err.message : err, + ); + } + + // 5. Wire settings event listeners + this.wireSettingsListeners(store); + + // 6. Wire auto-merge on task:moved + this.wireAutoMerge(store, cwd); + + // 7. Auto-merge startup sweep + await this.startupMergeSweep(store); + + // 8. Start periodic merge retry sweep + this.scheduleMergeRetry(store); + + runtimeLog.log(`ProjectEngine started for ${this.config.projectId}`); + } + + /** + * Gracefully stop the engine and all subsystems. + */ + async stop(): Promise { + this.shuttingDown = true; + + // Stop merge retry timer + if (this.mergeRetryTimer) { + clearTimeout(this.mergeRetryTimer); + this.mergeRetryTimer = null; + } + + // Terminate active merge session + if (this.activeMergeSession) { + this.activeMergeSession.dispose(); + this.activeMergeSession = null; + } + + // Remove event listeners + try { + const store = this.runtime.getTaskStore(); + for (const handler of this.settingsHandlers) { + store.off("settings:updated", handler); + } + if (this.taskMovedHandler) { + store.off("task:moved", this.taskMovedHandler); + } + } catch { + // Store may not be initialized if start() failed partway + } + + // Stop auxiliary subsystems + this.notifier?.stop(); + this.cronRunner?.stop(); + + // Stop the core runtime (Triage, Scheduler, Executor, etc.) + await this.runtime.stop(); + + runtimeLog.log(`ProjectEngine stopped for ${this.config.projectId}`); + } + + // ── Public accessors ── + + /** Get the underlying InProcessRuntime. */ + getRuntime(): InProcessRuntime { + return this.runtime; + } + + /** Get the TaskStore. Throws if not started. */ + getTaskStore(): TaskStore { + return this.runtime.getTaskStore(); + } + + /** Get the PrMonitor (if initialized). */ + getPrMonitor(): PrMonitor | undefined { + return this.prMonitor; + } + + /** Get the CronRunner (if initialized). */ + getCronRunner(): CronRunner | undefined { + return this.cronRunner; + } + + // ── Auto-merge subsystem ── + + private canMergeTask(task: Task): boolean { + const blocker = this.options.getTaskMergeBlocker?.(task); + if (blocker) return false; + return (task.mergeRetries ?? 0) < ProjectEngine.MAX_AUTO_MERGE_RETRIES; + } + + private enqueueMerge(taskId: string): void { + if (this.mergeActive.has(taskId)) return; + this.mergeActive.add(taskId); + this.mergeQueue.push(taskId); + void this.drainMergeQueue(); + } + + private async drainMergeQueue(): Promise { + if (this.mergeRunning) return; + this.mergeRunning = true; + + try { + const store = this.runtime.getTaskStore(); + const cwd = this.config.workingDirectory; + + while (this.mergeQueue.length > 0 && !this.shuttingDown) { + const taskId = this.mergeQueue.shift()!; + try { + const task = await store.getTask(taskId); + if (!task || task.column !== "in-review") { + continue; + } + + const settings = await store.getSettings(); + if (settings.globalPause || settings.enginePaused) break; + + const mergeStrategy = this.options.getMergeStrategy?.(settings) ?? "direct"; + + if (mergeStrategy === "pull-request" && this.options.processPullRequestMerge) { + runtimeLog.log(`Processing PR flow for ${taskId}...`); + const result = await this.options.processPullRequestMerge(store, cwd, taskId); + runtimeLog.log(`PR merge result for ${taskId}: ${result}`); + } else { + // Direct merge via AI agent, gated by semaphore + runtimeLog.log(`Merging ${taskId}...`); + const semaphore = (this.runtime as any).globalSemaphore; + const pool = (this.runtime as any).worktreePool; + const agentStore = (this.runtime as any).agentStore; + const usageLimitPauser = (this.runtime as any).usageLimitPauser; + + const rawMerge = () => + aiMergeTask(store, cwd, taskId, { + pool, + usageLimitPauser, + agentStore, + onSession: (session) => { + this.activeMergeSession = session; + }, + }); + + if (semaphore) { + await semaphore.run(rawMerge, PRIORITY_MERGE); + } else { + await rawMerge(); + } + + this.activeMergeSession = null; + runtimeLog.log(`Merged ${taskId}`); + + // Reset retries on success + if (task.mergeRetries && task.mergeRetries > 0) { + await store.updateTask(taskId, { mergeRetries: 0 }); + } + } + } catch (err: any) { + this.activeMergeSession = null; + const errorMsg = err?.message ?? String(err); + runtimeLog.error(`Merge failed for ${taskId}: ${errorMsg}`); + + // Conflict retry with exponential backoff + const isConflictError = + errorMsg.includes("conflict") || errorMsg.includes("Conflict"); + + if (isConflictError) { + try { + const task = await store.getTask(taskId); + const settings = await store.getSettings(); + if ( + task && + settings.autoResolveConflicts !== false && + (task.mergeRetries ?? 0) < ProjectEngine.MAX_AUTO_MERGE_RETRIES + ) { + const retryCount = (task.mergeRetries ?? 0) + 1; + await store.updateTask(taskId, { + mergeRetries: retryCount, + status: null, + }); + + // Exponential backoff: 5s, 10s, 20s + const delayMs = 5000 * Math.pow(2, (task.mergeRetries ?? 0)); + runtimeLog.log( + `Merge conflict retry ${retryCount}/${ProjectEngine.MAX_AUTO_MERGE_RETRIES} for ${taskId} in ${delayMs / 1000}s`, + ); + + setTimeout(() => { + if (!this.shuttingDown) this.enqueueMerge(taskId); + }, delayMs); + } + } catch { + // best-effort retry + } + } + + // Verification failure — move back to in-progress + const isVerificationError = + errorMsg.includes("Verification failed") || + errorMsg.includes("verification failed"); + + if (isVerificationError && !isConflictError) { + try { + const task = await store.getTask(taskId); + if (task?.column === "in-review") { + await store.moveTask(taskId, "in-progress"); + runtimeLog.log(`Verification failure — ${taskId} moved back to in-progress`); + } + } catch { + // best-effort + } + } + } finally { + this.mergeActive.delete(taskId); + } + } + } finally { + this.mergeRunning = false; + } + } + + private wireAutoMerge(store: TaskStore, _cwd: string): void { + this.taskMovedHandler = async ({ task, to }: { task: Task; to: string }) => { + if (to !== "in-review") return; + if (this.options.getTaskMergeBlocker?.(task)) return; + try { + const settings = await store.getSettings(); + if (settings.globalPause || settings.enginePaused) return; + if (!settings.autoMerge) return; + this.enqueueMerge(task.id); + } catch { + // ignore settings read errors + } + }; + store.on("task:moved", this.taskMovedHandler); + } + + private async startupMergeSweep(store: TaskStore): Promise { + try { + const settings = await store.getSettings(); + if (!settings.autoMerge) return; + + const tasks = await store.listTasks({ column: "in-review" }); + const eligible = tasks.filter((t) => this.canMergeTask(t)); + if (eligible.length > 0) { + runtimeLog.log(`Auto-merge startup sweep: enqueueing ${eligible.length} task(s)`); + for (const t of eligible) { + this.enqueueMerge(t.id); + } + } + } catch { + // ignore startup sweep errors + } + } + + private scheduleMergeRetry(store: TaskStore): void { + if (this.shuttingDown) return; + + const schedule = async () => { + if (this.shuttingDown) return; + + try { + const settings = await store.getSettings(); + if (!settings.globalPause && !settings.enginePaused && settings.autoMerge) { + const tasks = await store.listTasks({ column: "in-review" }); + for (const t of tasks) { + if (this.canMergeTask(t)) { + this.enqueueMerge(t.id); + } + } + } + } catch { + // ignore sweep errors + } + + if (!this.shuttingDown) { + const interval = await store + .getSettings() + .then((s) => s.pollIntervalMs ?? 15_000) + .catch(() => 15_000); + this.mergeRetryTimer = setTimeout(() => void schedule(), interval); + } + }; + + // Kick off the first sweep after a delay + this.mergeRetryTimer = setTimeout(() => void schedule(), 15_000); + } + + // ── Settings event listeners ── + + private wireSettingsListeners(store: TaskStore): void { + // 1. Global pause — terminate active merge session + const onGlobalPause = ({ settings, previous }: { settings: Settings; previous: Settings }) => { + if (settings.globalPause && !previous.globalPause) { + if (this.activeMergeSession) { + runtimeLog.log("Global pause — terminating active merge session"); + this.activeMergeSession.dispose(); + this.activeMergeSession = null; + } + } + }; + store.on("settings:updated", onGlobalPause); + this.settingsHandlers.push(onGlobalPause); + + // 2. Global unpause — resume orphaned tasks + sweep in-review + const onGlobalUnpause = async ({ settings: s, previous: prev }: { settings: Settings; previous: Settings }) => { + if (prev.globalPause && !s.globalPause) { + runtimeLog.log("Global unpause — resuming agentic activity"); + + try { + const executor = (this.runtime as any).executor; + executor?.resumeOrphaned?.().catch((err: Error) => + runtimeLog.error("Failed to resume orphaned tasks on unpause:", err), + ); + } catch { /* ignore */ } + + if (s.autoMerge) { + try { + const tasks = await store.listTasks({ column: "in-review" }); + for (const t of tasks) { + if (this.canMergeTask(t)) { + this.enqueueMerge(t.id); + } + } + } catch { /* ignore */ } + } + } + }; + store.on("settings:updated", onGlobalUnpause); + this.settingsHandlers.push(onGlobalUnpause); + + // 3. Engine unpause — same as global unpause + const onEngineUnpause = async ({ settings: s, previous: prev }: { settings: Settings; previous: Settings }) => { + if (prev.enginePaused && !s.enginePaused) { + runtimeLog.log("Engine unpaused — resuming agentic activity"); + + try { + const executor = (this.runtime as any).executor; + executor?.resumeOrphaned?.().catch((err: Error) => + runtimeLog.error("Failed to resume orphaned tasks on engine unpause:", err), + ); + } catch { /* ignore */ } + + if (s.autoMerge) { + try { + const tasks = await store.listTasks({ column: "in-review" }); + for (const t of tasks) { + if (this.canMergeTask(t)) { + this.enqueueMerge(t.id); + } + } + } catch { /* ignore */ } + } + } + }; + store.on("settings:updated", onEngineUnpause); + this.settingsHandlers.push(onEngineUnpause); + + // 4. Stuck task timeout change — trigger immediate check + const onStuckTimeoutChange = async ({ settings: s, previous: prev }: { settings: Settings; previous: Settings }) => { + if (s.taskStuckTimeoutMs !== prev.taskStuckTimeoutMs) { + runtimeLog.log( + `Stuck task timeout changed to ${s.taskStuckTimeoutMs}ms — running immediate check`, + ); + try { + const detector = (this.runtime as any).stuckTaskDetector; + await detector?.checkNow?.(); + } catch { /* ignore */ } + } + }; + store.on("settings:updated", onStuckTimeoutChange); + this.settingsHandlers.push(onStuckTimeoutChange); + + // 5. Insight extraction settings change — sync automation + const onInsightSettingsChange = async ({ settings: s, previous: prev }: { settings: Settings; previous: Settings }) => { + const insightKeys = [ + "insightExtractionEnabled", + "insightExtractionSchedule", + "insightExtractionMinIntervalMs", + ] as const; + + const changed = insightKeys.some( + (key) => (s as any)[key] !== (prev as any)[key], + ); + if (!changed || !this.automationStore) return; + + try { + const { syncInsightExtractionAutomation } = await import("@fusion/core"); + if (typeof syncInsightExtractionAutomation === "function") { + await syncInsightExtractionAutomation(this.automationStore, s); + runtimeLog.log("Insight extraction automation synced with settings"); + } + } catch (err) { + runtimeLog.warn( + "Failed to sync insight extraction automation:", + err instanceof Error ? err.message : err, + ); + } + }; + store.on("settings:updated", onInsightSettingsChange); + this.settingsHandlers.push(onInsightSettingsChange); + } +} diff --git a/packages/engine/src/runtimes/child-process-worker.ts b/packages/engine/src/runtimes/child-process-worker.ts index 2d64f8563..cf66792da 100644 --- a/packages/engine/src/runtimes/child-process-worker.ts +++ b/packages/engine/src/runtimes/child-process-worker.ts @@ -20,12 +20,10 @@ import { GET_METRICS, ERROR_EVENT, type StartRuntimePayload, - type StopRuntimePayload, } from "../ipc/ipc-protocol.js"; -import { InProcessRuntime } from "./in-process-runtime.js"; -import type { ProjectRuntimeConfig } from "../project-runtime.js"; import { runtimeLog } from "../logger.js"; import { CentralCore } from "@fusion/core"; +import { ProjectEngine } from "../project-engine.js"; // Only run if we're in a forked child process if (!process.send) { @@ -38,8 +36,8 @@ runtimeLog.log("Child process worker starting..."); // Create IPC worker const ipcWorker = new IpcWorker(); -// InProcessRuntime instance (created when START_RUNTIME is received) -let runtime: InProcessRuntime | null = null; +// ProjectEngine instance (created when START_RUNTIME is received) +let engine: ProjectEngine | null = null; // Create a minimal CentralCore stub for the child process // The child doesn't need full CentralCore functionality @@ -57,20 +55,24 @@ const createStubCentralCore = (): CentralCore => { // Register command handlers -// START_RUNTIME: Create and start the InProcessRuntime +// START_RUNTIME: Create and start the ProjectEngine (wraps InProcessRuntime + subsystems) ipcWorker.onCommand(START_RUNTIME, async (payload: unknown) => { const { config } = payload as StartRuntimePayload; runtimeLog.log(`Received START_RUNTIME command for project ${config.projectId}`); - if (runtime) { + if (engine) { throw new Error("Runtime already started"); } // Create stub CentralCore (real coordination happens in host) const centralCore = createStubCentralCore(); - // Create InProcessRuntime - runtime = new InProcessRuntime(config, centralCore); + // Create ProjectEngine (includes InProcessRuntime + triage, merge, PR, notifications, cron) + engine = new ProjectEngine(config, centralCore, { + projectId: config.projectId, + }); + + const runtime = engine.getRuntime(); // Forward runtime events to host runtime.on("task:created", (task) => { @@ -96,58 +98,56 @@ ipcWorker.onCommand(START_RUNTIME, async (payload: unknown) => { ipcWorker.sendEvent("HEALTH_CHANGED", data); }); - // Start the runtime - await runtime.start(); + // Start the engine (starts runtime + all subsystems) + await engine.start(); - runtimeLog.log("Runtime started successfully"); + runtimeLog.log("Engine started successfully"); return { status: runtime.getStatus() }; }); -// STOP_RUNTIME: Stop the runtime gracefully -ipcWorker.onCommand(STOP_RUNTIME, async (payload: unknown) => { +// STOP_RUNTIME: Stop the engine gracefully +ipcWorker.onCommand(STOP_RUNTIME, async (_payload: unknown) => { runtimeLog.log("Received STOP_RUNTIME command"); - if (!runtime) { + if (!engine) { throw new Error("Runtime not started"); } - const { timeoutMs } = (payload as StopRuntimePayload) || {}; + await engine.stop(); + engine = null; - await runtime.stop(); - runtime = null; - - runtimeLog.log("Runtime stopped successfully"); + runtimeLog.log("Engine stopped successfully"); return { stopped: true }; }); // GET_STATUS: Return current runtime status ipcWorker.onCommand(GET_STATUS, async () => { - if (!runtime) { + if (!engine) { return { status: "stopped" }; } - return { status: runtime.getStatus() }; + return { status: engine.getRuntime().getStatus() }; }); // GET_METRICS: Return runtime metrics ipcWorker.onCommand(GET_METRICS, async () => { - if (!runtime) { + if (!engine) { return { inFlightTasks: 0, activeAgents: 0, lastActivityAt: new Date().toISOString(), }; } - return runtime.getMetrics(); + return engine.getRuntime().getMetrics(); }); // Handle graceful shutdown process.on("SIGTERM", async () => { runtimeLog.log("Received SIGTERM, initiating graceful shutdown..."); - if (runtime) { + if (engine) { try { - await runtime.stop(); - runtimeLog.log("Runtime stopped gracefully"); + await engine.stop(); + runtimeLog.log("Engine stopped gracefully"); } catch (error) { runtimeLog.error("Error during graceful shutdown:", error); } @@ -159,10 +159,10 @@ process.on("SIGTERM", async () => { process.on("SIGINT", async () => { runtimeLog.log("Received SIGINT, initiating graceful shutdown..."); - if (runtime) { + if (engine) { try { - await runtime.stop(); - runtimeLog.log("Runtime stopped gracefully"); + await engine.stop(); + runtimeLog.log("Engine stopped gracefully"); } catch (error) { runtimeLog.error("Error during graceful shutdown:", error); } diff --git a/packages/engine/src/runtimes/in-process-runtime.ts b/packages/engine/src/runtimes/in-process-runtime.ts index 00e9e8fdf..47ee01da9 100644 --- a/packages/engine/src/runtimes/in-process-runtime.ts +++ b/packages/engine/src/runtimes/in-process-runtime.ts @@ -30,6 +30,7 @@ import { SelfHealingManager } from "../self-healing.js"; import { PluginRunner } from "../plugin-runner.js"; import { MissionAutopilot } from "../mission-autopilot.js"; import { MissionExecutionLoop } from "../mission-execution-loop.js"; +import { TriageProcessor } from "../triage.js"; /** * InProcessRuntime runs a project within the main process. @@ -88,6 +89,7 @@ export class InProcessRuntime private routineRunner?: RoutineRunner; private routineScheduler?: RoutineScheduler; private missionExecutionLoop?: MissionExecutionLoop; + private triageProcessor?: TriageProcessor; /** * @param config - Runtime configuration @@ -218,6 +220,7 @@ export class InProcessRuntime beforeRequeue: (taskId) => this.selfHealingManager?.checkStuckBudget(taskId) ?? Promise.resolve(true), onLoopDetected: (event) => this.executor?.handleLoopDetected(event) ?? Promise.resolve(false), onStuck: (event) => { + this.triageProcessor?.markStuckAborted(event.taskId); this.executor?.markStuckAborted(event.taskId, event.shouldRequeue); runtimeLog.warn( `Task ${event.taskId} stuck (${event.reason}) — ` + @@ -372,6 +375,29 @@ export class InProcessRuntime runtimeLog.warn(`AgentStore initialization failed (continuing without agent monitoring):`, agentErr); } + // 7. Initialize TriageProcessor (task specification) + // Created after AgentStore so per-agent custom instructions are available. + this.triageProcessor = new TriageProcessor( + this.taskStore, + this.config.workingDirectory, + { + semaphore: this.globalSemaphore, + stuckTaskDetector: this.stuckTaskDetector, + agentStore: this.agentStore, + onSpecifyStart: (t) => { + this.recordActivity(); + runtimeLog.log(`Specifying ${t.id}...`); + }, + onSpecifyComplete: (t) => { + this.recordActivity(); + runtimeLog.log(`Specified ${t.id} → todo`); + }, + onSpecifyError: (t, e) => { + runtimeLog.error(`Triage failed for ${t.id}: ${e.message}`); + }, + }, + ); + // Initialize RoutineScheduler (requires RoutineStore from FN-1519) try { const { RoutineStore: RoutineStoreClass } = await import("@fusion/core"); @@ -430,8 +456,9 @@ export class InProcessRuntime // SelfHealingManager so the policy lives in one place. await this.selfHealingManager.runStartupRecovery(); - // 11. Start scheduler + // 11. Start scheduler and triage processor this.scheduler.start(); + this.triageProcessor?.start(); // 12. Start MissionExecutionLoop for validation cycle handling this.missionExecutionLoop = missionExecutionLoop; @@ -524,7 +551,13 @@ export class InProcessRuntime runtimeLog.log("HeartbeatMonitor stopped"); } - // 6. Stop scheduler (prevents new task scheduling) + // 6. Stop triage processor (prevents new specifications) + if (this.triageProcessor) { + this.triageProcessor.stop(); + runtimeLog.log("TriageProcessor stopped"); + } + + // 7. Stop scheduler (prevents new task scheduling) if (this.scheduler) { this.scheduler.stop(); runtimeLog.log("Scheduler stopped"); @@ -668,6 +701,14 @@ export class InProcessRuntime return this.routineScheduler; } + /** + * Get the TriageProcessor instance (if initialized). + * Returns undefined before start() completes. + */ + getTriageProcessor(): TriageProcessor | undefined { + return this.triageProcessor; + } + /** * Execute a heartbeat run for an agent. *