diff --git a/packages/engine/src/__tests__/merge-error-recovery.test.ts b/packages/engine/src/__tests__/merge-error-recovery.test.ts index d6f9a8a702..5aa7b746cd 100644 --- a/packages/engine/src/__tests__/merge-error-recovery.test.ts +++ b/packages/engine/src/__tests__/merge-error-recovery.test.ts @@ -1124,13 +1124,15 @@ describe("ProjectEngine merge error recovery", () => { }); /* - FNXC:MergeQueue 2026-07-15-09:41: + FNXC:MergeQueue 2026-07-15-09:41 / 10:05: Repro for board-wide hung merge pump: active AI merge ignores AbortSignal (wedged tool), operator pauses the card, and without an outer race drainMergeQueue never settles so no later task gets status=merging. + Generation latch: FN-next must not start until the orphan wedged body settles. */ - it("unblocks the merge pump when a paused active merge ignores abort", async () => { + it("unblocks the merge pump when a paused active merge ignores abort, and waits for orphan body settle", async () => { const listeners = new Map void>>(); let wedgedPaused = false; - const store = makeStore(); + // Empty listedTasks so startup merge sweep does not race the test. + const store = makeStore({ listedTasks: [], tasks: [] }); store.on = vi.fn((event: string, listener: (...args: unknown[]) => void) => { const set = listeners.get(event) ?? new Set(); set.add(listener); @@ -1144,17 +1146,24 @@ describe("ProjectEngine merge error recovery", () => { mergeRetries: 0, }), ); + store.listTasks = vi.fn(async () => []); let wedgedStarted = false; + let releaseWedged: (() => void) | undefined; const disposeSession = vi.fn(); + const startedOrder: string[] = []; + vi.mocked(runAiMerge).mockReset(); vi.mocked(runAiMerge).mockImplementation(async (...args: unknown[]) => { const taskId = args[2] as string; const options = args[3] as { signal?: AbortSignal; onSession?: (session: { dispose: () => void }) => void }; options.onSession?.({ dispose: disposeSession }); + startedOrder.push(taskId); if (taskId === "FN-wedged") { wedgedStarted = true; - // Hang forever — do not observe abort (matches wedged agent tool). - await new Promise(() => {}); + // Hang until test releases — ignores abort signal (wedged agent tool). + await new Promise((resolve) => { + releaseWedged = resolve; + }); } return { merged: true, @@ -1168,6 +1177,8 @@ describe("ProjectEngine merge error recovery", () => { mergeRunning: boolean; activeMergeTaskId: string | null; mergeActive: Set; + mergeBodyInFlight: Promise | null; + mergeBodySettleTimeoutMs: number; enqueueMerge: (taskId: string) => void; }; @@ -1193,8 +1204,16 @@ describe("ProjectEngine merge error recovery", () => { expect(privateEngine.mergeRunning).toBe(false); }); + // Orphan body still in flight — next generation must wait. + expect(privateEngine.mergeBodyInFlight).not.toBeNull(); privateEngine.enqueueMerge("FN-next"); + await Promise.resolve(); + await Promise.resolve(); + await new Promise((r) => setTimeout(r, 20)); + expect(startedOrder).toEqual(["FN-wedged"]); + // Settle orphan body → next merge may start. + releaseWedged?.(); await vi.waitFor(() => { expect(runAiMerge).toHaveBeenCalledWith( expect.anything(), @@ -1203,7 +1222,63 @@ describe("ProjectEngine merge error recovery", () => { expect.anything(), ); }); + expect(startedOrder).toEqual(["FN-wedged", "FN-next"]); await engine.stop(); }); + + it("PR merge path races abort so pause unblocks the pump", async () => { + const listeners = new Map void>>(); + let prStarted = false; + let releasePr: (() => void) | undefined; + const store = makeStore({ listedTasks: [], tasks: [] }); + store.on = vi.fn((event: string, listener: (...args: unknown[]) => void) => { + const set = listeners.get(event) ?? new Set(); + set.add(listener); + listeners.set(event, set); + }); + store.getTask = vi.fn(async (id: string) => + makeTask({ id, paused: false, status: null, mergeRetries: 0 }), + ); + store.listTasks = vi.fn(async () => []); + + const processPullRequestMerge = vi.fn(async () => { + prStarted = true; + await new Promise((resolve) => { + releasePr = resolve; + }); + return "merged" as const; + }); + + const engine = createEngine(store, { + getMergeStrategy: () => "pull-request", + processPullRequestMerge, + }); + const privateEngine = engine as unknown as { + activeMergeTaskId: string | null; + mergeRunning: boolean; + enqueueMerge: (taskId: string) => void; + }; + + await engine.start(); + privateEngine.enqueueMerge("FN-pr"); + + await vi.waitFor(() => { + expect(prStarted).toBe(true); + expect(privateEngine.activeMergeTaskId).toBe("FN-pr"); + }); + + const updatedHandlers = [...(listeners.get("task:updated") ?? [])]; + for (const handler of updatedHandlers) { + await handler({ id: "FN-pr", column: "in-review", paused: true }); + } + + await vi.waitFor(() => { + expect(privateEngine.activeMergeTaskId).toBeNull(); + expect(privateEngine.mergeRunning).toBe(false); + }); + + releasePr?.(); + await engine.stop(); + }); }); diff --git a/packages/engine/src/__tests__/merge-reclaim-policy.test.ts b/packages/engine/src/__tests__/merge-reclaim-policy.test.ts new file mode 100644 index 0000000000..c12e918e68 --- /dev/null +++ b/packages/engine/src/__tests__/merge-reclaim-policy.test.ts @@ -0,0 +1,105 @@ +import { describe, expect, it } from "vitest"; +import { + canStartNextMergeBody, + DEFAULT_MERGING_PHASE_SILENCE_FLOOR_MS, + resolveMergingPhaseSilenceFloorMs, + shouldReclaimWedgedMerge, +} from "../merge-reclaim-policy.js"; + +/* +FNXC:MergeQueue 2026-07-15-10:05: +Pure-policy unit tests for status-aware silence reclaim and generation settle gating. +*/ + +describe("merge-reclaim-policy", () => { + const stuck = 15 * 60_000; + + describe("shouldReclaimWedgedMerge", () => { + it("does not reclaim below stuckTimeout for any status", () => { + expect( + shouldReclaimWedgedMerge({ status: "reviewing", silenceMs: stuck - 1, stuckTimeoutMs: stuck }), + ).toBe(false); + expect( + shouldReclaimWedgedMerge({ status: "merging", silenceMs: stuck - 1, stuckTimeoutMs: stuck }), + ).toBe(false); + expect( + shouldReclaimWedgedMerge({ status: null, silenceMs: stuck - 1, stuckTimeoutMs: stuck }), + ).toBe(false); + }); + + it("reclaims reviewing after stuckTimeout silence (post-squash hang shape)", () => { + expect( + shouldReclaimWedgedMerge({ status: "reviewing", silenceMs: stuck, stuckTimeoutMs: stuck }), + ).toBe(true); + expect( + shouldReclaimWedgedMerge({ status: "reviewing", silenceMs: stuck + 60_000, stuckTimeoutMs: stuck }), + ).toBe(true); + }); + + it("does not false-reclaim merging-phase silence at stuckTimeout alone", () => { + // Monorepo single bash can exceed stuckTimeout without agent logs. + expect( + shouldReclaimWedgedMerge({ status: "merging", silenceMs: stuck, stuckTimeoutMs: stuck }), + ).toBe(false); + expect( + shouldReclaimWedgedMerge({ status: "merging-pr", silenceMs: stuck, stuckTimeoutMs: stuck }), + ).toBe(false); + expect( + shouldReclaimWedgedMerge({ status: "merging-fix", silenceMs: stuck, stuckTimeoutMs: stuck }), + ).toBe(false); + }); + + it("reclaims merging-phase only after the higher silence floor", () => { + const floor = resolveMergingPhaseSilenceFloorMs(stuck); + expect(floor).toBe(DEFAULT_MERGING_PHASE_SILENCE_FLOOR_MS); + expect( + shouldReclaimWedgedMerge({ + status: "merging", + silenceMs: floor - 1, + stuckTimeoutMs: stuck, + }), + ).toBe(false); + expect( + shouldReclaimWedgedMerge({ + status: "merging", + silenceMs: floor, + stuckTimeoutMs: stuck, + }), + ).toBe(true); + }); + + it("reclaims null-status dead pump after stuckTimeout (identity without merge badge)", () => { + expect( + shouldReclaimWedgedMerge({ status: null, silenceMs: stuck, stuckTimeoutMs: stuck }), + ).toBe(true); + }); + + it("respects configured merging silence floor override", () => { + expect( + shouldReclaimWedgedMerge({ + status: "merging", + silenceMs: 20 * 60_000, + stuckTimeoutMs: stuck, + mergingSilenceFloorMs: 20 * 60_000, + }), + ).toBe(true); + expect( + shouldReclaimWedgedMerge({ + status: "merging", + silenceMs: 19 * 60_000, + stuckTimeoutMs: stuck, + mergingSilenceFloorMs: 20 * 60_000, + }), + ).toBe(false); + }); + }); + + describe("canStartNextMergeBody", () => { + it("allows a new body only when no prior body is in flight", () => { + expect(canStartNextMergeBody(null)).toBe(true); + expect(canStartNextMergeBody(undefined)).toBe(true); + const pending = new Promise(() => {}); + expect(canStartNextMergeBody(pending)).toBe(false); + }); + }); +}); diff --git a/packages/engine/src/__tests__/merger-plugin-runner-wiring.test.ts b/packages/engine/src/__tests__/merger-plugin-runner-wiring.test.ts index 939a935b67..76402688fe 100644 --- a/packages/engine/src/__tests__/merger-plugin-runner-wiring.test.ts +++ b/packages/engine/src/__tests__/merger-plugin-runner-wiring.test.ts @@ -2,15 +2,58 @@ import { readFileSync } from "node:fs"; import { dirname, resolve } from "node:path"; import { fileURLToPath } from "node:url"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import type { Settings } from "@fusion/core"; import * as fusionCore from "@fusion/core"; import { createResolvedAgentSession } from "../agent-session-helpers.js"; +import { makePrResponseAgentRunner } from "../pr-response-run-ops.js"; const __dirname = dirname(fileURLToPath(import.meta.url)); /* FNXC:GrokCliRouting 2026-07-15-09:45: Auto-merge was failing with "Grok CLI models require the bundled Grok CLI runtime" while dashboard chat worked, because project-engine's runAiMerge options omitted pluginRunner. ChatManager already receives engine.getPluginRunner(); the merge door must forward the same runner so createResolvedAgentSession can resolve getRuntimeById("grok") for grok-cli/no-key selections. + +FNXC:GrokCliRouting 2026-07-15-09:58: +Session-advisor and PR-response createResolvedAgentSession paths must also forward PluginRunner so grok-cli/no-key selections resolve via getRuntimeById("grok") instead of dual-remediation error or pi fallthrough. */ + +/** Stub PluginRunner that serves a grok plugin runtime for wiring assertions. */ +function makeGrokPluginRunnerStub() { + const createSession = vi.fn().mockResolvedValue({ + session: { + model: "grok-4.5", + messages: [], + dispose: vi.fn(), + }, + }); + const grokRuntime = { + id: "grok", + name: "Grok Runtime", + createSession, + promptWithFallback: vi.fn(), + describeModel: vi.fn(() => "grok/grok-4.5"), + }; + const registration = { + pluginId: "fusion-plugin-grok-runtime", + runtime: { + metadata: { runtimeId: "grok", name: "Grok Runtime" }, + factory: vi.fn().mockResolvedValue(grokRuntime), + }, + }; + const pluginRunner = { + getPluginRuntimes: vi.fn().mockReturnValue([registration]), + getRuntimeById: vi.fn().mockReturnValue(registration), + createRuntimeContext: vi.fn().mockResolvedValue({ + pluginId: "fusion-plugin-grok-runtime", + taskStore: {}, + settings: {}, + logger: { info: vi.fn(), warn: vi.fn(), error: vi.fn(), debug: vi.fn() }, + emitEvent: vi.fn(), + }), + }; + return { pluginRunner, createSession }; +} + describe("AI merge PluginRunner wiring for Grok CLI", () => { beforeEach(() => { vi.restoreAllMocks(); @@ -34,38 +77,7 @@ describe("AI merge PluginRunner wiring for Grok CLI", () => { it("createResolvedAgentSession routes merger grok-cli selections through the provided PluginRunner", async () => { vi.spyOn(fusionCore, "isGrokApiKeyFusionVisible").mockReturnValue(false); - const createSession = vi.fn().mockResolvedValue({ - session: { - model: "grok-4.5", - messages: [], - dispose: vi.fn(), - }, - }); - const grokRuntime = { - id: "grok", - name: "Grok Runtime", - createSession, - promptWithFallback: vi.fn(), - describeModel: vi.fn(() => "grok/grok-4.5"), - }; - const registration = { - pluginId: "fusion-plugin-grok-runtime", - runtime: { - metadata: { runtimeId: "grok", name: "Grok Runtime" }, - factory: vi.fn().mockResolvedValue(grokRuntime), - }, - }; - const pluginRunner = { - getPluginRuntimes: vi.fn().mockReturnValue([registration]), - getRuntimeById: vi.fn().mockReturnValue(registration), - createRuntimeContext: vi.fn().mockResolvedValue({ - pluginId: "fusion-plugin-grok-runtime", - taskStore: {}, - settings: {}, - logger: { info: vi.fn(), warn: vi.fn(), error: vi.fn(), debug: vi.fn() }, - emitEvent: vi.fn(), - }), - }; + const { pluginRunner, createSession } = makeGrokPluginRunnerStub(); const result = await createResolvedAgentSession({ sessionPurpose: "merger", @@ -94,3 +106,91 @@ describe("AI merge PluginRunner wiring for Grok CLI", () => { })).rejects.toThrow(/Install and enable the Grok CLI runtime plugin, or set GROK_API_KEY/); }); }); + +describe("Session advisor + PR response PluginRunner wiring for Grok CLI", () => { + beforeEach(() => { + vi.restoreAllMocks(); + }); + + afterEach(() => { + vi.restoreAllMocks(); + }); + + it("project-engine session advisor forwards pluginRunner: this.getPluginRunner()", () => { + const source = readFileSync(resolve(__dirname, "../project-engine.ts"), "utf8"); + // Anchor on the session-advisor agentFactory complete() createResolvedAgentSession call. + const advisorAnchor = source.indexOf("session advisor complete failed"); + // Search backwards from the warn log for the createResolvedAgentSession block. + const sessionCreateIndex = source.lastIndexOf("createResolvedAgentSession({", advisorAnchor); + const pluginRunnerIndex = source.indexOf("pluginRunner: this.getPluginRunner()", sessionCreateIndex); + const nextCreateIndex = source.indexOf("createResolvedAgentSession({", sessionCreateIndex + 1); + + expect(sessionCreateIndex).toBeGreaterThanOrEqual(0); + expect(pluginRunnerIndex).toBeGreaterThan(sessionCreateIndex); + // The pluginRunner must land inside this advisor createResolvedAgentSession call. + if (nextCreateIndex >= 0) { + expect(pluginRunnerIndex).toBeLessThan(nextCreateIndex); + } + }); + + it("makePrResponseAgentRunner forwards pluginRunner into createResolvedAgentSession", () => { + const source = readFileSync(resolve(__dirname, "../pr-response-run-ops.ts"), "utf8"); + const fnIndex = source.indexOf("export function makePrResponseAgentRunner("); + const createIndex = source.indexOf("createResolvedAgentSession({", fnIndex); + const pluginRunnerParamIndex = source.indexOf("pluginRunner?:", fnIndex); + const pluginRunnerArgIndex = source.indexOf("pluginRunner,", createIndex); + const nextFnIndex = source.indexOf("export function", fnIndex + 1); + + expect(fnIndex).toBeGreaterThanOrEqual(0); + expect(pluginRunnerParamIndex).toBeGreaterThan(fnIndex); + expect(createIndex).toBeGreaterThan(fnIndex); + expect(pluginRunnerArgIndex).toBeGreaterThan(createIndex); + if (nextFnIndex >= 0) { + expect(pluginRunnerArgIndex).toBeLessThan(nextFnIndex); + } + }); + + it("buildPrNodeDeps / in-process-runtime thread pluginRunner into PR respond", () => { + const prNodes = readFileSync(resolve(__dirname, "../pr-nodes.ts"), "utf8"); + const runtime = readFileSync(resolve(__dirname, "../runtimes/in-process-runtime.ts"), "utf8"); + + expect(prNodes).toMatch(/buildRespondCallback\([\s\S]*pluginRunner/); + expect(prNodes).toMatch(/makePrResponseAgentRunner\([\s\S]*pluginRunner\)/); + expect(prNodes).toMatch(/export function buildPrNodeDeps\([\s\S]*pluginRunner\?:/); + expect(runtime).toMatch(/buildPrNodeDeps\(\(\) => this\.taskStore, prNodeGithubOps, this\.pluginRunner\)/); + }); + + it("makePrResponseAgentRunner with stub PluginRunner + grok-cli resolves runtime id grok", async () => { + vi.spyOn(fusionCore, "isGrokApiKeyFusionVisible").mockReturnValue(false); + + // Keep the post-session prompt path inert so the test only asserts runtime routing. + const pi = await import("../pi.js"); + const usageLimit = await import("../usage-limit-detector.js"); + vi.spyOn(pi, "promptWithFallback").mockResolvedValue(undefined as never); + vi.spyOn(usageLimit, "checkSessionError").mockReturnValue(undefined as never); + + const { pluginRunner, createSession } = makeGrokPluginRunnerStub(); + const settings = { + mergerProvider: "grok-cli", + mergerModelId: "grok-4.5", + defaultProvider: "grok-cli", + defaultModelId: "grok-4.5", + } as Settings; + + const runAgent = makePrResponseAgentRunner( + settings, + "FN-test", + "/tmp/fusion-pr-response", + undefined, + pluginRunner as never, + ); + await runAgent({ + prompt: "Resolve review threads", + systemPrompt: "System", + threads: [{ id: "thread-1" }], + }); + + expect(pluginRunner.getRuntimeById).toHaveBeenCalledWith("grok"); + expect(createSession).toHaveBeenCalled(); + }); +}); diff --git a/packages/engine/src/__tests__/self-healing-wedged-active-merge.test.ts b/packages/engine/src/__tests__/self-healing-wedged-active-merge.test.ts index 5beefa5150..899157e349 100644 --- a/packages/engine/src/__tests__/self-healing-wedged-active-merge.test.ts +++ b/packages/engine/src/__tests__/self-healing-wedged-active-merge.test.ts @@ -1,10 +1,12 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import type { Settings, TaskStore } from "@fusion/core"; import { SelfHealingManager } from "../self-healing.js"; +import { DEFAULT_MERGING_PHASE_SILENCE_FLOOR_MS } from "../merge-reclaim-policy.js"; /* -FNXC:MergeQueue 2026-07-15-09:50: -Self-healing must reclaim a wedged in-process active merge when the AI merge review pass hangs (status=reviewing / merger agent silence) so the single-flight pump is not stuck with no merging badge on the board. +FNXC:MergeQueue 2026-07-15-10:05: +Self-healing must reclaim a wedged in-process active merge when the AI merge review pass hangs +(status=reviewing / merger agent silence) without false-reclaiming a live merging-phase long bash. */ function createTask(id: string, overrides: Record = {}) { @@ -28,12 +30,14 @@ describe("SelfHealingManager wedged active merge recovery", () => { let tasks: Map>; let store: TaskStore; let agentLogs: Array<{ agent?: string; timestamp?: string; type?: string; text?: string }>; + let auditEvents: Array>; beforeEach(() => { vi.useFakeTimers(); vi.setSystemTime(new Date("2026-01-01T01:00:00.000Z")); tasks = new Map(); agentLogs = []; + auditEvents = []; const wedged = createTask("FN-WEDGE"); tasks.set("FN-WEDGE", wedged); @@ -58,6 +62,9 @@ describe("SelfHealingManager wedged active merge recovery", () => { }), logEntry: vi.fn().mockResolvedValue(undefined), getAgentLogs: vi.fn().mockImplementation(async () => agentLogs), + recordRunAuditEvent: vi.fn().mockImplementation(async (event: Record) => { + auditEvents.push(event); + }), getCompletionHandoffAcceptedMarker: vi.fn().mockReturnValue(null), parseFileScopeFromPrompt: vi.fn().mockResolvedValue([]), } as unknown as TaskStore; @@ -85,8 +92,8 @@ describe("SelfHealingManager wedged active merge recovery", () => { manager.stop(); }); - it("reclaims active merge after merger agent silence past stuck timeout", async () => { - // Last merger activity 30 minutes ago; stuck timeout is 15 minutes. + it("reclaims reviewing after merger agent silence past stuck timeout and emits audit", async () => { + // Last merger activity 30 minutes ago; stuck timeout is 15 minutes; status=reviewing. agentLogs = [ { agent: "merger", timestamp: "2026-01-01T00:30:00.000Z", type: "tool", text: "fn_task_show" }, { agent: "executor", timestamp: "2026-01-01T00:59:00.000Z", type: "text", text: "noise" }, @@ -113,6 +120,63 @@ describe("SelfHealingManager wedged active merge recovery", () => { "FN-WEDGE", expect.stringContaining("wedged active merge reclaimed"), ); + expect(auditEvents).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + taskId: "FN-WEDGE", + mutationType: "task:reconcile-wedged-active-merge", + metadata: expect.objectContaining({ + taskId: "FN-WEDGE", + reason: "wedged-active-merge-no-merger-progress", + limitMs: 15 * 60_000, + status: "reviewing", + }), + }), + ]), + ); + manager.stop(); + }); + + it("does not false-reclaim merging-phase silence at stuckTimeout alone", async () => { + tasks.set("FN-WEDGE", createTask("FN-WEDGE", { status: "merging" })); + // Silence = 30m > stuck 15m but < 45m floor + agentLogs = [ + { agent: "merger", timestamp: "2026-01-01T00:30:00.000Z", type: "tool", text: "bash" }, + ]; + const abortActiveMerge = vi.fn().mockReturnValue(true); + const manager = new SelfHealingManager(store, { + rootDir: "/tmp/test-project", + getActiveMergeTaskId: () => "FN-WEDGE", + getActiveMergeStartedAtMs: () => Date.parse("2026-01-01T00:00:00.000Z"), + abortActiveMerge, + }); + + const recovered = await manager.recoverWedgedActiveMerge(); + expect(recovered).toBe(0); + expect(abortActiveMerge).not.toHaveBeenCalled(); + manager.stop(); + }); + + it("reclaims merging-phase only after the higher silence floor", async () => { + tasks.set("FN-WEDGE", createTask("FN-WEDGE", { status: "merging" })); + // System time 01:00; silence from 00:00 => 60m > 45m floor + agentLogs = [ + { agent: "merger", timestamp: "2026-01-01T00:00:00.000Z", type: "tool", text: "bash" }, + ]; + expect(DEFAULT_MERGING_PHASE_SILENCE_FLOOR_MS).toBe(45 * 60_000); + const abortActiveMerge = vi.fn().mockReturnValue(true); + const manager = new SelfHealingManager(store, { + rootDir: "/tmp/test-project", + getActiveMergeTaskId: () => "FN-WEDGE", + getActiveMergeStartedAtMs: () => Date.parse("2026-01-01T00:00:00.000Z"), + abortActiveMerge, + enqueueMerge: vi.fn().mockReturnValue(true), + clearMergeActive: vi.fn(), + }); + + const recovered = await manager.recoverWedgedActiveMerge(); + expect(recovered).toBe(1); + expect(abortActiveMerge).toHaveBeenCalled(); manager.stop(); }); @@ -139,7 +203,8 @@ describe("SelfHealingManager wedged active merge recovery", () => { manager.stop(); }); - it("reclaims when agent logs are empty but claim wall-clock exceeds stuck timeout", async () => { + it("reclaims when agent logs are empty but claim wall-clock exceeds stuck timeout (dead pump)", async () => { + tasks.set("FN-WEDGE", createTask("FN-WEDGE", { status: null })); agentLogs = []; const abortActiveMerge = vi.fn().mockReturnValue(true); const manager = new SelfHealingManager(store, { diff --git a/packages/engine/src/merge-reclaim-policy.ts b/packages/engine/src/merge-reclaim-policy.ts new file mode 100644 index 0000000000..e060dd6b3a --- /dev/null +++ b/packages/engine/src/merge-reclaim-policy.ts @@ -0,0 +1,77 @@ +/* +FNXC:MergeQueue 2026-07-15-10:05: +Pure decision helpers for wedged-merge reclaim and merge-generation settle. +Keeps status-aware silence policy and generation gating unit-testable without booting ProjectEngine. +*/ + +/** Transient merge-activity statuses set by the AI/PR merge pipeline. */ +export const MERGE_ACTIVITY_STATUSES = new Set([ + "merging", + "merging-pr", + "merging-fix", + "reviewing", +]); + +/** Statuses that mean the merge agent is still in verify/land work (long bash tools are normal). */ +export const MERGING_PHASE_STATUSES = new Set(["merging", "merging-pr", "merging-fix"]); + +/** + * Minimum silence before reclaiming a live merge still in the `merging*` phase. + * Monorepo install/test can exceed default taskStuckTimeoutMs as a single tool call with no agent logs. + */ +export const DEFAULT_MERGING_PHASE_SILENCE_FLOOR_MS = 45 * 60_000; + +export function resolveMergingPhaseSilenceFloorMs( + stuckTimeoutMs: number, + configuredFloorMs?: number | null, +): number { + const floor = + configuredFloorMs != null && Number.isFinite(configuredFloorMs) && configuredFloorMs > 0 + ? configuredFloorMs + : DEFAULT_MERGING_PHASE_SILENCE_FLOOR_MS; + return Math.max(stuckTimeoutMs, floor); +} + +export type WedgedMergeReclaimInput = { + /** Task status while the process still owns activeMergeTaskId. */ + status: string | null | undefined; + /** ms since last merger agent-log activity (or claim wall-clock fallback). */ + silenceMs: number; + /** Project taskStuckTimeoutMs. */ + stuckTimeoutMs: number; + /** Optional override for merging-phase silence floor. */ + mergingSilenceFloorMs?: number | null; +}; + +/* +FNXC:MergeQueue 2026-07-15-10:05: +Reclaim policy: +- reviewing (post-squash AI review): reclaim after stuckTimeoutMs of merger silence (the original hang shape). +- merging/merging-pr/merging-fix: require a higher silence floor so a single long bash (pnpm test) is not false-reclaimed. +- null/other with a live active owner: treat as a dead pump (identity without progress) and reclaim after stuckTimeoutMs. +*/ +export function shouldReclaimWedgedMerge(input: WedgedMergeReclaimInput): boolean { + const { silenceMs, stuckTimeoutMs } = input; + if (!Number.isFinite(stuckTimeoutMs) || stuckTimeoutMs <= 0) return false; + if (!Number.isFinite(silenceMs) || silenceMs < 0) return false; + if (silenceMs < stuckTimeoutMs) return false; + + const status = input.status ?? null; + if (status === "reviewing") return true; + + if (status != null && MERGING_PHASE_STATUSES.has(status)) { + const floor = resolveMergingPhaseSilenceFloorMs(stuckTimeoutMs, input.mergingSilenceFloorMs); + return silenceMs >= floor; + } + + // Dead pump: active owner but no merge-activity status (cleared/orphaned identity). + return true; +} + +/** + * Whether a new merge body may start given an outstanding prior body promise. + * Pure: true only when no prior body is tracked. + */ +export function canStartNextMergeBody(priorBodyInFlight: Promise | null | undefined): boolean { + return priorBodyInFlight == null; +} diff --git a/packages/engine/src/pr-nodes.ts b/packages/engine/src/pr-nodes.ts index b22dd3e167..9e9371720e 100644 --- a/packages/engine/src/pr-nodes.ts +++ b/packages/engine/src/pr-nodes.ts @@ -216,6 +216,11 @@ export function buildRespondCallback( getStore: () => PrNodeStore, ops: PrRespondGithubOps, audit?: PrNodeDeps["audit"], + /* + * FNXC:GrokCliRouting 2026-07-15-09:58: + * Forward the engine PluginRunner into the PR-response agent runner so grok-cli/no-key models use the same plugin-runtime path as chat/executor/merge. + */ + pluginRunner?: import("./plugin-runner.js").PluginRunner, ): NonNullable { const gitOps = makePrResponseGitOps(ops.getCwd); return async ({ entity }) => { @@ -243,7 +248,7 @@ export function buildRespondCallback( const taskId = ops.getTaskId(entity); const cwd = ops.getCwd(entity); - const runAgent = makePrResponseAgentRunner(settings, taskId, cwd, fullStore); + const runAgent = makePrResponseAgentRunner(settings, taskId, cwd, fullStore, pluginRunner); const result = await runPrResponseRun({ entity, @@ -276,12 +281,20 @@ export type { PrReviewThread, PrPushResult }; * GitHub ops. Used by the runtime/executor wiring so the CLI layer stays free of * any store reference and the engine never imports the dashboard client. */ -export function buildPrNodeDeps(getStore: () => PrNodeStore, ops: PrNodeGithubOps): PrNodeDeps { +export function buildPrNodeDeps( + getStore: () => PrNodeStore, + ops: PrNodeGithubOps, + /* + * FNXC:GrokCliRouting 2026-07-15-09:58: + * Optional PluginRunner from the in-process runtime so PR-respond sessions can resolve grok-cli via getRuntimeById("grok"). + */ + pluginRunner?: import("./plugin-runner.js").PluginRunner, +): PrNodeDeps { // U5: when the CLI injects `respondOps`, build the real review-response run // callback here (the engine binds the store + audit). An explicit `respond` // takes precedence (tests/specialized wiring); absent both → inert default. const respond = ops.respond - ?? (ops.respondOps ? buildRespondCallback(getStore, ops.respondOps, ops.audit) : undefined); + ?? (ops.respondOps ? buildRespondCallback(getStore, ops.respondOps, ops.audit, pluginRunner) : undefined); return { getStore, resolvePrSource: ops.resolvePrSource, diff --git a/packages/engine/src/pr-response-run-ops.ts b/packages/engine/src/pr-response-run-ops.ts index f9dcab1e04..62a02801ce 100644 --- a/packages/engine/src/pr-response-run-ops.ts +++ b/packages/engine/src/pr-response-run-ops.ts @@ -88,6 +88,11 @@ export function makePrResponseAgentRunner( taskId: string, cwd: string, store?: TaskStore, + /* + * FNXC:GrokCliRouting 2026-07-15-09:58: + * PR-response createResolvedAgentSession must share chat/executor plugin-runtime injection so grok-cli/no-key merger models resolve via getRuntimeById("grok") instead of dual-remediation error or pi fallthrough. + */ + pluginRunner?: import("./plugin-runner.js").PluginRunner, ): (input: { prompt: string; systemPrompt: string; @@ -115,6 +120,7 @@ export function makePrResponseAgentRunner( ].join("\n"); const { session } = await createResolvedAgentSession({ sessionPurpose: "merger", + pluginRunner, cwd, systemPrompt: fullSystem, tools: "coding", diff --git a/packages/engine/src/project-engine.ts b/packages/engine/src/project-engine.ts index 8421821655..3f7eb011be 100644 --- a/packages/engine/src/project-engine.ts +++ b/packages/engine/src/project-engine.ts @@ -66,6 +66,7 @@ import { sweepStaleAutostashes, VerificationError } from "./merger.js"; import { runAiMerge, landWorkspaceTask, WorkspacePartialLandError, WorkspaceRepoLandBusyError } from "./merger-ai.js"; import { promoteBranchGroup, type BranchGroupPromotionResult, type CreateGroupPrFn, type SyncGroupPrFn } from "./group-merge-coordinator.js"; import { PRIORITY_MERGE } from "./concurrency.js"; +import { canStartNextMergeBody } from "./merge-reclaim-policy.js"; import { registerProjectVerificationLimit, unregisterProjectVerificationLimit, @@ -462,6 +463,11 @@ export class ProjectEngine { private activeMergeTaskId: string | null = null; /** Wall-clock when `activeMergeTaskId` was claimed; self-healing uses this when agent logs are silent. */ private activeMergeStartedAtMs: number | null = null; + /* + FNXC:MergeQueue 2026-07-15-10:05: + Tracks the underlying merge body promise (not the abort race). After force-abort the race rejects so drain can continue, but the orphan body may still be mid-tool. The next claim waits for this latch so two runAiMerge/land paths cannot advance main concurrently. + */ + private mergeBodyInFlight: Promise | null = null; private mergeAbortController: AbortController | null = null; private mergeRetryTimer: ReturnType | null = null; private autostashSweepTimer: ReturnType | null = null; @@ -647,6 +653,58 @@ export class ProjectEngine { } } + /* + FNXC:MergeQueue 2026-07-15-10:05: + Bound how long drain waits for an orphan body after abort. If the agent ignores abort forever, a hard settle timeout releases the latch so the board does not stay permanently blocked — last-resort only; prefer clean body settle. Overridable in tests via mergeBodySettleTimeoutMs. + */ + private mergeBodySettleTimeoutMs = 60_000; + + private trackMergeBody(body: Promise): Promise { + const tracked = body.finally(() => { + if (this.mergeBodyInFlight === tracked) { + this.mergeBodyInFlight = null; + } + }); + this.mergeBodyInFlight = tracked; + return tracked; + } + + private async awaitPriorMergeBodySettle(): Promise { + const prior = this.mergeBodyInFlight; + if (canStartNextMergeBody(prior)) return; + let timeoutHandle: ReturnType | undefined; + const timeoutMs = this.mergeBodySettleTimeoutMs; + const timeout = new Promise<"timeout">((resolve) => { + timeoutHandle = setTimeout(() => resolve("timeout"), timeoutMs); + timeoutHandle.unref?.(); + }); + try { + const winner = await Promise.race([ + prior!.then(() => "settled" as const).catch(() => "settled" as const), + timeout, + ]); + if (winner === "timeout") { + runtimeLog.warn( + `Prior merge body did not settle within ${timeoutMs}ms after abort — releasing latch for next generation`, + ); + if (this.mergeBodyInFlight === prior) { + this.mergeBodyInFlight = null; + } + } + } finally { + if (timeoutHandle) clearTimeout(timeoutHandle); + } + } + + /** + * Race a merge body with abort, while tracking the underlying body so the next + * generation cannot start until the orphan work settles. + */ + private runAbortableMergeBody(bodyFactory: () => Promise, signal: AbortSignal, taskId: string): Promise { + const body = this.trackMergeBody(bodyFactory()); + return this.raceMergeWithAbort(body, signal, taskId); + } + /* FNXC:Workspace 2026-06-22-16:40 (Phase D P1 TOCTOU — merge-queue dispatch blind spot): A workspace task is "merge-pending" if it sits ANYWHERE in this engine's in-memory merge @@ -1078,6 +1136,7 @@ export class ProjectEngine { this.mergeAbortController = null; this.activeMergeTaskId = null; this.activeMergeStartedAtMs = null; + this.mergeBodyInFlight = null; this.pausedReviewTaskIds.clear(); const queuedTaskIds = [...this.mergeQueue]; @@ -2772,6 +2831,10 @@ export class ProjectEngine { investigative-only; systemPrompt is the advisor contract; user batch is the session-update delta only. */ + /* + FNXC:GrokCliRouting 2026-07-15-09:58: + Session-advisor createResolvedAgentSession must forward the engine PluginRunner so grok-cli advisor models use the same no-visible-key CLI runtime path as chat/executor/merge. + */ const { session } = await createResolvedAgentSession({ sessionPurpose: "reviewer", cwd: String(cwd), @@ -2780,6 +2843,7 @@ export class ProjectEngine { defaultProvider: model.provider, defaultModelId: model.modelId, settings, + pluginRunner: this.getPluginRunner(), }); try { await session.prompt(user); @@ -3409,14 +3473,26 @@ export class ProjectEngine { const mergeCandidate = await store.getTask(taskId).catch(() => null); const routeWorkspaceDirect = !!mergeCandidate && isWorkspaceTask(mergeCandidate); + // FNXC:MergeQueue 2026-07-15-10:05: Wait for any orphan body from a prior abort race before claiming the next generation. + await this.awaitPriorMergeBodySettle(); + if (mergeStrategy === "pull-request" && this.options.processPullRequestMerge && !routeWorkspaceDirect) { - this.claimActiveMerge(taskId); + /* + FNXC:MergeQueue 2026-07-15-10:05: + PR merge dispatch shares the single-flight pump. Race the PR body with abort so pause/reclaim unblocks drainMergeQueue even when processPullRequestMerge ignores cooperative abort. + */ + const abortSignal = this.claimActiveMerge(taskId); runtimeLog.log(`${hasManualResolver ? "Manual" : "Auto"}-merge processing PR flow for ${taskId}...`); - const result = await this.options.processPullRequestMerge( - store, - cwd, + const result = await this.runAbortableMergeBody( + () => + this.options.processPullRequestMerge!( + store, + cwd, + taskId, + (this.runtime as any).worktreePool, + ), + abortSignal, taskId, - (this.runtime as any).worktreePool, ); if (result === "merged") { runtimeLog.log(`${hasManualResolver ? "Manual" : "Auto"}-merge PR merged: ${taskId}`); @@ -3482,8 +3558,10 @@ export class ProjectEngine { /* FNXC:MergeQueue 2026-07-15-09:41: Always race the merge body with the pause/cancel abort signal. Cooperative abort inside runAiMerge is best-effort; without this outer race a wedged agent tool parks drainMergeQueue forever (no merging badge board-wide). + FNXC:MergeQueue 2026-07-15-10:05: + Track the underlying body so abort-race reject does not allow a concurrent second generation while orphan work still runs. */ - return this.raceMergeWithAbort((async () => { + return this.runAbortableMergeBody(async () => { // FNXC:Workspace 2026-06-21-23:40 (Phase C U1, KTD2): // Engine merge dispatch door. A workspace-mode task (non-empty // `workspaceWorktrees`) routes to the per-repo merge loop @@ -3566,7 +3644,7 @@ export class ProjectEngine { allowDirtyLocalCheckoutSync: settings.merger?.allowDirtyLocalCheckoutSync === true, }; return runAiMerge(store, cwd, taskId, mergeOptionsWithSettings); - })(), abortSignal, taskId); + }, abortSignal, taskId); }; let result: MergeResult; diff --git a/packages/engine/src/run-audit.ts b/packages/engine/src/run-audit.ts index d09795567d..ebd23a197f 100644 --- a/packages/engine/src/run-audit.ts +++ b/packages/engine/src/run-audit.ts @@ -542,6 +542,12 @@ export type DatabaseMutationType = * Self-healing must leave file-scope lease queues intact while recording when stale durable Agent.taskId/state drift is cleared. Metadata: { agentId, taskId, taskColumn, agentState, status, blockedBy, overlapBlockedBy, hadFreshRun, hadActiveExecution, reason }. */ | "task:reconcile-stale-agent-assignment" + /** + * FNXC:MergeQueue 2026-07-15-10:05: + * Wedged single-flight merge reclaim. Metadata ids/outcomes-only: + * { taskId, reason, silenceMs?, limitMs, status?, column? }. + */ + | "task:reconcile-wedged-active-merge" /** Metadata: { taskId, branch, worktree, checkedOutBy, executionStartedAt, executionAgeMs, graceMs, liveWorktreeBoundBranch, reason } */ | "task:reclaim-self-owned-branch-conflict-no-action" | "task:orphan-detected-no-action" diff --git a/packages/engine/src/runtimes/in-process-runtime.ts b/packages/engine/src/runtimes/in-process-runtime.ts index 0c68f07f53..269440d5bb 100644 --- a/packages/engine/src/runtimes/in-process-runtime.ts +++ b/packages/engine/src/runtimes/in-process-runtime.ts @@ -723,8 +723,12 @@ export class InProcessRuntime // GitHub ops (createPr/mergePr/respond) + the engine-owned store. The CLI // layer never holds a store reference; the engine binds it here. Absent // ops → undefined → the pr-* node kinds fail closed. + /* + * FNXC:GrokCliRouting 2026-07-15-09:58: + * Thread this.pluginRunner into buildPrNodeDeps so pr-respond agent sessions share chat/executor Grok CLI plugin-runtime injection. + */ prNodes: prNodeGithubOps - ? buildPrNodeDeps(() => this.taskStore, prNodeGithubOps) + ? buildPrNodeDeps(() => this.taskStore, prNodeGithubOps, this.pluginRunner) : undefined, onSliceComplete: (slice) => { void this.scheduler.onSliceComplete(slice); diff --git a/packages/engine/src/self-healing.ts b/packages/engine/src/self-healing.ts index 407753ca90..0c7659e730 100644 --- a/packages/engine/src/self-healing.ts +++ b/packages/engine/src/self-healing.ts @@ -71,6 +71,7 @@ self-healing — a real import cycle. Importing from the predicate module breaks import { isRepoLanded } from "./workspace-land-predicate.js"; import { findAlreadyMergedTaskCommit, getCommitTaskOwnership } from "./already-merged-detector.js"; import { getTaskCompletionBlockerForStore } from "./task-completion.js"; +import { shouldReclaimWedgedMerge } from "./merge-reclaim-policy.js"; export const COMPLETED_BLOCKED_PAUSE_REASON = "completed-work-blocked"; import { advanceIntegrationBranchRef } from "./merger-ref-update-advance.js"; @@ -1911,28 +1912,57 @@ export class SelfHealingManager { } } - private async isActiveMergeWedged(taskId: string, timeoutMs: number): Promise { - if (!Number.isFinite(timeoutMs) || timeoutMs <= 0) return false; - const activeId = this.options.getActiveMergeTaskId?.() ?? null; - if (activeId !== taskId) return false; - + private async measureActiveMergeSilenceMs(taskId: string): Promise { const now = Date.now(); const lastMergerMs = await this.getLastMergerAgentActivityMs(taskId); - if (lastMergerMs != null) { - return now - lastMergerMs >= timeoutMs; - } + if (lastMergerMs != null) return now - lastMergerMs; const startedAt = this.options.getActiveMergeStartedAtMs?.() ?? null; if (startedAt != null && Number.isFinite(startedAt) && startedAt > 0) { - return now - startedAt >= timeoutMs; + return now - startedAt; } - return false; + return null; + } + + /* + FNXC:MergeQueue 2026-07-15-10:05: + Status-aware silence reclaim via shouldReclaimWedgedMerge: reviewing reclaims after stuckTimeout; + merging* requires a higher silence floor so monorepo single-bash verify is not false-reclaimed; + null/other with live owner reclaims as a dead pump. + */ + private async isActiveMergeWedged( + taskId: string, + timeoutMs: number, + status?: string | null, + ): Promise<{ wedged: boolean; silenceMs: number | null }> { + if (!Number.isFinite(timeoutMs) || timeoutMs <= 0) return { wedged: false, silenceMs: null }; + const activeId = this.options.getActiveMergeTaskId?.() ?? null; + if (activeId !== taskId) return { wedged: false, silenceMs: null }; + + const silenceMs = await this.measureActiveMergeSilenceMs(taskId); + if (silenceMs == null) return { wedged: false, silenceMs: null }; + + let resolvedStatus = status; + if (resolvedStatus === undefined) { + const task = await this.store.getTask(taskId).catch(() => null); + resolvedStatus = task?.status ?? null; + } + + return { + wedged: shouldReclaimWedgedMerge({ + status: resolvedStatus, + silenceMs, + stuckTimeoutMs: timeoutMs, + }), + silenceMs, + }; } private async isPastInterruptedMergeGraceAsync(task: Task, timeoutMs: number): Promise { const activeId = this.options.getActiveMergeTaskId?.() ?? null; if (activeId === task.id) { - // Live owner: require merger silence / claim-age, not updatedAt (overseer noise). - return this.isActiveMergeWedged(task.id, timeoutMs); + // Live owner: status-aware merger silence / claim-age, not updatedAt (overseer noise). + const { wedged } = await this.isActiveMergeWedged(task.id, timeoutMs, task.status); + return wedged; } return this.isPastInterruptedMergeGrace(task, timeoutMs); } @@ -3024,16 +3054,23 @@ export class SelfHealingManager { const activeId = this.options.getActiveMergeTaskId?.() ?? null; if (!activeId) return 0; - if (!(await this.isActiveMergeWedged(activeId, timeoutMs))) { + const task = await this.store.getTask(activeId).catch(() => null); + const { wedged, silenceMs } = await this.isActiveMergeWedged(activeId, timeoutMs, task?.status ?? null); + if (!wedged) { return 0; } - const task = await this.store.getTask(activeId).catch(() => null); if (task?.paused) { // Pause should already abort; if identity remains, force-clear the lane. const aborted = this.options.abortActiveMerge?.(activeId, "wedged-active-merge-while-paused") ?? false; if (aborted) { log.warn(`Force-aborted wedged active merge ${activeId} that remained after pause`); + await this.emitWedgedActiveMergeAudit(activeId, { + reason: "wedged-active-merge-while-paused", + silenceMs, + limitMs: timeoutMs, + status: task.status ?? null, + }); return 1; } return 0; @@ -3043,6 +3080,13 @@ export class SelfHealingManager { const aborted = this.options.abortActiveMerge?.(activeId, "wedged-active-merge-left-in-review") ?? false; if (aborted) { log.warn(`Force-aborted wedged active merge ${activeId}: task column is ${task.column}`); + await this.emitWedgedActiveMergeAudit(activeId, { + reason: "wedged-active-merge-left-in-review", + silenceMs, + limitMs: timeoutMs, + status: task.status ?? null, + column: task.column, + }); return 1; } return 0; @@ -3062,6 +3106,12 @@ export class SelfHealingManager { ) .catch(() => undefined); } + await this.emitWedgedActiveMergeAudit(activeId, { + reason: "wedged-active-merge-no-merger-progress", + silenceMs, + limitMs: timeoutMs, + status: task?.status ?? null, + }); if (task && allowsAutoMergeProcessing(task, settings) && !task.paused && task.column === "in-review") { try { this.options.enqueueMerge?.(activeId); @@ -3079,6 +3129,41 @@ export class SelfHealingManager { } } + private async emitWedgedActiveMergeAudit( + taskId: string, + metadata: { + reason: string; + silenceMs: number | null; + limitMs: number; + status: string | null; + column?: string; + }, + ): Promise { + try { + await this.store.recordRunAuditEvent({ + taskId, + agentId: "self-healing", + runId: generateSyntheticRunId("wedged-active-merge", taskId), + domain: "database", + // Ids/outcomes-only: reason enum-ish string, silence/limit counts, status — never prose/prompt. + mutationType: "task:reconcile-wedged-active-merge", + target: taskId, + metadata: { + taskId, + reason: metadata.reason, + silenceMs: metadata.silenceMs ?? undefined, + limitMs: metadata.limitMs, + status: metadata.status, + column: metadata.column, + }, + }); + } catch (err: unknown) { + log.warn( + `Failed to audit wedged active merge reclaim for ${taskId}: ${err instanceof Error ? err.message : String(err)}`, + ); + } + } + async reclaimPrConflicts(): Promise { const tasks = await this.store.listTasks({ slim: true }); const candidates = tasks.filter((task) => {