From 64db34da9816cfba5162bf1e991d98cf09358908 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Wed, 15 Jul 2026 11:34:12 -0700 Subject: [PATCH] fix: share engine TaskStore with host extension and harden hang paths Kill the dual-boot FN-7956 class hang for in-process agent tools: - setHostTaskStore/clearHostTaskStores inject the live dashboard/serve/daemon store - Prefer host-injected store over createTaskStoreForBackend; race-safe with external overwrite - fn_skills_install kills npx on abort/timeout so orphan install processes cannot outlive the turn - Raise budgets for task plan, experiment finalize, and mission backfill - Hard-cap import/browse batch size at 50 (GitHub + GitLab) --- .../__tests__/extension-tool-timeout.test.ts | 29 +++- packages/cli/src/commands/daemon.ts | 7 + packages/cli/src/commands/dashboard.ts | 8 + packages/cli/src/commands/serve.ts | 7 + packages/cli/src/extension.ts | 142 +++++++++++++++--- 5 files changed, 167 insertions(+), 26 deletions(-) diff --git a/packages/cli/src/__tests__/extension-tool-timeout.test.ts b/packages/cli/src/__tests__/extension-tool-timeout.test.ts index 4c2a378fa4..87f89c216b 100644 --- a/packages/cli/src/__tests__/extension-tool-timeout.test.ts +++ b/packages/cli/src/__tests__/extension-tool-timeout.test.ts @@ -1,8 +1,11 @@ import { afterEach, describe, expect, it, vi } from "vitest"; import { __clearExtensionStoreBootStateForTesting, + clampImportBrowseLimit, + clearHostTaskStores, raceWithTimeoutAndAbort, resolveExtensionToolTimeoutMs, + setHostTaskStore, wrapExtensionToolExecute, } from "../extension.js"; @@ -16,6 +19,7 @@ Host extension research tools are off; budgets cover remaining long host tools o afterEach(() => { __clearExtensionStoreBootStateForTesting(); + clearHostTaskStores(); vi.restoreAllMocks(); }); @@ -25,14 +29,37 @@ describe("resolveExtensionToolTimeoutMs", () => { expect(resolveExtensionToolTimeoutMs("fn_task_list")).toBe(60_000); }); - it("gives multi-minute budgets to skills install and import/browse tools", () => { + it("gives multi-minute budgets to long host tools", () => { expect(resolveExtensionToolTimeoutMs("fn_skills_install")).toBe(300_000); + expect(resolveExtensionToolTimeoutMs("fn_task_plan")).toBe(300_000); + expect(resolveExtensionToolTimeoutMs("fn_experiment_finalize")).toBe(180_000); + expect(resolveExtensionToolTimeoutMs("fn_mission_backfill_assertions")).toBe(180_000); expect(resolveExtensionToolTimeoutMs("fn_task_import_github")).toBe(180_000); expect(resolveExtensionToolTimeoutMs("fn_task_browse_github_issues")).toBe(180_000); expect(resolveExtensionToolTimeoutMs("fn_web_fetch")).toBe(90_000); }); }); +describe("clampImportBrowseLimit", () => { + it("defaults to 30 and hard-caps at 50", () => { + expect(clampImportBrowseLimit(undefined)).toBe(30); + expect(clampImportBrowseLimit(100)).toBe(50); + expect(clampImportBrowseLimit(0)).toBe(1); + expect(clampImportBrowseLimit(12)).toBe(12); + }); +}); + +describe("setHostTaskStore", () => { + it("is available as a production injection seam for dashboard/serve/daemon", () => { + /* + FNXC:MergeQueue 2026-07-15-11:40: + Host injection must be a real export so dashboard/serve can share the engine TaskStore without dual-boot. + */ + expect(typeof setHostTaskStore).toBe("function"); + expect(typeof clearHostTaskStores).toBe("function"); + }); +}); + describe("raceWithTimeoutAndAbort", () => { it("resolves when the promise wins", async () => { await expect( diff --git a/packages/cli/src/commands/daemon.ts b/packages/cli/src/commands/daemon.ts index 3bb500c5b6..3c4abdbb74 100644 --- a/packages/cli/src/commands/daemon.ts +++ b/packages/cli/src/commands/daemon.ts @@ -37,6 +37,7 @@ import { setHostExtensionPaths, createFusionAuthStorage, } from "@fusion/engine"; +import { setHostTaskStore, clearHostTaskStores } from "../extension.js"; import { DefaultPackageManager, ModelRegistry, @@ -462,6 +463,11 @@ export async function runDaemon(opts: DaemonOptions = {}) { ); const store = primaryEngine.getTaskStore(); + /* + FNXC:MergeQueue 2026-07-15-11:40: + Share the daemon primary TaskStore with the host pi extension so agent fn_* tools reuse the engine pool (no dual-boot). + */ + setHostTaskStore(primaryCwd, store); const getGlobalSettingsStore = () => { const candidate = store as { getGlobalSettingsStore?: () => { getSettings: () => Promise } }; @@ -1002,6 +1008,7 @@ export async function runDaemon(opts: DaemonOptions = {}) { FNXC:PostgresResourceLifecycle 2026-07-14-22:07: Preserve the command-level TaskStore close barrier before CentralCore releases its retained backend. Runtime shutdown normally closes this store first; the idempotent explicit close also covers partial-start and test-owned runtimes. */ + clearHostTaskStores(); await store.close(); // Stop peer exchange service diff --git a/packages/cli/src/commands/dashboard.ts b/packages/cli/src/commands/dashboard.ts index 5f7c062b41..63b6597080 100644 --- a/packages/cli/src/commands/dashboard.ts +++ b/packages/cli/src/commands/dashboard.ts @@ -65,6 +65,7 @@ import { setHostExtensionPaths, createFusionAuthStorage, } from "@fusion/engine"; +import { setHostTaskStore, clearHostTaskStores } from "../extension.js"; import { DefaultPackageManager, ModelRegistry, SettingsManager, discoverAndLoadExtensions, createExtensionRuntime } from "@earendil-works/pi-coding-agent"; import { getMergeStrategy, @@ -889,6 +890,11 @@ export async function runDashboard(port: number, opts: { paused?: boolean; dev?: // PostgreSQL-only; factory failure is surfaced instead of creating a dead store. store = dashboardBackendBoot.taskStore; const dashboardBackendShutdown = dashboardBackendBoot.shutdown; + /* + FNXC:MergeQueue 2026-07-15-11:40: + Share the dashboard TaskStore with the host pi extension so in-process agent fn_* tools never dual-boot a second createTaskStoreForBackend (FN-7956 hang class). + */ + setHostTaskStore(cwd, store); const dashboardLayer = store.getAsyncLayer(); if (!dashboardLayer) throw new Error("Dashboard runtime requires the project PostgreSQL AsyncDataLayer"); // FNXC:PhysicalDeleteSqliteClass 2026-06-26-14:05: @@ -1000,6 +1006,7 @@ export async function runDashboard(port: number, opts: { paused?: boolean; dev?: const boot = await createTaskStoreForBackend({ rootDir: projectPath }); projectStore = boot.taskStore; projectStoreShutdowns.set(projectPath, boot.shutdown); + setHostTaskStore(projectPath, projectStore); } projectStores.set(projectPath, projectStore); return projectStore; @@ -1940,6 +1947,7 @@ export async function runDashboard(port: number, opts: { paused?: boolean; dev?: Dispose secondary stores first, explicitly close the cwd TaskStore so its watcher and timers stop, then invoke the startup factory shutdown that releases the remaining backend resources. The exported dispose path must await every stage. */ disposeCallbacks.push(async () => { + clearHostTaskStores(); await closeProjectStores(); await store?.close(); if (dashboardBackendShutdown) { diff --git a/packages/cli/src/commands/serve.ts b/packages/cli/src/commands/serve.ts index 34725834ea..6c0c0ec63e 100644 --- a/packages/cli/src/commands/serve.ts +++ b/packages/cli/src/commands/serve.ts @@ -37,6 +37,7 @@ import { setHostExtensionPaths, createFusionAuthStorage, } from "@fusion/engine"; +import { setHostTaskStore, clearHostTaskStores } from "../extension.js"; import { DefaultPackageManager, ModelRegistry, @@ -293,6 +294,11 @@ export async function runServe( * Serve must share one successfully booted PostgreSQL layer between CentralCore and the cwd engine. A backend boot error is fatal; constructing a layerless CentralCore would make project discovery appear empty and split control-plane state. */ const centralBootResult = await createTaskStoreForBackend({ rootDir: cwd }); + /* + FNXC:MergeQueue 2026-07-15-11:40: + Share the serve TaskStore with the host pi extension so agent fn_* tools reuse the engine pool (no dual-boot). + */ + setHostTaskStore(cwd, centralBootResult.taskStore); let centralBackendShutdownPromise: Promise | undefined; const shutdownCentralBackendOnce = (): Promise => { centralBackendShutdownPromise ??= centralBootResult.shutdown(); @@ -1185,6 +1191,7 @@ export async function runServe( * process because its runtime receives that store externally. Release the * complete boot result after every engine and CentralCore user has stopped. */ + clearHostTaskStores(); await shutdownCentralBackendOnce().catch((error) => { console.warn(`[serve] PostgreSQL shutdown failed: ${error instanceof Error ? error.message : String(error)}`); }); diff --git a/packages/cli/src/extension.ts b/packages/cli/src/extension.ts index 6847e7bd0c..5619d1f312 100644 --- a/packages/cli/src/extension.ts +++ b/packages/cli/src/extension.ts @@ -257,6 +257,14 @@ const EXTENSION_TOOL_TIMEOUT_MS = 60_000; const SKILLS_INSTALL_TIMEOUT_MS = 300_000; const IMPORT_BROWSE_TIMEOUT_MS = 180_000; const WEB_FETCH_TIMEOUT_MS = 90_000; +const TASK_PLAN_TIMEOUT_MS = 300_000; +const EXPERIMENT_FINALIZE_TIMEOUT_MS = 180_000; +const MISSION_BACKFILL_TIMEOUT_MS = 180_000; +/* +FNXC:MergeQueue 2026-07-15-11:40: +Hard ceiling for import/browse batch size so agents cannot request unbounded GitHub/GitLab fan-out under the host extension. +*/ +export const MAX_IMPORT_BROWSE_ITEMS = 50; function isAbortError(error: unknown): boolean { return ( @@ -276,11 +284,20 @@ function isAbortError(error: unknown): boolean { export function resolveExtensionToolTimeoutMs(toolName: string, _params?: unknown): number { const name = toolName.trim(); if (name === "fn_skills_install") return SKILLS_INSTALL_TIMEOUT_MS; + if (name === "fn_task_plan") return TASK_PLAN_TIMEOUT_MS; + if (name === "fn_experiment_finalize") return EXPERIMENT_FINALIZE_TIMEOUT_MS; + if (name === "fn_mission_backfill_assertions") return MISSION_BACKFILL_TIMEOUT_MS; if (name.startsWith("fn_task_import_") || name.startsWith("fn_task_browse_")) return IMPORT_BROWSE_TIMEOUT_MS; if (name === "fn_web_fetch") return WEB_FETCH_TIMEOUT_MS; return EXTENSION_TOOL_TIMEOUT_MS; } +/** Clamp import/browse limits so host tools cannot request unbounded remote fan-out. */ +export function clampImportBrowseLimit(limit: number | undefined, defaultLimit = 30): number { + const raw = typeof limit === "number" && Number.isFinite(limit) ? Math.floor(limit) : defaultLimit; + return Math.min(MAX_IMPORT_BROWSE_ITEMS, Math.max(1, raw)); +} + /** * Race a promise against a wall-clock timeout and optional AbortSignal. * Does not cancel the underlying work (Node has no structured cancel for store boot), @@ -391,6 +408,10 @@ export function wrapExtensionToolExecute( }; } +/* +FNXC:MergeQueue 2026-07-15-11:40: +When dashboard/serve/daemon injects the live engine TaskStore via setHostTaskStore, getStore must never call createTaskStoreForBackend for that project root — dual-boot was the FN-7956 hang class (second pool + schema advisory lock). CLI one-shot sessions without a host store still boot a short-lived cache entry. +*/ async function getStore(cwd: string, signal?: AbortSignal): Promise { const projectRoot = resolveProjectRoot(cwd); const existing = storeCache.get(projectRoot); @@ -412,7 +433,7 @@ async function getStore(cwd: string, signal?: AbortSignal): Promise { PostgreSQL factory result; the removed SQLite opt-out is an explicit error. FNXC:MergeQueue 2026-07-15-11:08: - First extension tool call in a dashboard/engine process boots a second TaskStore. + First extension tool call without a host-injected store boots a TaskStore (CLI path). Bound that boot and coalesce concurrent callers so a wedged boot cannot park every fn_* tool forever. FNXC:MergeQueue 2026-07-15-11:20: @@ -422,6 +443,12 @@ async function getStore(cwd: string, signal?: AbortSignal): Promise { try { const boot = await createTaskStoreForBackend({ rootDir: projectRoot }); storeBootFailureCooldown.delete(projectRoot); + // Do not overwrite a host-injected external store that landed while we were booting. + const raced = storeCache.get(projectRoot); + if (raced?.external) { + await boot.shutdown().catch(() => undefined); + return raced.store; + } storeCache.set(projectRoot, { store: boot.taskStore, shutdown: boot.shutdown }); return boot.taskStore; } catch (error) { @@ -463,6 +490,38 @@ async function getStore(cwd: string, signal?: AbortSignal): Promise { } } +/** + * FNXC:MergeQueue 2026-07-15-11:40: + * Publish the live engine/dashboard TaskStore into the host extension cache so in-process agent tools share one pool and never dual-boot embedded PostgreSQL. + * The entry is external: closeCachedStores / clearHostTaskStores will not shut it down — the host owns lifecycle. + */ +export function setHostTaskStore(projectRoot: string, store: TaskStore): void { + const canonical = resolveProjectRoot(projectRoot); + storeCache.set(canonical, { store, external: true }); + storeBootInflight.delete(canonical); + storeBootFailureCooldown.delete(canonical); +} + +/** + * Remove host-injected store entries without closing them (host owns lifecycle). + * Pass projectRoot to clear one project; omit to clear all external entries. + */ +export function clearHostTaskStores(projectRoot?: string): void { + if (projectRoot) { + const canonical = resolveProjectRoot(projectRoot); + const entry = storeCache.get(canonical); + if (entry?.external) storeCache.delete(canonical); + storeBootInflight.delete(canonical); + storeBootFailureCooldown.delete(canonical); + return; + } + for (const [key, entry] of [...storeCache.entries()]) { + if (entry.external) storeCache.delete(key); + } + storeBootInflight.clear(); + storeBootFailureCooldown.clear(); +} + /** * @internal Test-only: clear inflight boot map and failure cooldown without closing cached stores. */ @@ -2053,7 +2112,7 @@ export default function kbExtension(pi: ExtensionAPI) { async execute(_toolCallId, params, signal, _onUpdate, ctx) { const [owner, repo] = params.ownerRepo.split("/"); - const limit = params.limit ?? 30; + const limit = clampImportBrowseLimit(params.limit, 30); const labels = params.labels; const issues = await fetchGitHubIssuesViaGh(owner, repo, { limit, labels, signal }); @@ -2244,7 +2303,8 @@ export default function kbExtension(pi: ExtensionAPI) { }), async execute(_toolCallId, params, signal, _onUpdate, ctx) { - const { owner, repo, limit = 30, labels } = params; + const { owner, repo, labels } = params; + const limit = clampImportBrowseLimit(params.limit, 30); const issues = await fetchGitHubIssuesViaGh(owner, repo, { limit, labels, signal }); if (issues.length === 0) { @@ -2329,7 +2389,7 @@ export default function kbExtension(pi: ExtensionAPI) { parameters: Type.Object({ project: Type.String({ description: "GitLab project path or numeric ID" }), limit: Type.Optional(Type.Number({ minimum: 1, maximum: 100 })), labels: Type.Optional(Type.Array(Type.String())) }), async execute(_id, params, _signal, _onUpdate, ctx) { const { client } = await createGitLabClient(ctx); - const issues = await client.listProjectIssues(params.project, { limit: params.limit, labels: params.labels }); + const issues = await client.listProjectIssues(params.project, { limit: clampImportBrowseLimit(params.limit, 30), labels: params.labels }); return { content: [{ type: "text", text: issues.map((issue) => `#${issue.iid}: ${issue.title}\n${issue.webUrl}`).join("\n") || `No GitLab project issues found in ${params.project}.` }], details: { count: issues.length, issues } }; }, }); @@ -2342,7 +2402,7 @@ export default function kbExtension(pi: ExtensionAPI) { parameters: Type.Object({ project: Type.String({ description: "GitLab project path or numeric ID" }), limit: Type.Optional(Type.Number({ minimum: 1, maximum: 100 })), labels: Type.Optional(Type.Array(Type.String())) }), async execute(_id, params, _signal, _onUpdate, ctx) { const { client } = await createGitLabClient(ctx); - const issues = await client.listProjectIssues(params.project, { limit: params.limit, labels: params.labels }); + const issues = await client.listProjectIssues(params.project, { limit: clampImportBrowseLimit(params.limit, 30), labels: params.labels }); const createdTasks = await importGitLabItems(ctx, "project_issue", params.project, issues); return { content: [{ type: "text", text: `Imported ${createdTasks.length} GitLab project issue tasks.` }], details: { createdTasks } }; }, @@ -2356,7 +2416,7 @@ export default function kbExtension(pi: ExtensionAPI) { parameters: Type.Object({ group: Type.String({ description: "GitLab group path or numeric ID" }), limit: Type.Optional(Type.Number({ minimum: 1, maximum: 100 })), labels: Type.Optional(Type.Array(Type.String())) }), async execute(_id, params, _signal, _onUpdate, ctx) { const { client } = await createGitLabClient(ctx); - const issues = await client.listGroupIssues(params.group, { limit: params.limit, labels: params.labels }); + const issues = await client.listGroupIssues(params.group, { limit: clampImportBrowseLimit(params.limit, 30), labels: params.labels }); return { content: [{ type: "text", text: issues.map((issue) => `#${issue.iid}: ${issue.projectPath ?? issue.projectId} — ${issue.title}\n${issue.webUrl}`).join("\n") || `No GitLab group issues found in ${params.group}.` }], details: { count: issues.length, issues } }; }, }); @@ -2369,7 +2429,7 @@ export default function kbExtension(pi: ExtensionAPI) { parameters: Type.Object({ group: Type.String({ description: "GitLab group path or numeric ID" }), limit: Type.Optional(Type.Number({ minimum: 1, maximum: 100 })), labels: Type.Optional(Type.Array(Type.String())) }), async execute(_id, params, _signal, _onUpdate, ctx) { const { client } = await createGitLabClient(ctx); - const issues = await client.listGroupIssues(params.group, { limit: params.limit, labels: params.labels }); + const issues = await client.listGroupIssues(params.group, { limit: clampImportBrowseLimit(params.limit, 30), labels: params.labels }); const createdTasks = await importGitLabItems(ctx, "group_issue", params.group, issues); return { content: [{ type: "text", text: `Imported ${createdTasks.length} GitLab group issue tasks.` }], details: { createdTasks } }; }, @@ -2383,7 +2443,7 @@ export default function kbExtension(pi: ExtensionAPI) { parameters: Type.Object({ project: Type.String({ description: "GitLab project path or numeric ID" }), limit: Type.Optional(Type.Number({ minimum: 1, maximum: 100 })), labels: Type.Optional(Type.Array(Type.String())) }), async execute(_id, params, _signal, _onUpdate, ctx) { const { client } = await createGitLabClient(ctx); - const mergeRequests = await client.listMergeRequests(params.project, { limit: params.limit, labels: params.labels }); + const mergeRequests = await client.listMergeRequests(params.project, { limit: clampImportBrowseLimit(params.limit, 30), labels: params.labels }); return { content: [{ type: "text", text: mergeRequests.map((mr) => `!${mr.iid}: ${mr.title}\n${mr.webUrl}`).join("\n") || `No GitLab merge requests found in ${params.project}.` }], details: { count: mergeRequests.length, mergeRequests } }; }, }); @@ -2396,7 +2456,7 @@ export default function kbExtension(pi: ExtensionAPI) { parameters: Type.Object({ project: Type.String({ description: "GitLab project path or numeric ID" }), limit: Type.Optional(Type.Number({ minimum: 1, maximum: 100 })), labels: Type.Optional(Type.Array(Type.String())) }), async execute(_id, params, _signal, _onUpdate, ctx) { const { client } = await createGitLabClient(ctx); - const mergeRequests = await client.listMergeRequests(params.project, { limit: params.limit, labels: params.labels }); + const mergeRequests = await client.listMergeRequests(params.project, { limit: clampImportBrowseLimit(params.limit, 30), labels: params.labels }); const createdTasks = await importGitLabItems(ctx, "merge_request", params.project, mergeRequests); return { content: [{ type: "text", text: `Imported ${createdTasks.length} GitLab merge request tasks.` }], details: { createdTasks } }; }, @@ -5296,7 +5356,11 @@ export default function kbExtension(pi: ExtensionAPI) { ), }), - async execute(_toolCallId, params, _signal, _onUpdate, ctx) { + /* + FNXC:MergeQueue 2026-07-15-11:40: + Kill npx on abort/timeout so outer tool budgets cannot leave orphan install processes after the agent turn fails closed. + */ + async execute(_toolCallId, params, signal, _onUpdate, ctx) { // Validate source format if (!/^[^/]+\/[^/]+$/.test(params.source)) { return { @@ -5321,7 +5385,14 @@ export default function kbExtension(pi: ExtensionAPI) { // Non-interactive mode (-y) targeting pi agent (-a pi) npxArgs.push("-y", "-a", "pi"); - // Execute via spawn + if (signal?.aborted) { + return { + content: [{ type: "text", text: "fn_skills_install aborted." }], + isError: true, + details: { error: "aborted" }, + }; + } + const child = spawn("npx", npxArgs, { cwd: resolveProjectRoot(ctx.cwd), stdio: "pipe", @@ -5329,27 +5400,48 @@ export default function kbExtension(pi: ExtensionAPI) { }); let stderr = ""; - child.stdout?.on("data", () => {}); - child.stderr?.on("data", (data) => { stderr += data.toString(); }); - const exitCode = await new Promise((resolve) => { - child.on("exit", (code) => { - resolve(code ?? 1); - }); - child.on("error", () => { - resolve(1); - }); - }); + const killChild = () => { + try { + child.kill("SIGTERM"); + } catch { + /* ignore */ + } + try { + child.kill("SIGKILL"); + } catch { + /* ignore */ + } + }; + const onAbort = () => killChild(); + signal?.addEventListener("abort", onAbort, { once: true }); + + let exitCode: number; try { - // Always dispose the child process - child.kill(); - } catch { - // Ignore errors during cleanup + exitCode = await new Promise((resolve) => { + child.on("exit", (code) => { + resolve(code ?? 1); + }); + child.on("error", () => { + resolve(1); + }); + }); + } finally { + signal?.removeEventListener("abort", onAbort); + killChild(); + } + + if (signal?.aborted) { + return { + content: [{ type: "text", text: "fn_skills_install aborted." }], + isError: true, + details: { error: "aborted", stderr }, + }; } if (exitCode !== 0) {