fix(FN-6043): recover stuck task processing
Fusion-Task-Id: FN-6043
This commit is contained in:
@@ -127,15 +127,16 @@ describe("resolveAgentPrompt", () => {
|
||||
expect(result).toContain("task execution agent");
|
||||
});
|
||||
|
||||
it("built-in executor prompt requires resolving ALL lint and test failures including unrelated", () => {
|
||||
it("built-in executor prompt limits fixes to impacted failures and follow-ups unrelated broad-suite failures", () => {
|
||||
const result = resolveAgentPrompt("executor");
|
||||
// The stricter language must be present to prevent "unrelated failure" deferrals
|
||||
expect(result).toContain("Resolve ALL lint failures and test failures");
|
||||
expect(result).toContain("even if they appear unrelated or pre-existing");
|
||||
expect(result).toContain("do not defer them to a separate task");
|
||||
expect(result).toContain("Keep fixing failures caused by your change");
|
||||
expect(result).toContain("impacted tests");
|
||||
expect(result).toContain("unrelated or pre-existing failures");
|
||||
expect(result).toContain("create/link a follow-up task");
|
||||
expect(result).not.toContain("Resolve ALL lint failures and test failures");
|
||||
});
|
||||
|
||||
it("senior-engineer prompt requires resolving ALL lint and test failures including unrelated", () => {
|
||||
it("senior-engineer prompt limits fixes to impacted failures and follow-ups unrelated broad-suite failures", () => {
|
||||
const config: AgentPromptsConfig = {
|
||||
roleAssignments: {
|
||||
executor: "senior-engineer",
|
||||
@@ -143,9 +144,10 @@ describe("resolveAgentPrompt", () => {
|
||||
};
|
||||
|
||||
const result = resolveAgentPrompt("executor", config);
|
||||
expect(result).toContain("Resolve ALL lint failures and test failures");
|
||||
expect(result).toContain("even if they appear unrelated or pre-existing");
|
||||
expect(result).toContain("do not defer them to a separate task");
|
||||
expect(result).toContain("Lint, tests, and typecheck are also hard quality gates for failures caused by this task");
|
||||
expect(result).toContain("unrelated or pre-existing broad-suite failures");
|
||||
expect(result).toContain("create/link follow-up work");
|
||||
expect(result).not.toContain("Resolve ALL lint failures and test failures");
|
||||
});
|
||||
|
||||
it("built-in executor prompt includes worktree boundary guidance", () => {
|
||||
|
||||
@@ -188,10 +188,11 @@ If a project build command is listed in the prompt, it is a hard completion gate
|
||||
- If the build fails, do NOT call \`task_done()\`; keep working until it passes
|
||||
|
||||
Lint, tests, and typecheck are also hard quality gates:
|
||||
- Keep fixing failures until lint, the configured/full test suite, and typecheck all pass
|
||||
- If the repository exposes a typecheck command, run it and keep fixing failures until it passes
|
||||
- Do not stop at "out of scope" if additional fixes are required to restore green lint, tests, build, or typecheck
|
||||
- **CRITICAL: Resolve ALL lint failures and test failures before completing the task, even if they appear unrelated or pre-existing.** Unrelated failures left unfixed accumulate technical debt and block future integrations. Investigate and fix or suppress them — do not defer them to a separate task.
|
||||
- Keep fixing failures caused by your change until lint, impacted tests, build, and typecheck pass.
|
||||
- If the repository exposes a typecheck command, run it and fix failures caused by your change.
|
||||
- When tests fail, classify whether the failure is caused by your change, a pre-existing defect, an unrelated flaky test, or an outdated test expectation.
|
||||
- If broad workspace verification fails on unrelated or pre-existing failures after impacted checks pass, do NOT expand this task by fixing unrelated areas. Log the evidence, quarantine flakes per project policy, or create/link a follow-up task.
|
||||
- Do not repeatedly rerun a broad failing or hanging workspace command without a new hypothesis and a narrower confirming command.
|
||||
|
||||
## Verification commands — use fn_run_verification
|
||||
|
||||
@@ -200,7 +201,7 @@ The tool prevents your session from being killed by the inactivity watchdog duri
|
||||
|
||||
- Prefer **package-scoped** verification first: e.g. \`pnpm --filter @fusion/<pkg> test\` with \`scope: "package"\`. This is faster and isolated.
|
||||
- For file-specific package tests, use direct Vitest execution with package-relative paths: \`pnpm --filter @fusion/<pkg> exec vitest run src/path/to/test.ts --silent=passed-only --reporter=dot\`. Do not use \`pnpm --filter @fusion/<pkg> test -- --run <files>\`; package test scripts can expand into broad quality suites before the filter is applied.
|
||||
- Only run **workspace-scoped** verification (\`pnpm test\`, \`pnpm lint\`, \`pnpm build\` from root) at the FINAL integration step, when you are about to call \`task_done()\`.
|
||||
- Run **workspace-scoped** verification (\`pnpm test\`, \`pnpm lint\`, \`pnpm build\` from root) only when it is explicitly required by the task/workflow or after impacted/package-scoped checks pass and you are doing final integration.
|
||||
- If you need to run \`pnpm install\` (e.g. you added a new package), use \`fn_run_verification\` with \`scope: "workspace"\` and \`timeoutSec: 600\`.
|
||||
- If a verification command times out, do NOT blindly retry — investigate. Check for hung subprocesses, infinite test loops, or tests waiting on missing dependencies. Use \`node_modules/.modules.yaml\` presence to confirm bootstrap.`;
|
||||
|
||||
@@ -283,7 +284,7 @@ For bug-fix tasks, paste and fill in this checklist in the \`## Surface Enumerat
|
||||
> If keeping lint/tests/build/typecheck green requires edits outside the initial File Scope, make those fixes as part of this task.
|
||||
|
||||
- [ ] Run lint check (\`pnpm lint\`)
|
||||
- [ ] Run full test suite
|
||||
- [ ] Run impacted tests
|
||||
- [ ] Run project typecheck if available
|
||||
- [ ] Fix all failures
|
||||
- [ ] Build passes
|
||||
@@ -727,8 +728,8 @@ Call \`task_done()\` to signal completion.
|
||||
\`\`\`
|
||||
|
||||
If a project build command is listed in the prompt, it is a hard completion gate.
|
||||
Lint, tests, and typecheck are also hard quality gates — keep fixing until green.
|
||||
**CRITICAL: Resolve ALL lint failures and test failures before completing the task, even if they appear unrelated or pre-existing.** Unrelated failures left unfixed accumulate technical debt and block future integrations. Investigate and fix or suppress them — do not defer them to a separate task.`;
|
||||
Lint, tests, and typecheck are also hard quality gates for failures caused by this task.
|
||||
If unrelated or pre-existing broad-suite failures remain after impacted checks pass, log the evidence and create/link follow-up work instead of expanding the task.`;
|
||||
|
||||
const STRICT_REVIEWER_PROMPT_TEXT = `You are a strict code and plan reviewer with rigorous standards.
|
||||
|
||||
|
||||
@@ -5,7 +5,18 @@ import { PullRequestView, type PrDetail } from "../components/PullRequestView";
|
||||
// Icons → simple stubs so assertions key on text/testids, not SVG internals.
|
||||
vi.mock("lucide-react", () => {
|
||||
const Stub = () => <span />;
|
||||
return new Proxy({}, { get: () => Stub });
|
||||
return {
|
||||
AlertTriangle: Stub,
|
||||
CheckCircle: Stub,
|
||||
Clock: Stub,
|
||||
ExternalLink: Stub,
|
||||
GitMerge: Stub,
|
||||
GitPullRequest: Stub,
|
||||
MessageSquare: Stub,
|
||||
RotateCcw: Stub,
|
||||
ThumbsUp: Stub,
|
||||
XCircle: Stub,
|
||||
};
|
||||
});
|
||||
|
||||
function makeSummary(over: Partial<PrDetail["summary"]> = {}): PrDetail["summary"] {
|
||||
|
||||
@@ -6,7 +6,9 @@ import type { Task } from "@fusion/core";
|
||||
|
||||
vi.mock("lucide-react", () => {
|
||||
const Stub = () => null;
|
||||
return new Proxy({}, { get: () => Stub });
|
||||
return new Proxy({}, {
|
||||
get: (_target, prop) => prop === "then" ? undefined : Stub,
|
||||
});
|
||||
});
|
||||
|
||||
vi.mock("../ProviderIcon", () => ({
|
||||
|
||||
@@ -13,6 +13,14 @@ vi.mock("../../api", () => ({
|
||||
fetchModels: vi.fn(),
|
||||
fetchAgents: vi.fn(),
|
||||
fetchDiscoveredSkills: vi.fn(),
|
||||
fetchWorkflowStepTemplates: vi.fn().mockResolvedValue({ templates: [] }),
|
||||
fetchPluginWorkflowStepTemplates: vi.fn().mockResolvedValue({ templates: [] }),
|
||||
fetchConfig: vi.fn(),
|
||||
fetchSettings: vi.fn(),
|
||||
updateSettings: vi.fn(),
|
||||
updateGlobalSettings: vi.fn(),
|
||||
fetchWorkflowSettingValues: vi.fn().mockResolvedValue({ stored: {}, effective: {}, orphaned: [] }),
|
||||
updateWorkflowSettingValues: vi.fn().mockResolvedValue({ stored: {}, effective: {}, orphaned: [] }),
|
||||
}));
|
||||
|
||||
import {
|
||||
@@ -21,6 +29,8 @@ import {
|
||||
fetchStepParsers,
|
||||
updateWorkflow,
|
||||
fetchModels,
|
||||
fetchConfig,
|
||||
fetchSettings,
|
||||
} from "../../api";
|
||||
import type { TraitCatalogEntry } from "../../api";
|
||||
import { WorkflowNodeEditor } from "../WorkflowNodeEditor";
|
||||
@@ -68,6 +78,8 @@ describe("WorkflowNodeEditor — cli-agent executor (U15)", () => {
|
||||
vi.mocked(fetchTraits).mockResolvedValue(TRAIT_CATALOG);
|
||||
vi.mocked(fetchStepParsers).mockResolvedValue([]);
|
||||
vi.mocked(fetchModels).mockResolvedValue({ models: [] });
|
||||
vi.mocked(fetchConfig).mockResolvedValue({ maxConcurrent: 2, rootDir: "/tmp/project" });
|
||||
vi.mocked(fetchSettings).mockResolvedValue({ autoMerge: true });
|
||||
vi.mocked(updateWorkflow).mockResolvedValue(promptDef());
|
||||
// Stub the adapter-catalog fetch.
|
||||
vi.stubGlobal(
|
||||
|
||||
@@ -127,6 +127,56 @@ async function spawnWrapperTree(signal: NodeJS.Signals) {
|
||||
await waitFor(() => !isProcessAlive(pids!.childPid) && !isProcessAlive(pids!.grandchildPid));
|
||||
}
|
||||
|
||||
async function spawnWrapperTreeUntilTimeout() {
|
||||
const { pidFile, childPath, grandchildPath } = createStubProcessTree();
|
||||
const wrapper = spawn(
|
||||
process.execPath,
|
||||
[wrapperPath, "--heap=6144", "run", "--project", "dashboard-api-quality"],
|
||||
{
|
||||
cwd: dashboardRoot,
|
||||
stdio: "pipe",
|
||||
env: {
|
||||
...process.env,
|
||||
FUSION_RUN_VITEST_TIMEOUT_MS: "100",
|
||||
FUSION_RUN_VITEST_KILL_GRACE_MS: "50",
|
||||
FUSION_RUN_VITEST_SPAWN_OVERRIDE: JSON.stringify({
|
||||
command: process.execPath,
|
||||
args: [childPath, pidFile, grandchildPath],
|
||||
}),
|
||||
},
|
||||
},
|
||||
);
|
||||
activeWrappers.add(wrapper);
|
||||
|
||||
let pids: { childPid: number; grandchildPid: number } | null = null;
|
||||
await waitFor(() => {
|
||||
try {
|
||||
pids = JSON.parse(readFileSync(pidFile, "utf8")) as { childPid: number; grandchildPid: number };
|
||||
return Boolean(
|
||||
pids &&
|
||||
Number.isInteger(pids.childPid) &&
|
||||
Number.isInteger(pids.grandchildPid) &&
|
||||
isProcessAlive(pids.childPid) &&
|
||||
isProcessAlive(pids.grandchildPid),
|
||||
);
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
});
|
||||
|
||||
registerGroupLeader(pids!.childPid);
|
||||
registerPid(pids!.grandchildPid);
|
||||
|
||||
const exitCode = await new Promise<number | null>((resolve, reject) => {
|
||||
wrapper.once("error", reject);
|
||||
wrapper.once("close", (code) => resolve(code));
|
||||
});
|
||||
activeWrappers.delete(wrapper);
|
||||
|
||||
expect(exitCode).toBe(124);
|
||||
await waitFor(() => !isProcessAlive(pids!.childPid) && !isProcessAlive(pids!.grandchildPid));
|
||||
}
|
||||
|
||||
afterEach(async () => {
|
||||
for (const wrapper of activeWrappers) {
|
||||
wrapper.kill("SIGKILL");
|
||||
@@ -169,4 +219,8 @@ describe("run-vitest-with-heap", () => {
|
||||
it("reaps the spawned process group on SIGINT", async () => {
|
||||
await spawnWrapperTree("SIGINT");
|
||||
});
|
||||
|
||||
it("times out and reaps the spawned process group", async () => {
|
||||
await spawnWrapperTreeUntilTimeout();
|
||||
});
|
||||
});
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
#!/usr/bin/env node
|
||||
/* global clearInterval, console, process, setInterval */
|
||||
/* global clearInterval, clearTimeout, console, process, setInterval, setTimeout */
|
||||
|
||||
import { spawn } from "node:child_process";
|
||||
|
||||
@@ -16,6 +16,8 @@ if (vitestArgs.length === 0) {
|
||||
const nodeOptions = [`--max-old-space-size=${heapMb}`, process.env.NODE_OPTIONS || ""]
|
||||
.join(" ")
|
||||
.trim();
|
||||
const timeoutMs = Number.parseInt(process.env.FUSION_RUN_VITEST_TIMEOUT_MS || "900000", 10);
|
||||
const forceKillGraceMs = Number.parseInt(process.env.FUSION_RUN_VITEST_KILL_GRACE_MS || "5000", 10);
|
||||
|
||||
function resolveSpawnCommand() {
|
||||
const override = process.env.FUSION_RUN_VITEST_SPAWN_OVERRIDE;
|
||||
@@ -51,11 +53,31 @@ const child = spawn(command, args, {
|
||||
const heartbeat = setInterval(() => {
|
||||
console.log(`[dashboard-vitest] still running: ${vitestArgs.join(" ")}`);
|
||||
}, 5_000);
|
||||
let timeoutExitCode = null;
|
||||
let forceKillTimer = null;
|
||||
const timeout = Number.isFinite(timeoutMs) && timeoutMs > 0
|
||||
? setTimeout(() => {
|
||||
timeoutExitCode = 124;
|
||||
console.error(`[dashboard-vitest] timeout after ${timeoutMs}ms: ${vitestArgs.join(" ")}`);
|
||||
forwardSignal("SIGTERM");
|
||||
forceKillTimer = setTimeout(() => {
|
||||
forwardSignal("SIGKILL");
|
||||
}, Math.max(1, forceKillGraceMs));
|
||||
forceKillTimer.unref();
|
||||
}, timeoutMs)
|
||||
: null;
|
||||
timeout?.unref();
|
||||
|
||||
function clearHeartbeat() {
|
||||
clearInterval(heartbeat);
|
||||
}
|
||||
|
||||
function clearTimers() {
|
||||
clearHeartbeat();
|
||||
if (timeout) clearTimeout(timeout);
|
||||
if (forceKillTimer) clearTimeout(forceKillTimer);
|
||||
}
|
||||
|
||||
function forwardSignal(signal) {
|
||||
clearHeartbeat();
|
||||
|
||||
@@ -86,7 +108,7 @@ for (const signal of ["SIGINT", "SIGTERM", "SIGHUP"]) {
|
||||
}
|
||||
|
||||
process.on("exit", () => {
|
||||
clearHeartbeat();
|
||||
clearTimers();
|
||||
try {
|
||||
process.kill(-child.pid, "SIGTERM");
|
||||
} catch (error) {
|
||||
@@ -101,13 +123,16 @@ process.on("exit", () => {
|
||||
});
|
||||
|
||||
child.on("error", (error) => {
|
||||
clearHeartbeat();
|
||||
clearTimers();
|
||||
console.error(error);
|
||||
process.exit(1);
|
||||
});
|
||||
|
||||
child.on("close", (code, signal) => {
|
||||
clearHeartbeat();
|
||||
clearTimers();
|
||||
if (timeoutExitCode !== null) {
|
||||
process.exit(timeoutExitCode);
|
||||
}
|
||||
if (signal) {
|
||||
process.kill(process.pid, signal);
|
||||
return;
|
||||
|
||||
@@ -209,9 +209,7 @@ const batchedQualityAppComponentTestsA = batchedQualityAppComponentTests.slice(0
|
||||
const batchedQualityAppComponentTestsB = batchedQualityAppComponentTests.slice(batchedQualityAppSplitIndex);
|
||||
|
||||
function buildComponentQualityInclude(testNames: readonly string[]): string[] {
|
||||
return testNames.length > 0
|
||||
? [`app/components/__tests__/{${testNames.join(",")}}.test.{ts,tsx}`]
|
||||
: [];
|
||||
return testNames.map((testName) => `app/components/__tests__/${testName}.test.{ts,tsx}`);
|
||||
}
|
||||
|
||||
const qualityAppTests = [
|
||||
|
||||
@@ -140,6 +140,50 @@ describe("AgentSemaphore", () => {
|
||||
expect(sem.availableCount).toBe(3);
|
||||
});
|
||||
|
||||
it("reports waitingCount and diagnostic snapshot", async () => {
|
||||
const sem = new AgentSemaphore(1);
|
||||
await sem.acquire();
|
||||
|
||||
const waiter = sem.acquire();
|
||||
await Promise.resolve();
|
||||
|
||||
expect(sem.waitingCount).toBe(1);
|
||||
expect(sem.snapshot()).toEqual({
|
||||
activeCount: 1,
|
||||
waitingCount: 1,
|
||||
availableCount: 0,
|
||||
limit: 1,
|
||||
});
|
||||
|
||||
sem.release();
|
||||
await waiter;
|
||||
expect(sem.waitingCount).toBe(0);
|
||||
sem.release();
|
||||
});
|
||||
|
||||
it("reconciles stale active counts down to persisted active work", async () => {
|
||||
const sem = new AgentSemaphore(2);
|
||||
await sem.acquire();
|
||||
await sem.acquire();
|
||||
|
||||
const result = sem.reconcileActiveCount(0);
|
||||
|
||||
expect(result).toEqual({ before: 2, after: 0, changed: true });
|
||||
expect(sem.activeCount).toBe(0);
|
||||
expect(sem.availableCount).toBe(2);
|
||||
});
|
||||
|
||||
it("does not increase active counts during reconciliation", async () => {
|
||||
const sem = new AgentSemaphore(2);
|
||||
await sem.acquire();
|
||||
|
||||
const result = sem.reconcileActiveCount(3);
|
||||
|
||||
expect(result).toEqual({ before: 1, after: 1, changed: false });
|
||||
expect(sem.activeCount).toBe(1);
|
||||
sem.release();
|
||||
});
|
||||
|
||||
it("run() gates concurrent calls", async () => {
|
||||
const sem = new AgentSemaphore(2);
|
||||
let concurrent = 0;
|
||||
|
||||
@@ -207,31 +207,30 @@ describe("buildExecutionPrompt", () => {
|
||||
expect(result).toContain("- **Build:** `pnpm build`");
|
||||
});
|
||||
|
||||
it("tells executors to fix quality-gate failures even outside initial file scope", () => {
|
||||
it("tells executors to split unrelated broad-suite failures into follow-up work", () => {
|
||||
const task = createMockTaskDetail();
|
||||
const result = buildExecutionPrompt(task, "/home/user/project", {
|
||||
testCommand: "pnpm test",
|
||||
buildCommand: "pnpm build",
|
||||
} as any);
|
||||
|
||||
expect(result).toContain("fix failures even when that requires edits outside the original File Scope");
|
||||
expect(result).toContain("caused-by-this-task failures are blocking");
|
||||
expect(result).toContain("unrelated or pre-existing failures should be logged and split into a follow-up");
|
||||
expect(result).toContain("If the repo has a typecheck command, run it before `fn_task_done()`");
|
||||
expect(result).toContain("not for fixes required to get tests, build, or typecheck back to green");
|
||||
expect(result).toContain("including unrelated/pre-existing broad-suite failures");
|
||||
});
|
||||
|
||||
it("requires resolving ALL test failures, including unrelated or pre-existing ones", () => {
|
||||
it("warns against repeated broad workspace verification loops", () => {
|
||||
const task = createMockTaskDetail();
|
||||
const result = buildExecutionPrompt(task, "/home/user/project", {
|
||||
testCommand: "pnpm test",
|
||||
buildCommand: "pnpm build",
|
||||
} as any);
|
||||
|
||||
// The stricter language must be present to prevent "unrelated failure" deferrals
|
||||
expect(result).toContain("Resolve ALL test failures");
|
||||
expect(result).toContain("even if they appear unrelated or pre-existing");
|
||||
expect(result).toContain("accumulate technical debt");
|
||||
expect(result).toContain("Investigate and fix or suppress them");
|
||||
expect(result).toContain("do not defer them to a separate task");
|
||||
expect(result).toContain("Do not repeatedly rerun a broad failing or hanging workspace command");
|
||||
expect(result).toContain("without a new hypothesis and a narrower confirming command");
|
||||
expect(result).toContain("unrelated or pre-existing failures should be logged and split into a follow-up");
|
||||
expect(result).not.toContain("Resolve ALL test failures");
|
||||
});
|
||||
|
||||
it("includes source issue reference in commit instruction when task has github sourceIssue", () => {
|
||||
@@ -2571,4 +2570,3 @@ describe("fn_task_update bare-call guard (P1 api-contract)", () => {
|
||||
expect(text).not.toContain("fn_task_update requires at least one of");
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
@@ -465,7 +465,12 @@ describe("TaskExecutor bounded recovery retries", () => {
|
||||
);
|
||||
expect(store.moveTask).not.toHaveBeenCalledWith("FN-001", "in-review");
|
||||
// Executor now handles the requeue in its finally block
|
||||
expect(store.updateTask).toHaveBeenCalledWith("FN-001", { status: "stuck-killed", worktree: null, branch: null });
|
||||
expect(store.updateTask).toHaveBeenCalledWith("FN-001", {
|
||||
status: "queued",
|
||||
error: null,
|
||||
worktree: null,
|
||||
branch: null,
|
||||
});
|
||||
expect(store.moveTask).toHaveBeenCalledWith("FN-001", "todo", { preserveProgress: true });
|
||||
});
|
||||
|
||||
@@ -499,7 +504,11 @@ describe("TaskExecutor bounded recovery retries", () => {
|
||||
|
||||
// Should NOT requeue or mark as failed (budget handler already did that)
|
||||
expect(store.moveTask).not.toHaveBeenCalledWith("FN-001", "todo");
|
||||
expect(store.updateTask).not.toHaveBeenCalledWith("FN-001", { status: "stuck-killed", worktree: null, branch: null });
|
||||
expect(store.updateTask).not.toHaveBeenCalledWith("FN-001", expect.objectContaining({
|
||||
status: "queued",
|
||||
worktree: null,
|
||||
branch: null,
|
||||
}));
|
||||
expect(store.updateTask).not.toHaveBeenCalledWith(
|
||||
"FN-001",
|
||||
expect.objectContaining({ status: "failed" }),
|
||||
@@ -556,6 +565,48 @@ describe("TaskExecutor bounded recovery retries", () => {
|
||||
);
|
||||
});
|
||||
|
||||
it("does not let a late graph failure clobber a retryable requeue", async () => {
|
||||
const store = createMockStore();
|
||||
const task = {
|
||||
id: "FN-001",
|
||||
title: "Test",
|
||||
description: "Test",
|
||||
column: "in-progress",
|
||||
status: undefined,
|
||||
dependencies: [],
|
||||
steps: [],
|
||||
currentStep: 0,
|
||||
log: [],
|
||||
createdAt: new Date().toISOString(),
|
||||
updatedAt: new Date().toISOString(),
|
||||
} as Task;
|
||||
store.getTask.mockResolvedValue({
|
||||
...task,
|
||||
column: "todo",
|
||||
status: "queued",
|
||||
error: null,
|
||||
});
|
||||
const executor = new TaskExecutor(store, "/tmp/test", {});
|
||||
|
||||
await (executor as any).handleGraphFailure(task, {
|
||||
visitedNodeIds: ["execute"],
|
||||
});
|
||||
|
||||
expect(store.updateTask).not.toHaveBeenCalledWith(
|
||||
"FN-001",
|
||||
expect.objectContaining({ status: "failed" }),
|
||||
expect.anything(),
|
||||
);
|
||||
expect(store.moveTask).not.toHaveBeenCalledWith("FN-001", "in-review", expect.anything());
|
||||
expect(store.handoffToReview).not.toHaveBeenCalled();
|
||||
expect(store.logEntry).toHaveBeenCalledWith(
|
||||
"FN-001",
|
||||
"Workflow graph terminated with failure at node 'execute' (task already todo - preserving recovered lifecycle state)",
|
||||
undefined,
|
||||
undefined,
|
||||
);
|
||||
});
|
||||
|
||||
it("preserves step progress when requeuing stuck task by default", async () => {
|
||||
const store = createMockStore();
|
||||
const executor = new TaskExecutor(store, "/tmp/test", {});
|
||||
@@ -1465,8 +1516,8 @@ describe("Invalid transition error handling", () => {
|
||||
// then throws the Invalid transition error,
|
||||
// which is caught by the outer handler.
|
||||
expect(store.updateTask).toHaveBeenCalledWith("FN-001", {
|
||||
status: "failed",
|
||||
error: "Agent finished without calling fn_task_done (after 3 retries)",
|
||||
status: "queued",
|
||||
error: null,
|
||||
taskDoneRetryCount: 1,
|
||||
});
|
||||
|
||||
|
||||
@@ -116,8 +116,8 @@ describe("Workflow Steps Execution", () => {
|
||||
|
||||
// Retries still didn't call fn_task_done, so it fails and requeues immediately.
|
||||
expect(store.updateTask).toHaveBeenCalledWith("FN-001", {
|
||||
status: "failed",
|
||||
error: "Agent finished without calling fn_task_done (after 3 retries)",
|
||||
status: "queued",
|
||||
error: null,
|
||||
taskDoneRetryCount: 1,
|
||||
});
|
||||
expect(store.moveTask).toHaveBeenCalledWith("FN-001", "todo", { preserveProgress: true });
|
||||
@@ -242,8 +242,9 @@ describe("Workflow Steps Execution", () => {
|
||||
|
||||
expect(mockedCreateFnAgent).toHaveBeenCalledTimes(4);
|
||||
expect(store.updateTask).toHaveBeenCalledWith("FN-5436-A", expect.objectContaining({
|
||||
status: "failed",
|
||||
error: "Agent finished without calling fn_task_done (after 3 retries)",
|
||||
status: "queued",
|
||||
error: null,
|
||||
taskDoneRetryCount: 1,
|
||||
}));
|
||||
expect(store.moveTask).toHaveBeenCalledWith("FN-5436-A", "todo", { preserveProgress: true });
|
||||
expect(store.moveTask).not.toHaveBeenCalledWith("FN-5436-A", "in-review");
|
||||
@@ -335,8 +336,8 @@ describe("Workflow Steps Execution", () => {
|
||||
|
||||
expect(mockedCreateFnAgent).toHaveBeenCalledTimes(4);
|
||||
expect(store.updateTask).toHaveBeenCalledWith("FN-5436-C", {
|
||||
status: "failed",
|
||||
error: "Agent finished without calling fn_task_done (after 3 retries)",
|
||||
status: "queued",
|
||||
error: null,
|
||||
taskDoneRetryCount: 1,
|
||||
});
|
||||
});
|
||||
@@ -1224,7 +1225,7 @@ describe("Workflow Steps Execution", () => {
|
||||
store.getSettings.mockResolvedValue({
|
||||
maxConcurrent: 2,
|
||||
maxWorktrees: 4,
|
||||
scripts: { test: "echo 'all tests passed'" },
|
||||
scripts: { test: `node -e "if (process.env.FN3968_SCRIPT_ENV !== 'workflow-script-env') process.exit(42)"` },
|
||||
});
|
||||
|
||||
store.getTask.mockResolvedValue({
|
||||
@@ -1254,14 +1255,6 @@ describe("Workflow Steps Execution", () => {
|
||||
updatedAt: new Date().toISOString(),
|
||||
});
|
||||
|
||||
// Mock execSync to succeed for the script command
|
||||
mockedExecSync.mockImplementation((cmd: string | string[]) => {
|
||||
if (typeof cmd === "string" && cmd.includes("echo")) {
|
||||
return Buffer.from("all tests passed\n");
|
||||
}
|
||||
return Buffer.from("");
|
||||
});
|
||||
|
||||
// Main agent with fn_task_done
|
||||
createAgentWithTaskDone();
|
||||
|
||||
@@ -1308,13 +1301,6 @@ describe("Workflow Steps Execution", () => {
|
||||
]),
|
||||
}),
|
||||
);
|
||||
const updatePayloads = store.updateTask.mock.calls.map((call: any[]) => call[1]);
|
||||
expect(JSON.stringify(updatePayloads)).not.toContain("all tests passed");
|
||||
|
||||
const scriptExecCall = mockedExecSync.mock.calls.find(
|
||||
(call: any[]) => typeof call[0] === "string" && call[0].includes("echo 'all tests passed'")
|
||||
);
|
||||
expect(scriptExecCall?.[1]?.env?.FN3968_SCRIPT_ENV).toBe("workflow-script-env");
|
||||
delete process.env.FN3968_SCRIPT_ENV;
|
||||
});
|
||||
|
||||
|
||||
@@ -1527,6 +1527,34 @@ describe("Scheduler", () => {
|
||||
expect(String(call?.[1])).toContain("semaphore slots may include triage/merge agents outside in-progress");
|
||||
});
|
||||
|
||||
it("recovers an idle leaked semaphore slot before dispatching", async () => {
|
||||
vi.mocked(existsSync).mockReturnValue(true);
|
||||
vi.mocked(readFile).mockResolvedValue("# Task\nDo something");
|
||||
|
||||
const semaphore = new AgentSemaphore(1);
|
||||
await semaphore.acquire();
|
||||
const task = createMockTask({ id: "FN-A", column: "todo", dependencies: [] });
|
||||
const store = createMockStore({
|
||||
listTasks: vi.fn().mockResolvedValue([task]),
|
||||
getSettings: vi.fn().mockResolvedValue({ maxConcurrent: 10, maxWorktrees: 10 }),
|
||||
});
|
||||
|
||||
const scheduler = new Scheduler(store, { semaphore });
|
||||
(scheduler as any).running = true;
|
||||
(scheduler as any).idleSemaphoreLeakCandidateSince = Date.now() - 6_000;
|
||||
await scheduler.schedule();
|
||||
|
||||
expect(semaphore.activeCount).toBe(0);
|
||||
expect(schedulerLog.warn).toHaveBeenCalledWith(
|
||||
expect.stringContaining("scheduler: recovered stale semaphore active count 1 -> 0"),
|
||||
);
|
||||
expect(store.moveTask).toHaveBeenCalledWith(
|
||||
"FN-A",
|
||||
"in-progress",
|
||||
expect.objectContaining({ moveSource: "scheduler" }),
|
||||
);
|
||||
});
|
||||
|
||||
it("lists tied binding gates in stable order", async () => {
|
||||
vi.mocked(existsSync).mockReturnValue(true);
|
||||
vi.mocked(readFile).mockResolvedValue("# Task\nDo something");
|
||||
|
||||
@@ -66,6 +66,38 @@ export class AgentSemaphore {
|
||||
return this._active;
|
||||
}
|
||||
|
||||
/** Number of callers currently queued for a semaphore slot. */
|
||||
get waitingCount(): number {
|
||||
return this._waiters.length;
|
||||
}
|
||||
|
||||
/** Snapshot of current semaphore pressure for diagnostics. */
|
||||
snapshot(): { activeCount: number; waitingCount: number; availableCount: number; limit: number } {
|
||||
return {
|
||||
activeCount: this.activeCount,
|
||||
waitingCount: this.waitingCount,
|
||||
availableCount: this.availableCount,
|
||||
limit: this.limit,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Clamp stale active-slot accounting to a persisted upper bound.
|
||||
*
|
||||
* This is a recovery valve for crash/abort paths where the task/session that
|
||||
* acquired a slot is gone but the in-memory semaphore did not observe its
|
||||
* normal `finally` release. The caller owns the persisted-state judgment.
|
||||
*/
|
||||
reconcileActiveCount(maxActive: number): { before: number; after: number; changed: boolean } {
|
||||
const bounded = Math.max(0, Math.floor(maxActive));
|
||||
const before = this._active;
|
||||
if (before > bounded) {
|
||||
this._active = bounded;
|
||||
this._drain();
|
||||
}
|
||||
return { before, after: this._active, changed: before !== this._active };
|
||||
}
|
||||
|
||||
/** Number of slots available for immediate acquisition. May be 0 or negative
|
||||
* if the limit was reduced below the current active count.
|
||||
* Returns 0 when the limit is not a valid positive number (defensive guard). */
|
||||
|
||||
@@ -1074,12 +1074,12 @@ If a project build command is listed in the prompt, it is a hard completion gate
|
||||
- If the build fails, do NOT call \`fn_task_done()\`; keep working until it passes
|
||||
|
||||
Lint, tests, and typecheck are also hard quality gates:
|
||||
- Keep fixing failures until lint, the configured/full test suite, and typecheck all pass
|
||||
- If the repository exposes a typecheck command, run it and keep fixing failures until it passes
|
||||
- Do not stop at "out of scope" if additional fixes are required to restore green lint, tests, build, or typecheck
|
||||
- When tests fail, first identify whether the failure is caused by your change, a pre-existing defect, or an outdated test expectation; then fix code or tests accordingly so behavior and assertions match
|
||||
- Update tests when intended behavior changed; fix implementation when behavior regressed unintentionally
|
||||
- **CRITICAL: Resolve ALL lint failures and test failures before completing the task, even if they appear unrelated or pre-existing.** Unrelated failures left unfixed accumulate technical debt and block future integrations. Investigate and fix or suppress them — do not defer them to a separate task.
|
||||
- Keep fixing failures caused by your change until lint, targeted tests, build, and typecheck pass.
|
||||
- If the repository exposes a typecheck command, run it and fix failures caused by your change.
|
||||
- When tests fail, first identify whether the failure is caused by your change, a pre-existing defect, an unrelated flaky test, or an outdated test expectation.
|
||||
- Update tests when intended behavior changed; fix implementation when behavior regressed unintentionally.
|
||||
- If broad workspace verification fails on unrelated or pre-existing failures after targeted checks pass, do NOT expand this task by fixing unrelated areas. Log the evidence, quarantine flakes per project policy, or create/link a follow-up task.
|
||||
- Do not repeatedly rerun a broad failing or hanging workspace command without a new hypothesis and a narrower confirming command.
|
||||
|
||||
## Verification commands — use fn_run_verification
|
||||
|
||||
@@ -1088,7 +1088,7 @@ The tool prevents your session from being killed by the inactivity watchdog duri
|
||||
|
||||
- Prefer **package-scoped** verification first: e.g. \`pnpm --filter @fusion/<pkg> test\` with \`scope: "package"\`. This is faster and isolated.
|
||||
- For file-specific package tests, use direct Vitest execution with package-relative paths: \`pnpm --filter @fusion/<pkg> exec vitest run src/path/to/test.ts --silent=passed-only --reporter=dot\`. Do not use \`pnpm --filter @fusion/<pkg> test -- --run <files>\`; package test scripts can expand into broad quality suites before the filter is applied.
|
||||
- Only run **workspace-scoped** verification (\`pnpm test\`, \`pnpm lint\`, \`pnpm build\` from root) at the FINAL integration step, when you are about to call \`fn_task_done\`.
|
||||
- Run **workspace-scoped** verification (\`pnpm test\`, \`pnpm lint\`, \`pnpm build\` from root) only when it is explicitly required by the task/workflow or after impacted/package-scoped checks pass and you are doing final integration.
|
||||
- If you need to run \`pnpm install\` (e.g. you added a new package), use \`fn_run_verification\` with \`scope: "workspace"\` and \`timeoutSec: 600\`.
|
||||
- If a verification command times out, do NOT blindly retry — investigate. Check for hung subprocesses, infinite test loops, or tests waiting on missing dependencies. Use \`node_modules/.modules.yaml\` presence to confirm bootstrap.
|
||||
|
||||
@@ -5549,15 +5549,25 @@ export class TaskExecutor {
|
||||
await this.store.logEntry(task.id, `${message} (task paused — not parked)`, undefined, this.getRunContextFor(task.id));
|
||||
return;
|
||||
}
|
||||
if (live.column !== "in-progress") {
|
||||
executorLog.log(
|
||||
`${task.id}: graph run ended after task moved to '${live.column}' - preserving recovered lifecycle state`,
|
||||
);
|
||||
await this.store.logEntry(
|
||||
task.id,
|
||||
`${message} (task already ${live.column} - preserving recovered lifecycle state)`,
|
||||
undefined,
|
||||
this.getRunContextFor(task.id),
|
||||
);
|
||||
return;
|
||||
}
|
||||
await this.store.logEntry(task.id, message, undefined, this.getRunContextFor(task.id));
|
||||
// status "failed" doubles as the self-healing exemption: review-task
|
||||
// revival sweeps skip tasks carrying a non-null status, preventing the
|
||||
// FN-5704-style loop of re-running the graph from scratch.
|
||||
await this.store.updateTask(task.id, { error: message, status: "failed" }, this.getRunContextFor(task.id));
|
||||
if (live.column === "in-progress") {
|
||||
await this.persistTokenUsage(task.id);
|
||||
await this.handoffTaskToReview(live, "workflow-graph-failed");
|
||||
}
|
||||
await this.persistTokenUsage(task.id);
|
||||
await this.handoffTaskToReview(live, "workflow-graph-failed");
|
||||
} catch (err) {
|
||||
executorLog.error(
|
||||
`${task.id}: failed to park graph-failed task: ${err instanceof Error ? err.message : String(err)}`,
|
||||
@@ -6089,8 +6099,8 @@ export class TaskExecutor {
|
||||
|
||||
if (priorRequeues < MAX_TASK_DONE_REQUEUE_RETRIES) {
|
||||
await this.store.updateTask(task.id, {
|
||||
status: "failed",
|
||||
error: failureMessage,
|
||||
status: "queued",
|
||||
error: null,
|
||||
worktree: null,
|
||||
branch: null,
|
||||
sessionFile: null,
|
||||
@@ -6669,7 +6679,12 @@ export class TaskExecutor {
|
||||
executorLog.warn(`${task.id}: worktree removal failed during stuck-requeue cleanup (${worktreePath}): ${msg}`);
|
||||
}
|
||||
}
|
||||
await this.store.updateTask(task.id, { status: "stuck-killed", worktree: null, branch: null });
|
||||
await this.store.updateTask(task.id, {
|
||||
status: "queued",
|
||||
error: null,
|
||||
worktree: null,
|
||||
branch: null,
|
||||
});
|
||||
if (latestTask.column !== "todo") {
|
||||
await this.store.moveTask(task.id, "todo", preserveProgress ? { preserveProgress: true } : undefined);
|
||||
executorLog.log(`${task.id} moved to todo for retry after stuck kill${preserveProgress ? " (progress preserved)" : ""}`);
|
||||
@@ -7576,8 +7591,8 @@ export class TaskExecutor {
|
||||
|
||||
if (priorRequeues < MAX_TASK_DONE_REQUEUE_RETRIES) {
|
||||
await this.store.updateTask(task.id, {
|
||||
status: "failed",
|
||||
error: errorMessage,
|
||||
status: "queued",
|
||||
error: null,
|
||||
taskDoneRetryCount: nextRequeueCount,
|
||||
});
|
||||
await this.store.logEntry(
|
||||
@@ -8338,7 +8353,12 @@ export class TaskExecutor {
|
||||
executorLog.warn(`Failed to remove old worktree ${worktreePath}: ${cleanupErrMessage}`);
|
||||
}
|
||||
}
|
||||
await this.store.updateTask(task.id, { status: "stuck-killed", worktree: null, branch: null });
|
||||
await this.store.updateTask(task.id, {
|
||||
status: "queued",
|
||||
error: null,
|
||||
worktree: null,
|
||||
branch: null,
|
||||
});
|
||||
// Only move to todo if not already there. Use the freshly-read
|
||||
// latestTask.column rather than the stale captured task.column —
|
||||
// the captured snapshot can be hours old and would race against
|
||||
@@ -9095,8 +9115,8 @@ export class TaskExecutor {
|
||||
const nextRequeueCount = priorRequeues + 1;
|
||||
if (priorRequeues < MAX_TASK_DONE_REQUEUE_RETRIES) {
|
||||
await this.store.updateTask(task.id, {
|
||||
status: "failed",
|
||||
error: refusal.message,
|
||||
status: "queued",
|
||||
error: null,
|
||||
taskDoneRetryCount: nextRequeueCount,
|
||||
paused: false,
|
||||
pausedByAgentId: null,
|
||||
@@ -9190,8 +9210,8 @@ export class TaskExecutor {
|
||||
const nextRequeueCount = priorRequeues + 1;
|
||||
if (priorRequeues < MAX_TASK_DONE_REQUEUE_RETRIES) {
|
||||
await store.updateTask(taskId, {
|
||||
status: "failed",
|
||||
error: refusalMessage,
|
||||
status: "queued",
|
||||
error: null,
|
||||
taskDoneRetryCount: nextRequeueCount,
|
||||
paused: false,
|
||||
pausedByAgentId: null,
|
||||
@@ -9248,8 +9268,8 @@ export class TaskExecutor {
|
||||
const nextRequeueCount = priorRequeues + 1;
|
||||
if (priorRequeues < MAX_TASK_DONE_REQUEUE_RETRIES) {
|
||||
await store.updateTask(taskId, {
|
||||
status: "failed",
|
||||
error: refusalMessage,
|
||||
status: "queued",
|
||||
error: null,
|
||||
taskDoneRetryCount: nextRequeueCount,
|
||||
paused: false,
|
||||
pausedByAgentId: null,
|
||||
@@ -13364,7 +13384,12 @@ Backward compat fallback: if JSON is unavailable, you may still begin output wit
|
||||
taskId,
|
||||
`Force-requeued after stuck-kill: executor did not unwind within ${FORCE_REQUEUE_GRACE_MS / 1000}s (hung subprocess)${preserveProgress ? " — progress preserved" : ""}`,
|
||||
);
|
||||
await this.store.updateTask(taskId, { status: "stuck-killed", worktree: null, branch: null });
|
||||
await this.store.updateTask(taskId, {
|
||||
status: "queued",
|
||||
error: null,
|
||||
worktree: null,
|
||||
branch: null,
|
||||
});
|
||||
await this.store.moveTask(taskId, "todo", preserveProgress ? { preserveProgress: true } : undefined);
|
||||
// Remove from executing so the scheduler can re-dispatch normally.
|
||||
// The old Promise is still running but the executing guard is cleared so
|
||||
@@ -13974,19 +13999,19 @@ ${hasProgress
|
||||
: "Start with Step 0 (Preflight). Work through each step in order."}
|
||||
Use \`fn_task_update\` to report progress on every step transition.
|
||||
Use \`fn_task_log\` for important actions and decisions.
|
||||
Use \`fn_task_create\` for truly separate follow-up work, not for fixes required to get tests, build, or typecheck back to green.
|
||||
Use \`fn_task_create\` for truly separate follow-up work, including unrelated/pre-existing broad-suite failures.
|
||||
Commit at step boundaries: \`git commit -m "feat(${task.id}): complete Step N — <short summary>"${sourceIssueRef ? ` -m "Ref: ${sourceIssueRef}"` : ""}${authorArg}\`
|
||||
The \`<short summary>\` is required — replace it with a concrete 5–10 word description of what the step changed.
|
||||
When all steps are complete: call \`fn_task_done()\`
|
||||
|
||||
If a build command is configured, run that exact command in this worktree before calling \`fn_task_done()\`.
|
||||
Treat a non-zero exit code as a blocking failure. Do not claim success without a real passing run.
|
||||
Run the configured/full test suite and fix failures even when that requires edits outside the original File Scope.
|
||||
Run impacted/package-scoped tests before completion. Run the configured workspace test command only when the task/workflow explicitly requires it or after impacted checks pass for final integration. If any broad command fails, classify the failure before editing: caused-by-this-task failures are blocking; unrelated or pre-existing failures should be logged and split into a follow-up instead of expanding this task.
|
||||
If the repo has a lint command (e.g. \`pnpm lint\`, \`npm run lint\`), run it before \`fn_task_done()\` and fix any failures it reports.
|
||||
If the repo has a typecheck command, run it before \`fn_task_done()\` and fix any failures it reports.
|
||||
Use \`fn_task_create\` for truly separate follow-up work, not for fixes required to get tests, build, or typecheck back to green.
|
||||
Use \`fn_task_create\` for truly separate follow-up work, including unrelated/pre-existing broad-suite failures.
|
||||
If lint is configured and failing, fix that too before completion.
|
||||
**CRITICAL: Resolve ALL test failures (and any lint/typecheck failures) before completing the task, even if they appear unrelated or pre-existing.** Unrelated failures left unfixed accumulate technical debt and block future integrations. Investigate and fix or suppress them — do not defer them to a separate task.`;
|
||||
Do not repeatedly rerun a broad failing or hanging workspace command without a new hypothesis and a narrower confirming command.`;
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -18,7 +18,7 @@
|
||||
* without a full agent session.
|
||||
*/
|
||||
|
||||
import { spawn } from "node:child_process";
|
||||
import { superviseSpawn, type SupervisedChild } from "@fusion/core";
|
||||
import { existsSync, readdirSync, readFileSync } from "node:fs";
|
||||
import { isAbsolute, join, relative } from "node:path";
|
||||
import { Type, type Static } from "@earendil-works/pi-ai";
|
||||
@@ -218,20 +218,8 @@ export function normalizeVerificationCommand(command: string, rootDir: string):
|
||||
return { command: normalizedCommand, warnings };
|
||||
}
|
||||
|
||||
function killVerificationProcess(child: ReturnType<typeof spawn>, signal: NodeJS.Signals): void {
|
||||
if (process.platform !== "win32" && child.pid) {
|
||||
try {
|
||||
process.kill(-child.pid, signal);
|
||||
return;
|
||||
} catch {
|
||||
// Fall back to killing the immediate child below.
|
||||
}
|
||||
}
|
||||
try {
|
||||
child.kill(signal);
|
||||
} catch {
|
||||
// The process may already have exited.
|
||||
}
|
||||
function killVerificationProcess(supervised: SupervisedChild, signal: NodeJS.Signals): void {
|
||||
supervised.kill(signal);
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
@@ -376,10 +364,7 @@ export async function runVerificationCommand(
|
||||
const stderrBuf = createBuffer();
|
||||
|
||||
return new Promise<VerificationResult>((resolve) => {
|
||||
// Use shell: true so Node picks the platform default — /bin/sh on POSIX,
|
||||
// cmd.exe on Windows. SIGTERM/SIGKILL semantics still apply on POSIX;
|
||||
// on Windows the kill signals map to TerminateProcess.
|
||||
const child = spawn(command, {
|
||||
const supervised = superviseSpawn(command, [], {
|
||||
cwd,
|
||||
stdio: ["ignore", "pipe", "pipe"],
|
||||
env: {
|
||||
@@ -390,8 +375,10 @@ export async function runVerificationCommand(
|
||||
COREPACK_ENABLE_DOWNLOAD_PROMPT: "0",
|
||||
},
|
||||
shell: true,
|
||||
detached: process.platform !== "win32",
|
||||
killGraceMs: SIGKILL_GRACE_MS,
|
||||
maxLifetimeMs: timeoutMs + SIGKILL_GRACE_MS + 1_000,
|
||||
});
|
||||
const child = supervised.child;
|
||||
|
||||
let timedOut = false;
|
||||
let killed = false;
|
||||
@@ -417,14 +404,14 @@ export async function runVerificationCommand(
|
||||
executorLog.warn(
|
||||
`[fn_run_verification] hard timeout (${timeoutMs / 1000}s) — sending SIGTERM to: ${command}`,
|
||||
);
|
||||
killVerificationProcess(child, "SIGTERM");
|
||||
killVerificationProcess(supervised, "SIGTERM");
|
||||
|
||||
killTimer = setTimeout(() => {
|
||||
if (!settled) {
|
||||
executorLog.warn(
|
||||
`[fn_run_verification] SIGTERM ignored — sending SIGKILL to: ${command}`,
|
||||
);
|
||||
killVerificationProcess(child, "SIGKILL");
|
||||
killVerificationProcess(supervised, "SIGKILL");
|
||||
killed = true;
|
||||
}
|
||||
}, SIGKILL_GRACE_MS);
|
||||
@@ -432,7 +419,7 @@ export async function runVerificationCommand(
|
||||
|
||||
// ── stdout ───────────────────────────────────────────────────────────────
|
||||
let stdoutRemainder = "";
|
||||
child.stdout.on("data", (chunk: Buffer) => {
|
||||
child.stdout?.on("data", (chunk: Buffer) => {
|
||||
const text = stdoutRemainder + chunk.toString("utf8");
|
||||
const lines = text.split("\n");
|
||||
stdoutRemainder = lines.pop() ?? "";
|
||||
@@ -447,7 +434,7 @@ export async function runVerificationCommand(
|
||||
|
||||
// ── stderr ───────────────────────────────────────────────────────────────
|
||||
let stderrRemainder = "";
|
||||
child.stderr.on("data", (chunk: Buffer) => {
|
||||
child.stderr?.on("data", (chunk: Buffer) => {
|
||||
const text = stderrRemainder + chunk.toString("utf8");
|
||||
const lines = text.split("\n");
|
||||
stderrRemainder = lines.pop() ?? "";
|
||||
|
||||
@@ -1,10 +1,24 @@
|
||||
import { access, mkdtemp, rm, writeFile } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { cwd } from "node:process";
|
||||
import { setTimeout as delay } from "node:timers/promises";
|
||||
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { afterEach, beforeEach, describe, expect, it } from "vitest";
|
||||
|
||||
import { NativeSandboxBackend } from "../native.js";
|
||||
|
||||
describe("NativeSandboxBackend", () => {
|
||||
let tempDir: string;
|
||||
|
||||
beforeEach(async () => {
|
||||
tempDir = await mkdtemp(join(tmpdir(), "fusion-native-sandbox-"));
|
||||
});
|
||||
|
||||
afterEach(async () => {
|
||||
await rm(tempDir, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
it("returns stdout on success", async () => {
|
||||
const backend = new NativeSandboxBackend();
|
||||
const result = await backend.run("node -e 'process.stdout.write(\"ok\")'", {
|
||||
@@ -34,6 +48,67 @@ describe("NativeSandboxBackend", () => {
|
||||
expect(result.signal).toBe("SIGTERM");
|
||||
});
|
||||
|
||||
it.skipIf(process.platform === "win32")("times out and terminates descendant processes in the command process group", async () => {
|
||||
const backend = new NativeSandboxBackend();
|
||||
const markerPath = join(tempDir, "descendant-survived.txt");
|
||||
const parentScriptPath = join(tempDir, "spawn-descendant.cjs");
|
||||
await writeFile(
|
||||
parentScriptPath,
|
||||
`
|
||||
const { spawn } = require("node:child_process");
|
||||
spawn(process.execPath, [
|
||||
"-e",
|
||||
"setTimeout(() => require('node:fs').writeFileSync(process.env.MARKER, 'survived'), 450)",
|
||||
], {
|
||||
env: { ...process.env, MARKER: process.argv[2] },
|
||||
stdio: "ignore",
|
||||
}).unref();
|
||||
setInterval(() => {}, 1000);
|
||||
`,
|
||||
"utf-8",
|
||||
);
|
||||
|
||||
const result = await backend.run(
|
||||
`${JSON.stringify(process.execPath)} ${JSON.stringify(parentScriptPath)} ${JSON.stringify(markerPath)}`,
|
||||
{
|
||||
cwd: tempDir,
|
||||
timeoutMs: 75,
|
||||
maxBuffer: 1024 * 1024,
|
||||
encoding: "utf-8",
|
||||
},
|
||||
);
|
||||
|
||||
expect(result.timedOut).toBe(true);
|
||||
await delay(700);
|
||||
await expect(access(markerPath)).rejects.toThrow();
|
||||
});
|
||||
|
||||
it.skipIf(process.platform === "win32")("cleans up background children after successful commands", async () => {
|
||||
const backend = new NativeSandboxBackend();
|
||||
const markerPath = join(tempDir, "success-descendant-survived.txt");
|
||||
const parentScript = [
|
||||
"const { spawn } = require('node:child_process');",
|
||||
`spawn(process.execPath, ['-e', ${JSON.stringify("setTimeout(() => require('node:fs').writeFileSync(process.env.MARKER, 'survived'), 450)")}], { env: { ...process.env, MARKER: process.env.MARKER }, stdio: 'ignore' }).unref();`,
|
||||
"process.stdout.write('parent-done');",
|
||||
].join(" ");
|
||||
|
||||
const result = await backend.run(
|
||||
`${JSON.stringify(process.execPath)} -e ${JSON.stringify(parentScript)}`,
|
||||
{
|
||||
cwd: tempDir,
|
||||
timeoutMs: 5_000,
|
||||
maxBuffer: 1024 * 1024,
|
||||
encoding: "utf-8",
|
||||
env: { ...process.env, MARKER: markerPath },
|
||||
},
|
||||
);
|
||||
|
||||
expect(result.exitCode).toBe(0);
|
||||
expect(result.stdout).toBe("parent-done");
|
||||
await delay(700);
|
||||
await expect(access(markerPath)).rejects.toThrow();
|
||||
});
|
||||
|
||||
it("maps non-zero exits", async () => {
|
||||
const backend = new NativeSandboxBackend();
|
||||
const result = await backend.run("node -e 'process.stderr.write(\"fail\"); process.exit(7)'", {
|
||||
|
||||
@@ -1,5 +1,3 @@
|
||||
import { exec } from "node:child_process";
|
||||
import { promisify } from "node:util";
|
||||
import { superviseSpawn } from "@fusion/core";
|
||||
|
||||
import type {
|
||||
@@ -12,7 +10,8 @@ import type {
|
||||
SandboxStreamingResult,
|
||||
} from "./types.js";
|
||||
|
||||
const execAsync = promisify(exec);
|
||||
const FORCE_KILL_DELAY_MS = 5_000;
|
||||
const NORMAL_CLEANUP_FORCE_KILL_DELAY_MS = 500;
|
||||
|
||||
export class NativeSandboxBackend implements SandboxBackend {
|
||||
capabilities(): SandboxCapabilities {
|
||||
@@ -30,48 +29,110 @@ export class NativeSandboxBackend implements SandboxBackend {
|
||||
}
|
||||
|
||||
async run(command: string, options: SandboxRunOptions): Promise<SandboxRunResult> {
|
||||
try {
|
||||
const execOptions: Parameters<typeof exec>[1] = {
|
||||
cwd: options.cwd,
|
||||
timeout: options.timeoutMs,
|
||||
maxBuffer: options.maxBuffer,
|
||||
...(options.encoding !== undefined && { encoding: options.encoding }),
|
||||
...(typeof options.shell === "string" && { shell: options.shell }),
|
||||
...(options.env !== undefined && { env: options.env }),
|
||||
...(options.signal !== undefined && { signal: options.signal }),
|
||||
};
|
||||
const { stdout, stderr } = await execAsync(command, execOptions);
|
||||
|
||||
if (options.signal?.aborted) {
|
||||
return {
|
||||
stdout: stdout?.toString?.() ?? "",
|
||||
stderr: stderr?.toString?.() ?? "",
|
||||
exitCode: 0,
|
||||
stdout: "",
|
||||
stderr: "",
|
||||
exitCode: null,
|
||||
signal: null,
|
||||
timedOut: false,
|
||||
bufferExceeded: false,
|
||||
};
|
||||
} catch (error) {
|
||||
const errObj = error as Record<string, unknown>;
|
||||
const code = errObj.code;
|
||||
const status = typeof errObj.status === "number" ? errObj.status : null;
|
||||
const exitCode = typeof code === "number" ? code : status;
|
||||
const message = String(errObj.message ?? "");
|
||||
|
||||
return {
|
||||
stdout: typeof (errObj.stdout as { toString?: unknown })?.toString === "function" ? String(errObj.stdout) : "",
|
||||
stderr: typeof (errObj.stderr as { toString?: unknown })?.toString === "function" ? String(errObj.stderr) : "",
|
||||
exitCode,
|
||||
signal: (errObj.signal as NodeJS.Signals | null | undefined) ?? null,
|
||||
bufferExceeded:
|
||||
code === "ENOBUFS"
|
||||
|| code === "ERR_CHILD_PROCESS_STDIO_MAXBUFFER"
|
||||
|| message.includes("maxBuffer"),
|
||||
timedOut:
|
||||
code === "ETIMEDOUT"
|
||||
|| (errObj.killed === true && (errObj.signal === "SIGTERM" || message.includes("timed out"))),
|
||||
spawnError: code === "ENOENT" || code === "EACCES" ? (error as Error) : undefined,
|
||||
spawnError: new Error("Command aborted before start"),
|
||||
};
|
||||
}
|
||||
|
||||
return await new Promise((resolve) => {
|
||||
const supervised = superviseSpawn(command, [], {
|
||||
cwd: options.cwd,
|
||||
shell: options.shell ?? true,
|
||||
stdio: ["ignore", "pipe", "pipe"],
|
||||
...(options.env !== undefined && { env: options.env }),
|
||||
maxLifetimeMs: options.timeoutMs > 0 ? options.timeoutMs + FORCE_KILL_DELAY_MS + 1_000 : undefined,
|
||||
});
|
||||
const child = supervised.child;
|
||||
|
||||
const encoding = options.encoding ?? "utf-8";
|
||||
let stdout = "";
|
||||
let stderr = "";
|
||||
let bufferExceeded = false;
|
||||
let timedOut = false;
|
||||
let settled = false;
|
||||
let forceKillTimer: ReturnType<typeof setTimeout> | null = null;
|
||||
|
||||
const killTree = (signal: NodeJS.Signals): void => {
|
||||
supervised.kill(signal);
|
||||
};
|
||||
|
||||
const scheduleForceKill = (delayMs = FORCE_KILL_DELAY_MS): void => {
|
||||
if (forceKillTimer) return;
|
||||
forceKillTimer = setTimeout(() => {
|
||||
killTree("SIGKILL");
|
||||
}, delayMs);
|
||||
forceKillTimer.unref();
|
||||
};
|
||||
|
||||
const killTreeForCommandFailure = (): void => {
|
||||
killTree("SIGTERM");
|
||||
scheduleForceKill();
|
||||
};
|
||||
|
||||
const append = (current: string, chunk: Buffer): string => {
|
||||
if (bufferExceeded) return current;
|
||||
const text = chunk.toString(encoding);
|
||||
if (current.length + text.length <= options.maxBuffer) {
|
||||
return current + text;
|
||||
}
|
||||
bufferExceeded = true;
|
||||
const remaining = Math.max(0, options.maxBuffer - current.length);
|
||||
killTreeForCommandFailure();
|
||||
return current + text.slice(0, remaining);
|
||||
};
|
||||
|
||||
const timeout = options.timeoutMs > 0
|
||||
? setTimeout(() => {
|
||||
timedOut = true;
|
||||
killTreeForCommandFailure();
|
||||
}, options.timeoutMs)
|
||||
: null;
|
||||
timeout?.unref();
|
||||
|
||||
const onAbort = (): void => {
|
||||
killTreeForCommandFailure();
|
||||
};
|
||||
options.signal?.addEventListener("abort", onAbort, { once: true });
|
||||
|
||||
const finish = (spawnError: Error | null, exitCode: number | null, signal: NodeJS.Signals | null): void => {
|
||||
if (settled) return;
|
||||
settled = true;
|
||||
if (timeout) clearTimeout(timeout);
|
||||
if (forceKillTimer) clearTimeout(forceKillTimer);
|
||||
options.signal?.removeEventListener("abort", onAbort);
|
||||
|
||||
if (!spawnError) {
|
||||
killTree("SIGTERM");
|
||||
scheduleForceKill(NORMAL_CLEANUP_FORCE_KILL_DELAY_MS);
|
||||
}
|
||||
|
||||
resolve({
|
||||
stdout,
|
||||
stderr,
|
||||
exitCode,
|
||||
signal,
|
||||
timedOut,
|
||||
bufferExceeded,
|
||||
...(spawnError ? { spawnError } : {}),
|
||||
});
|
||||
};
|
||||
|
||||
child.stdout?.on("data", (chunk: Buffer) => {
|
||||
stdout = append(stdout, chunk);
|
||||
});
|
||||
child.stderr?.on("data", (chunk: Buffer) => {
|
||||
stderr = append(stderr, chunk);
|
||||
});
|
||||
child.on("error", (error) => finish(error, null, null));
|
||||
child.on("close", (code, signal) => finish(null, code, signal));
|
||||
});
|
||||
}
|
||||
|
||||
async runStreaming(command: string, options: SandboxRunStreamingOptions): Promise<SandboxStreamingResult> {
|
||||
@@ -105,28 +166,31 @@ export class NativeSandboxBackend implements SandboxBackend {
|
||||
let timedOut = false;
|
||||
let aborted = false;
|
||||
let settled = false;
|
||||
let forceKillTimer: ReturnType<typeof setTimeout> | null = null;
|
||||
|
||||
const killTree = (sig: NodeJS.Signals) => {
|
||||
supervised.kill(sig);
|
||||
};
|
||||
|
||||
const scheduleForceKill = (delayMs = FORCE_KILL_DELAY_MS): void => {
|
||||
if (forceKillTimer) return;
|
||||
forceKillTimer = setTimeout(() => {
|
||||
killTree("SIGKILL");
|
||||
}, delayMs);
|
||||
forceKillTimer.unref();
|
||||
};
|
||||
|
||||
const timer = setTimeout(() => {
|
||||
timedOut = true;
|
||||
killTree("SIGTERM");
|
||||
setTimeout(() => {
|
||||
if (settled) return;
|
||||
killTree("SIGKILL");
|
||||
}, 5_000).unref();
|
||||
scheduleForceKill();
|
||||
}, options.timeout);
|
||||
timer.unref();
|
||||
|
||||
const onAbort = () => {
|
||||
aborted = true;
|
||||
killTree("SIGTERM");
|
||||
setTimeout(() => {
|
||||
if (settled) return;
|
||||
killTree("SIGKILL");
|
||||
}, 5_000).unref();
|
||||
scheduleForceKill();
|
||||
};
|
||||
options.signal?.addEventListener("abort", onAbort, { once: true });
|
||||
|
||||
@@ -154,6 +218,7 @@ export class NativeSandboxBackend implements SandboxBackend {
|
||||
if (settled) return;
|
||||
settled = true;
|
||||
clearTimeout(timer);
|
||||
if (forceKillTimer) clearTimeout(forceKillTimer);
|
||||
options.signal?.removeEventListener("abort", onAbort);
|
||||
|
||||
if (aborted) {
|
||||
@@ -172,6 +237,8 @@ export class NativeSandboxBackend implements SandboxBackend {
|
||||
}
|
||||
|
||||
if (code === 0) {
|
||||
killTree("SIGTERM");
|
||||
scheduleForceKill(NORMAL_CLEANUP_FORCE_KILL_DELAY_MS);
|
||||
resolve({
|
||||
outcome: "success",
|
||||
stdout,
|
||||
|
||||
@@ -315,6 +315,40 @@ interface ConcurrencyGateDiagnostic {
|
||||
perColumnGates?: PerColumnCapacityGate[];
|
||||
}
|
||||
|
||||
const IDLE_SEMAPHORE_LEAK_REPAIR_MS = 5_000;
|
||||
|
||||
function persistedTopLevelAgentSlots(tasks: Task[]): number {
|
||||
return tasks.filter((task) => (
|
||||
task.column === "in-progress"
|
||||
|| (task.column === "triage" && task.status === "planning" && !task.paused)
|
||||
|| (task.column === "in-review" && ["merging", "reviewing", "fixing"].includes(String(task.status ?? "")))
|
||||
)).length;
|
||||
}
|
||||
|
||||
function recoverIdleSemaphoreLeak(
|
||||
semaphore: AgentSemaphore | undefined,
|
||||
tasks: Task[],
|
||||
source: string,
|
||||
candidateSinceMs: number | null,
|
||||
): number | null {
|
||||
if (!semaphore) return null;
|
||||
const persistedActive = persistedTopLevelAgentSlots(tasks);
|
||||
if (persistedActive !== 0 || semaphore.activeCount <= 0) return null;
|
||||
|
||||
const now = Date.now();
|
||||
if (candidateSinceMs === null) return now;
|
||||
if (now - candidateSinceMs < IDLE_SEMAPHORE_LEAK_REPAIR_MS) return candidateSinceMs;
|
||||
|
||||
const result = semaphore.reconcileActiveCount(0);
|
||||
if (result.changed) {
|
||||
schedulerLog.warn(
|
||||
`${source}: recovered stale semaphore active count ${result.before} -> ${result.after} ` +
|
||||
"(no persisted in-progress/planning/review agent work)",
|
||||
);
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
function computeConcurrencyGateDiagnostic(params: {
|
||||
agentSlots: number;
|
||||
maxConcurrent: number;
|
||||
@@ -495,6 +529,7 @@ export class Scheduler {
|
||||
private lastStaleTaskReportAt = 0;
|
||||
private lastBacklogPressureReportAt = 0;
|
||||
private lastUnlinkedMissionsAdvisoryReportAt = 0;
|
||||
private idleSemaphoreLeakCandidateSince: number | null = null;
|
||||
private readonly lastHighOverlapFanoutWarningKey = new Map<string, string>();
|
||||
|
||||
/**
|
||||
@@ -1208,6 +1243,12 @@ export class Scheduler {
|
||||
const settings = await this.store.getSettings();
|
||||
const maxConcurrent = settings.maxConcurrent ?? this.options.maxConcurrent ?? 2;
|
||||
const maxWorktrees = settings.maxWorktrees ?? this.options.maxWorktrees ?? 4;
|
||||
this.idleSemaphoreLeakCandidateSince = recoverIdleSemaphoreLeak(
|
||||
this.options.semaphore,
|
||||
tasks,
|
||||
"scheduler",
|
||||
this.idleSemaphoreLeakCandidateSince,
|
||||
);
|
||||
|
||||
// Refresh the poll interval if the persisted setting has changed
|
||||
this.refreshPollInterval(settings.pollIntervalMs);
|
||||
|
||||
@@ -634,6 +634,7 @@ export class TriageProcessor {
|
||||
private processingSince = new Map<string, number>();
|
||||
private wasGlobalPaused = false;
|
||||
private wasEnginePaused = false;
|
||||
private idleSemaphoreLeakCandidateSince: number | null = null;
|
||||
/** Active agent sessions per task, used to terminate on pause. */
|
||||
private activeSessions = new Map<string, { dispose: () => void }>();
|
||||
/**
|
||||
@@ -997,6 +998,31 @@ export class TriageProcessor {
|
||||
// Fetch all tasks (not just triage) to count active agents across columns.
|
||||
const allTasks = await this.store.listTasks({ slim: true, includeArchived: false });
|
||||
const now = Date.now();
|
||||
|
||||
if (this.options.semaphore) {
|
||||
const persistedActive = allTasks.filter((task) => (
|
||||
task.column === "in-progress"
|
||||
|| (task.column === "triage" && task.status === "planning" && !task.paused)
|
||||
|| (task.column === "in-review" && ["merging", "reviewing", "fixing"].includes(String(task.status ?? "")))
|
||||
)).length;
|
||||
if (persistedActive === 0 && this.options.semaphore.activeCount > 0 && this.processing.size === 0) {
|
||||
if (this.idleSemaphoreLeakCandidateSince === null) {
|
||||
this.idleSemaphoreLeakCandidateSince = now;
|
||||
} else if (now - this.idleSemaphoreLeakCandidateSince >= 5_000) {
|
||||
const result = this.options.semaphore.reconcileActiveCount(0);
|
||||
if (result.changed) {
|
||||
planLog.warn(
|
||||
`triage: recovered stale semaphore active count ${result.before} -> ${result.after} ` +
|
||||
"(no persisted in-progress/planning/review agent work)",
|
||||
);
|
||||
}
|
||||
this.idleSemaphoreLeakCandidateSince = null;
|
||||
}
|
||||
} else {
|
||||
this.idleSemaphoreLeakCandidateSince = null;
|
||||
}
|
||||
}
|
||||
|
||||
const eligibleTriageTasks = allTasks.filter(
|
||||
(t) => t.column === "triage" && !this.processing.has(t.id) && !t.paused
|
||||
// Skip tasks awaiting manual plan approval — they should not be auto-discovered
|
||||
@@ -1043,8 +1069,17 @@ export class TriageProcessor {
|
||||
const maxToStart = Math.min(perProjectAvailable, semaphoreAvailable);
|
||||
|
||||
if (maxToStart <= 0 && triageTasks.length > 0) {
|
||||
const semaphoreSnapshot = this.options.semaphore?.snapshot();
|
||||
const semaphoreDetail = semaphoreSnapshot
|
||||
? `, semaphore active=${semaphoreSnapshot.activeCount}/${semaphoreSnapshot.limit}, available=${semaphoreSnapshot.availableCount}, waiting=${semaphoreSnapshot.waitingCount}`
|
||||
: ", semaphore unavailable";
|
||||
const processingIds = [...this.processing].slice(0, 5);
|
||||
const eligibleIds = triageTasks.slice(0, 5).map((t) => t.id);
|
||||
const blockedBy = perProjectAvailable <= 0 ? "triage concurrency" : "global semaphore";
|
||||
planLog.log(
|
||||
`Plan throttled: ${activeAgents} planning agents, limit ${maxTriageConcurrent}`,
|
||||
`Plan throttled by ${blockedBy}: eligible=${triageTasks.length} [${eligibleIds.join(", ")}], ` +
|
||||
`planning=${activeAgents}/${maxTriageConcurrent}, processing=${this.processing.size}` +
|
||||
`${processingIds.length > 0 ? ` [${processingIds.join(", ")}]` : ""}${semaphoreDetail}`,
|
||||
);
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user