feat(workflow): add fn_workflow_step_resume operator escape hatch for stuck pending merge-review steps (#3339)
## Summary Adds an **operator-only** escape hatch for a card stranded `in-review` (or `in-progress`) with a workflow step permanently stuck in `pending` status — the leading real-world cause being a dispatched prompt node (e.g. `code-review`) whose verdict callback was never received (see #1946). Transitions the stuck `pending` pre-merge step to `status: "failed"` with resume audit metadata, so the existing `fn_task_bypass_review` escape hatch can then clear the merge blocker. ## What changed - **`WorkflowStepResult`** gains resume audit fields: `resumedBy`, `resumedAt`, `resumeReason`, `resumedFromStatus`. They are pure audit trail and **do not** participate in merge-blocking (`getTaskMergeBlocker`). - **`findPendingPreMergeStep`** (new helper, exported from `@fusion/core`) summarizes the stuck-pending pre-merge state for operator tooling. Ignores post-merge steps; returns the newest pending pre-merge result. - **`TaskStore.resumeWorkflowStep(id, { stepId, reason, actor })`** — the store primitive (eligibility-gated: task must be `in-review`/`in-progress`, not paused; step must exist and be `pending`; a mandatory non-blank `reason` and `stepId` are required). Runs under `withTaskLock`, writes the resume as a terminal `failed` result, appends a task-log breadcrumb, and emits the new `task:resume-step` run-audit event. - **`fn_workflow_step_resume`** — new CLI/pi-extension tool registered **only** on the operator surface (deliberately **not** wired into executor/reviewer/triage agent tool lists). Accepts `{ id, stepId, reason }`; the actor defaults to `cli-operator`. - **Run-audit**: new `task:resume-step` `DatabaseMutationType` member. ## Why A prompt-node verdict callback can be lost (dispatched prompt never receives a verdict), leaving the step `pending` forever. Previously the only recourse was `fn_task_bypass_review`, which requires a terminal *failed* pre-merge step to clear the blocker — a permanently `pending` step could not be bypassed. This PR bridges that gap: resume (pending → failed) then bypass (failed merge-blocker cleared). ## Verification - **Typecheck**: `@fusion/core`, `@fusion/engine`, `@runfusion/fusion` all clean. - **`task-merge-bypass.test.ts`**: 15/15 pass (incl. 5 new `findPendingPreMergeStep` cases). - **`store-resume-step.test.ts`** (new, PG-backed): 9/9 pass — eligibility gating, resume rewrite + audit fields, run-audit event, non-pending/non-found/blank-argument rejection, in-progress column support, property preservation. - **`extension.test.ts`**: 75/75 pass (expected-tool registration includes the new tool). ## Files - `packages/core/src/types/workflow/workflow-steps.ts` - `packages/core/src/merge/task-merge.ts` - `packages/core/src/store.ts` - `packages/core/src/index.ts` - `packages/core/src/__tests__/store-resume-step.test.ts` (new) - `packages/core/src/__tests__/task-merge-bypass.test.ts` - `packages/engine/src/util/run-audit.ts` - `packages/cli/src/extension.ts` - `packages/cli/src/__tests__/extension.test.ts` - `.changeset/stas-032-resume-workflow-step.md` (minor, feature) <!-- This is an auto-generated comment: release notes by coderabbit.ai --> ## Summary by CodeRabbit * **New Features** * Added an operator-only workflow recovery tool for permanently pending pre-merge steps. * Operators can mark eligible pending steps as failed by providing a required audit reason. * Recovery actions record operator details, timestamps, reasons, prior status, task logs, and audit events. * **Bug Fixes** * Improved selection of the latest pending pre-merge workflow step while excluding post-merge steps. * Added validation to prevent recovery of paused, invalid, or out-of-scope workflow steps. <!-- end of auto-generated comment: release notes by coderabbit.ai --> --------- Co-authored-by: schindler <schindler@users.noreply.github.com> Co-authored-by: gsxdsm <gsxdsm@users.noreply.github.com>
This commit is contained in:
ischindl
7
.changeset/stas-032-resume-workflow-step.md
Normal file
7
.changeset/stas-032-resume-workflow-step.md
Normal file
@@ -0,0 +1,7 @@
|
||||
---
|
||||
"@runfusion/fusion": minor
|
||||
---
|
||||
|
||||
summary: Add fn_workflow_step_resume operator tool to unstick permanently-pending merge review steps.
|
||||
category: feature
|
||||
dev: New CLI/pi-extension operator-only tool `fn_workflow_step_resume` (with `TaskStore.resumeWorkflowStep` + `findPendingPreMergeStep` helper) transitions a stuck `pending` pre-merge workflow step to `failed` with resume audit metadata so the existing `fn_task_bypass_review` escape hatch can clear the merge blocker. Audit-logged via the new `task:resume-step` run-audit event. Not exposed to executor/reviewer/triage agent surfaces.
|
||||
@@ -270,6 +270,7 @@ legacyDescribe("fn pi extension (legacy exhaustive suite)", () => {
|
||||
"fn_task_unpause",
|
||||
"fn_task_retry",
|
||||
"fn_task_bypass_review",
|
||||
"fn_workflow_step_resume",
|
||||
"fn_task_duplicate",
|
||||
"fn_task_refine",
|
||||
"fn_task_import_github",
|
||||
|
||||
@@ -871,6 +871,7 @@ The guard runs FIRST in each tool's execute, before any store access or param va
|
||||
const WITHHELD_FROM_AGENT_EXTENSION_TOOLS: ReadonlySet<string> = new Set([
|
||||
"fn_task_delete",
|
||||
"fn_task_bypass_review",
|
||||
"fn_workflow_step_resume",
|
||||
"fn_mission_delete",
|
||||
"fn_milestone_delete",
|
||||
"fn_slice_delete",
|
||||
@@ -2575,6 +2576,67 @@ export default function kbExtension(pi: ExtensionAPI) {
|
||||
},
|
||||
});
|
||||
|
||||
// ── fn_workflow_step_resume ────────────────────────────────────
|
||||
|
||||
/*
|
||||
* FNXC:StepResume 2026-08-06-17:42:
|
||||
* Operator escape hatch for in-review/in-progress tasks with workflow steps
|
||||
* stuck in `pending` because a dispatched prompt node verdict callback was
|
||||
* never received (Runfusion/Fusion#1946). Transitions the stuck `pending` step to
|
||||
* `failed` so the existing `fn_task_bypass_review` escape hatch can then clear
|
||||
* the merge blocker. Registered ONLY on this pi-extension/CLI operator tool
|
||||
* surface — deliberately NOT wired into executor/reviewer/triage agent tool
|
||||
* lists; the WITHHELD_FROM_AGENT_EXTENSION_TOOLS guard below is the hard
|
||||
* enforcement that an agent session cannot reach it (operator-only, mandatory
|
||||
* `reason` and `stepId`, audit-logged via store.resumeWorkflowStep's run-audit
|
||||
* event).
|
||||
*/
|
||||
pi.registerTool({
|
||||
name: "fn_workflow_step_resume",
|
||||
label: "fn: Resume Stuck Pending Step",
|
||||
description:
|
||||
"Resume a stuck pending workflow step on an in-review or in-progress Fusion task " +
|
||||
"(operator-only, mandatory reason, audit-logged). When a prompt node (like code-review) " +
|
||||
"is dispatched but never receives a verdict callback (Runfusion/Fusion#1946), the step " +
|
||||
"stays in 'pending' status indefinitely. This tool transitions it to 'failed', enabling " +
|
||||
"the existing fn_task_bypass_review escape hatch to clear the merge blocker. Requires " +
|
||||
"a mandatory reason and step ID.",
|
||||
promptSnippet:
|
||||
"Resume a stuck pending workflow step on an in-review or in-progress Fusion task (operator-only, mandatory reason, audit-logged)",
|
||||
parameters: Type.Object({
|
||||
id: Type.String({ description: "Task ID (e.g. FN-001)" }),
|
||||
stepId: Type.String({ description: "Workflow step ID to resume (e.g. 'code-review', 'plan-review')" }),
|
||||
reason: Type.String({ description: "Mandatory justification for resuming the step (audit-logged)" }),
|
||||
}),
|
||||
|
||||
async execute(_toolCallId, params, _signal, _onUpdate, ctx) {
|
||||
const withheldDenied = denyWithheldToolForAgentPrincipal("fn_workflow_step_resume", ctx as ExtensionCallerContext);
|
||||
if (withheldDenied) return withheldDenied;
|
||||
const store = await getStore(ctx.cwd);
|
||||
const fnCtx = ctx as typeof ctx & { agentId?: string };
|
||||
const actor = fnCtx.agentId ?? "cli-operator";
|
||||
|
||||
try {
|
||||
const task = await store.resumeWorkflowStep(params.id, {
|
||||
stepId: params.stepId,
|
||||
reason: params.reason,
|
||||
actor,
|
||||
});
|
||||
return {
|
||||
content: [{ type: "text", text: `Resumed stuck pending workflow step '${params.stepId}' for ${task.id}` }],
|
||||
details: { taskId: task.id, stepId: params.stepId },
|
||||
};
|
||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||||
} catch (err: any) {
|
||||
return {
|
||||
content: [{ type: "text", text: `ERROR: Failed to resume step '${params.stepId}' for ${params.id}: ${err?.message ?? err}` }],
|
||||
isError: true,
|
||||
details: { taskId: params.id, stepId: params.stepId, error: String(err?.message ?? err) },
|
||||
};
|
||||
}
|
||||
},
|
||||
});
|
||||
|
||||
// ── fn_task_duplicate ─────────────────────────────────────────────
|
||||
|
||||
pi.registerTool({
|
||||
|
||||
233
packages/core/src/__tests__/store-resume-step.test.ts
Normal file
233
packages/core/src/__tests__/store-resume-step.test.ts
Normal file
@@ -0,0 +1,233 @@
|
||||
import { afterAll, afterEach, beforeAll, beforeEach, expect, it } from "vitest";
|
||||
import type { WorkflowStepResult } from "../types.js";
|
||||
import {
|
||||
pgDescribe,
|
||||
createSharedPgTaskStoreTestHarness,
|
||||
type SharedPgTaskStoreHarness,
|
||||
} from "../__test-utils__/pg-test-harness.js";
|
||||
import { queryRunAuditEvents } from "../task-store/async/async-audit.js";
|
||||
|
||||
/*
|
||||
* FNXC:StepResume 2026-07-24-13:00:
|
||||
* Store-level coverage for the resumeWorkflowStep primitive: eligibility gating
|
||||
* (in-review or in-progress, step is pending, mandatory reason + stepId), the
|
||||
* resume rewrite (status -> failed + audit metadata), the run-audit event/log
|
||||
* breadcrumb, and rejection of non-pending steps.
|
||||
*/
|
||||
|
||||
pgDescribe("TaskStore.resumeWorkflowStep", () => {
|
||||
const h: SharedPgTaskStoreHarness = createSharedPgTaskStoreTestHarness({
|
||||
prefix: "fusion_resume_step",
|
||||
});
|
||||
|
||||
beforeAll(h.beforeAll);
|
||||
beforeEach(h.beforeEach);
|
||||
afterEach(h.afterEach);
|
||||
afterAll(h.afterAll);
|
||||
|
||||
function pendingStep(overrides: Partial<WorkflowStepResult> = {}): WorkflowStepResult {
|
||||
return {
|
||||
workflowStepId: "code-review",
|
||||
workflowStepName: "Code Review",
|
||||
phase: "pre-merge",
|
||||
source: "optional-group",
|
||||
status: "pending",
|
||||
startedAt: "2026-07-17T16:10:10.052Z",
|
||||
...overrides,
|
||||
};
|
||||
}
|
||||
|
||||
function store() {
|
||||
return h.store();
|
||||
}
|
||||
|
||||
async function seedInReviewTask(
|
||||
id: string,
|
||||
options: {
|
||||
workflowStepResults?: WorkflowStepResult[];
|
||||
paused?: boolean;
|
||||
column?: string;
|
||||
} = {},
|
||||
) {
|
||||
const column = options.column ?? "in-review";
|
||||
await store().createTaskWithReservedId(
|
||||
{ description: `Task ${id}`, column },
|
||||
{ taskId: id, applyDefaultWorkflowSteps: false },
|
||||
);
|
||||
await store().updateTask(id, {
|
||||
workflowStepResults: options.workflowStepResults ?? null,
|
||||
paused: options.paused,
|
||||
});
|
||||
return store().getTask(id);
|
||||
}
|
||||
|
||||
it("transitions a pending step to failed with audit metadata", async () => {
|
||||
await seedInReviewTask("FN-RES-001", { workflowStepResults: [pendingStep()] });
|
||||
|
||||
const updated = await store().resumeWorkflowStep("FN-RES-001", {
|
||||
stepId: "code-review",
|
||||
reason: "Runfusion/Fusion#1946 no-verdict dispatch defect",
|
||||
actor: "operator-1",
|
||||
});
|
||||
|
||||
const result = updated.workflowStepResults?.[0];
|
||||
expect(result?.status).toBe("failed");
|
||||
expect(result?.completedAt).toBeDefined();
|
||||
expect(result?.resumedBy).toBe("operator-1");
|
||||
expect(result?.resumeReason).toBe("Runfusion/Fusion#1946 no-verdict dispatch defect");
|
||||
expect(result?.resumedFromStatus).toBe("pending");
|
||||
expect(typeof result?.resumedAt).toBe("string");
|
||||
|
||||
// Audit trail: task log entry recorded.
|
||||
const logged = updated.log?.some((entry) => entry.action.includes("Workflow step resumed"));
|
||||
expect(logged).toBe(true);
|
||||
});
|
||||
|
||||
it("records a run-audit event for the resume", async () => {
|
||||
await seedInReviewTask("FN-RES-002", { workflowStepResults: [pendingStep()] });
|
||||
await store().resumeWorkflowStep("FN-RES-002", {
|
||||
stepId: "code-review",
|
||||
reason: "infra failure",
|
||||
actor: "operator-2",
|
||||
});
|
||||
|
||||
const events = await queryRunAuditEvents(h.layer().db, { taskId: "FN-RES-002" });
|
||||
const resumeEvent = events.find((event) => event.mutationType === "task:resume-step");
|
||||
expect(resumeEvent).toBeDefined();
|
||||
expect(resumeEvent?.agentId).toBe("operator-2");
|
||||
});
|
||||
|
||||
it("rejects when the step is not pending", async () => {
|
||||
await seedInReviewTask("FN-RES-003", {
|
||||
workflowStepResults: [pendingStep({ status: "passed" })],
|
||||
});
|
||||
await expect(
|
||||
store().resumeWorkflowStep("FN-RES-003", {
|
||||
stepId: "code-review",
|
||||
reason: "x",
|
||||
actor: "operator",
|
||||
}),
|
||||
).rejects.toThrow(/only pending steps can be resumed/);
|
||||
});
|
||||
|
||||
it("rejects when the step is not found as a pending pre-merge step", async () => {
|
||||
await seedInReviewTask("FN-RES-004", { workflowStepResults: [] });
|
||||
await expect(
|
||||
store().resumeWorkflowStep("FN-RES-004", {
|
||||
stepId: "non-existent-step",
|
||||
reason: "x",
|
||||
actor: "operator",
|
||||
}),
|
||||
).rejects.toThrow(/not found as a pending pre-merge step/);
|
||||
});
|
||||
|
||||
it("rejects resuming a post-merge step (pre-merge boundary, FNXC:StepResume)", async () => {
|
||||
await seedInReviewTask("FN-RES-004B", {
|
||||
workflowStepResults: [pendingStep({ workflowStepId: "post-deploy", workflowStepName: "Post Deploy", phase: "post-merge" })],
|
||||
});
|
||||
await expect(
|
||||
store().resumeWorkflowStep("FN-RES-004B", {
|
||||
stepId: "post-deploy",
|
||||
reason: "x",
|
||||
actor: "operator",
|
||||
}),
|
||||
).rejects.toThrow(/not found as a pending pre-merge step/);
|
||||
});
|
||||
|
||||
it("rejects a blank reason", async () => {
|
||||
await seedInReviewTask("FN-RES-005", { workflowStepResults: [pendingStep()] });
|
||||
await expect(
|
||||
store().resumeWorkflowStep("FN-RES-005", {
|
||||
stepId: "code-review",
|
||||
reason: " ",
|
||||
actor: "operator",
|
||||
}),
|
||||
).rejects.toThrow(/non-empty reason/);
|
||||
});
|
||||
|
||||
it("rejects a blank stepId", async () => {
|
||||
await seedInReviewTask("FN-RES-006", { workflowStepResults: [pendingStep()] });
|
||||
await expect(
|
||||
store().resumeWorkflowStep("FN-RES-006", {
|
||||
stepId: "",
|
||||
reason: "x",
|
||||
actor: "operator",
|
||||
}),
|
||||
).rejects.toThrow(/non-empty stepId/);
|
||||
});
|
||||
|
||||
it("rejects when the task is not in-review or in-progress", async () => {
|
||||
await seedInReviewTask("FN-RES-007", {
|
||||
workflowStepResults: [pendingStep()],
|
||||
column: "todo",
|
||||
});
|
||||
await expect(
|
||||
store().resumeWorkflowStep("FN-RES-007", {
|
||||
stepId: "code-review",
|
||||
reason: "x",
|
||||
actor: "operator",
|
||||
}),
|
||||
).rejects.toThrow(/task is in 'todo', must be in .* or a WIP/);
|
||||
});
|
||||
|
||||
it("works on tasks in in-progress column", async () => {
|
||||
await seedInReviewTask("FN-RES-008", {
|
||||
workflowStepResults: [pendingStep()],
|
||||
column: "in-progress",
|
||||
});
|
||||
|
||||
const updated = await store().resumeWorkflowStep("FN-RES-008", {
|
||||
stepId: "code-review",
|
||||
reason: "stuck pending step in execution",
|
||||
actor: "operator-3",
|
||||
});
|
||||
|
||||
const result = updated.workflowStepResults?.[0];
|
||||
expect(result?.status).toBe("failed");
|
||||
expect(result?.resumedBy).toBe("operator-3");
|
||||
expect(result?.resumedFromStatus).toBe("pending");
|
||||
});
|
||||
|
||||
it("preserves existing pending step properties after resume", async () => {
|
||||
await seedInReviewTask("FN-RES-009", {
|
||||
workflowStepResults: [
|
||||
pendingStep({ source: "optional-group", startedAt: "2026-07-17T16:10:10.052Z" }),
|
||||
],
|
||||
});
|
||||
|
||||
const updated = await store().resumeWorkflowStep("FN-RES-009", {
|
||||
stepId: "code-review",
|
||||
reason: "dispatch callback never received",
|
||||
actor: "operator",
|
||||
});
|
||||
|
||||
const result = updated.workflowStepResults?.[0];
|
||||
expect(result?.workflowStepId).toBe("code-review");
|
||||
expect(result?.workflowStepName).toBe("Code Review");
|
||||
expect(result?.phase).toBe("pre-merge");
|
||||
expect(result?.source).toBe("optional-group");
|
||||
expect(result?.startedAt).toBe("2026-07-17T16:10:10.052Z");
|
||||
expect(result?.status).toBe("failed");
|
||||
expect(result?.resumedFromStatus).toBe("pending");
|
||||
});
|
||||
|
||||
it("clears lease ownership on the resumed step result (FNXC:StepResume lease cleanup)", async () => {
|
||||
await seedInReviewTask("FN-RES-010", {
|
||||
workflowStepResults: [
|
||||
pendingStep({ leaseOwner: "agent-reviewer-1", leaseNodeId: "review-1" }),
|
||||
],
|
||||
});
|
||||
|
||||
const updated = await store().resumeWorkflowStep("FN-RES-010", {
|
||||
stepId: "code-review",
|
||||
reason: "lease owner never completed the verdict callback",
|
||||
actor: "operator",
|
||||
});
|
||||
|
||||
const result = updated.workflowStepResults?.[0];
|
||||
expect(result?.status).toBe("failed");
|
||||
// A terminal 'failed' result must not carry the stale dispatch lease forward.
|
||||
expect(result?.leaseOwner).toBeUndefined();
|
||||
expect(result?.leaseNodeId).toBeUndefined();
|
||||
});
|
||||
});
|
||||
@@ -1,6 +1,6 @@
|
||||
import { describe, it, expect } from "vitest";
|
||||
import type { StepStatus, WorkflowStepResult } from "../types.js";
|
||||
import { getLatestFailedPreMergeReviewStep, getTaskMergeBlocker } from "../merge/task-merge.js";
|
||||
import { findPendingPreMergeStep, getLatestFailedPreMergeReviewStep, getTaskMergeBlocker } from "../merge/task-merge.js";
|
||||
|
||||
/*
|
||||
* FNXC:ReviewLaneBypass 2026-07-09-00:00:
|
||||
@@ -176,3 +176,44 @@ describe("bypass invariant on getTaskMergeBlocker", () => {
|
||||
expect(getTaskMergeBlocker(task)).toMatch(/marked 'stuck-killed'/);
|
||||
});
|
||||
});
|
||||
|
||||
describe("findPendingPreMergeStep", () => {
|
||||
it("returns undefined when workflowStepResults is undefined", () => {
|
||||
expect(findPendingPreMergeStep({ workflowStepResults: undefined })).toBeUndefined();
|
||||
});
|
||||
|
||||
it("returns undefined when no pre-merge step is pending", () => {
|
||||
const results = [
|
||||
stepResult({ status: "passed" }),
|
||||
stepResult({ status: "failed" }),
|
||||
stepResult({ workflowStepId: "WS-post", phase: "post-merge", status: "pending" }),
|
||||
];
|
||||
expect(findPendingPreMergeStep({ workflowStepResults: results })).toBeUndefined();
|
||||
});
|
||||
|
||||
it("returns the pending result when one exists", () => {
|
||||
const results = [stepResult({ status: "pending" })];
|
||||
const found = findPendingPreMergeStep({ workflowStepResults: results });
|
||||
expect(found?.workflowStepId).toBe("WS-001");
|
||||
expect(found?.status).toBe("pending");
|
||||
});
|
||||
|
||||
it("returns the latest pending result when multiple exist", () => {
|
||||
const results = [
|
||||
stepResult({ workflowStepId: "WS-older", startedAt: "2026-07-17T10:00:00.000Z", status: "pending" }),
|
||||
stepResult({ workflowStepId: "WS-newer", startedAt: "2026-07-17T16:10:10.052Z", status: "pending" }),
|
||||
];
|
||||
const found = findPendingPreMergeStep({ workflowStepResults: results });
|
||||
expect(found?.workflowStepId).toBe("WS-newer");
|
||||
});
|
||||
|
||||
it("does NOT return failed or passed results (only pending)", () => {
|
||||
const results = [
|
||||
stepResult({ workflowStepId: "WS-passed", status: "passed" }),
|
||||
stepResult({ workflowStepId: "WS-failed", status: "failed" }),
|
||||
stepResult({ workflowStepId: "WS-pending", status: "pending" }),
|
||||
];
|
||||
const found = findPendingPreMergeStep({ workflowStepResults: results });
|
||||
expect(found?.workflowStepId).toBe("WS-pending");
|
||||
});
|
||||
});
|
||||
|
||||
@@ -1153,6 +1153,7 @@ export {
|
||||
resolveTaskMergeTarget,
|
||||
AWAITING_APPROVAL_PAUSE_REASON,
|
||||
isTaskBlockedOnApproval,
|
||||
findPendingPreMergeStep,
|
||||
type MergeTargetResolution,
|
||||
type MergeTargetResolverOptions,
|
||||
} from "./merge/task-merge.js";
|
||||
|
||||
@@ -566,3 +566,26 @@ export async function getTaskCompletionBlocker(
|
||||
|
||||
return undefined;
|
||||
}
|
||||
|
||||
/*
|
||||
FNXC:StepResume 2026-07-19-21:34:
|
||||
Operator escape hatch for in-review tasks with permanently pending workflow steps.
|
||||
Finds the latest pre-merge workflow step in pending status so the operator can
|
||||
then resume/bypass it. Does not consider post-merge steps.
|
||||
*/
|
||||
export function findPendingPreMergeStep(
|
||||
task: Pick<Task, "workflowStepResults">,
|
||||
): WorkflowStepResult | undefined {
|
||||
if (!task.workflowStepResults) return undefined;
|
||||
|
||||
const pendingPreMerge = task.workflowStepResults.filter(
|
||||
(step) => step.phase !== "post-merge" && step.status === "pending",
|
||||
);
|
||||
|
||||
if (pendingPreMerge.length === 0) return undefined;
|
||||
|
||||
// Return the newest pending pre-merge step (by startedAt, descending)
|
||||
return pendingPreMerge.sort(
|
||||
(a, b) => new Date(b.startedAt ?? 0).getTime() - new Date(a.startedAt ?? 0).getTime(),
|
||||
)[0];
|
||||
}
|
||||
|
||||
@@ -79,7 +79,7 @@ import { EvalStore } from "./eval/eval-store.js";
|
||||
import { AsyncEvalStore } from "./async-stores/async-eval-store.js";
|
||||
import { CentralCore } from "./central/central-core.js";
|
||||
import { SecretsStore } from "./secrets/secrets-store.js";
|
||||
import { getLatestFailedPreMergeReviewStep } from "./merge/task-merge.js";
|
||||
import { getLatestFailedPreMergeReviewStep, findPendingPreMergeStep } from "./merge/task-merge.js";
|
||||
import { createLogger } from "./process/logger.js";
|
||||
import { type UsageEventInput } from "./tasks/usage-events.js";
|
||||
import { assertNotLinkedWorktreeOfExistingProject, assertProjectRootDir } from "./central/project-root-guard.js";
|
||||
@@ -1687,6 +1687,136 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
|
||||
return task;
|
||||
});
|
||||
}
|
||||
/*
|
||||
* FNXC:StepResume 2026-08-06-02:12:
|
||||
* STAS-032: Operator/privileged-only escape hatch for a card stranded in
|
||||
* `in-review` or `in-progress` with a workflow step permanently in `pending`
|
||||
* status (leading real-world cause: the Runfusion/Fusion#1946 dispatched
|
||||
* prompt node verdict callback never received). Transitions the stuck
|
||||
* `pending` pre-merge step to `status: "failed"` with resume audit metadata
|
||||
* (who/when/why/prior status) so the existing `fn_task_bypass_review` escape
|
||||
* hatch can then clear the merge blocker (FN-7720). Requires a mandatory
|
||||
* `reason` and `stepId`; audit-logged via the `task:resume-step` run-audit
|
||||
* event. A resumed result is a terminal `failed` result and does NOT clear,
|
||||
* create, or alter any other merge-blocker condition. NOT exposed to
|
||||
* executor/reviewer/triage agent tool surfaces — see `fn_workflow_step_resume`
|
||||
* registration comments for the same rule.
|
||||
*/
|
||||
async resumeWorkflowStep(
|
||||
id: string,
|
||||
options: { stepId: string; reason: string; actor: string },
|
||||
): Promise<Task> {
|
||||
const reason = options.reason?.trim();
|
||||
if (!reason) {
|
||||
throw new Error("resumeWorkflowStep requires a non-empty reason");
|
||||
}
|
||||
const stepId = options.stepId?.trim();
|
||||
if (!stepId) {
|
||||
throw new Error("resumeWorkflowStep requires a non-empty stepId");
|
||||
}
|
||||
const actor = options.actor?.trim() || "operator";
|
||||
|
||||
return this.withTaskLock(id, async () => {
|
||||
const dir = this.taskDir(id);
|
||||
const task = await this.readTaskJson(dir);
|
||||
|
||||
if (task.paused) {
|
||||
throw new Error(`Cannot resume workflow step for ${id}: task is paused`);
|
||||
}
|
||||
|
||||
// FNXC:StepResume 2026-08-06-17:42:
|
||||
// Resolve the review and WIP lanes against the task's actual workflow IR instead of
|
||||
// hardcoded 'in-review'/'in-progress' literals. A board whose review lane is named
|
||||
// differently (or carries review on a humanReview/mergeBlocker-only lane) would
|
||||
// otherwise reject a legitimately stuck task. This mirrors bypassFailedPreMergeReviewStep's
|
||||
// lane resolution; the WIP side uses the workflow's countsTowardWip columns.
|
||||
const resumeIr = await resolveWorkflowIrForTask(this, task.id).catch(() => undefined);
|
||||
const reviewColumns: ReadonlySet<string> =
|
||||
resumeIr === undefined || !declaresAnyLifecycleTrait(resumeIr)
|
||||
? new Set(["in-review"])
|
||||
: new Set(resolveReviewColumns(resumeIr));
|
||||
const wipColumns = await resolveProjectColumnsForRoles(this, ["countsTowardWip"]);
|
||||
const resumeInReview = reviewColumns.has(task.column);
|
||||
const resumeInProgress = wipColumns.has(task.column);
|
||||
if (!resumeInReview && !resumeInProgress) {
|
||||
const named = reviewColumns.size > 0 ? [...reviewColumns].map((c) => `'${c}'`).join(" or ") : "a review lane";
|
||||
throw new Error(
|
||||
`Cannot resume workflow step for ${id}: task is in '${task.column}', must be in ${named} or a WIP (in-progress) lane`,
|
||||
);
|
||||
}
|
||||
|
||||
const results = task.workflowStepResults ?? [];
|
||||
// Only a pending PRE-MERGE step may be resumed: post-merge steps are never the stuck prompt
|
||||
// verdict target, and resuming one would fabricate a terminal 'failed' result the merge gate
|
||||
// does not own. findPendingPreMergeStep (used to name the candidate below) enforces the same
|
||||
// pre-merge boundary as the operator-only resume surface.
|
||||
const target = results.find((r) => r.workflowStepId === stepId && r.phase !== "post-merge");
|
||||
if (!target) {
|
||||
const pendingPreMerge = findPendingPreMergeStep(task);
|
||||
const candidateName = pendingPreMerge ? pendingPreMerge.workflowStepName : undefined;
|
||||
throw new Error(
|
||||
`Cannot resume workflow step for ${id}: step '${stepId}' not found as a pending pre-merge step` +
|
||||
(candidateName ? ` (pending pre-merge step found: '${candidateName}')` : ""),
|
||||
);
|
||||
}
|
||||
if (target.status !== "pending") {
|
||||
throw new Error(
|
||||
`Cannot resume workflow step for ${id}: only pending steps can be resumed`,
|
||||
);
|
||||
}
|
||||
|
||||
const now = new Date().toISOString();
|
||||
const resumed: import("./types.js").WorkflowStepResult = {
|
||||
...target,
|
||||
status: "failed",
|
||||
completedAt: now,
|
||||
resumedAt: now,
|
||||
resumedBy: actor,
|
||||
resumeReason: reason,
|
||||
resumedFromStatus: target.status,
|
||||
};
|
||||
// A resumed result is terminal 'failed' — never carry lease ownership forward onto a
|
||||
// completed step result. The lease was held by the (never-completed) dispatched prompt node;
|
||||
// preserving leaseOwner/leaseNodeId on the failed record would strand successor lease logic.
|
||||
delete resumed.leaseOwner;
|
||||
delete resumed.leaseNodeId;
|
||||
|
||||
const nextResults = [...results];
|
||||
const targetIndex = nextResults.indexOf(target);
|
||||
nextResults[targetIndex] = resumed;
|
||||
task.workflowStepResults = nextResults;
|
||||
|
||||
if (!task.log) {
|
||||
task.log = [];
|
||||
}
|
||||
task.updatedAt = now;
|
||||
task.log.push({
|
||||
timestamp: now,
|
||||
action: `Workflow step resumed: ${target.workflowStepName} (${target.workflowStepId}) by ${actor} — ${reason}`,
|
||||
});
|
||||
|
||||
await this.recordRunAuditEvent({
|
||||
taskId: task.id,
|
||||
agentId: actor,
|
||||
runId: this.makeSyntheticDeleteRunId(task.id),
|
||||
domain: "database",
|
||||
mutationType: "task:resume-step",
|
||||
target: task.id,
|
||||
metadata: {
|
||||
workflowStepId: target.workflowStepId,
|
||||
workflowStepName: target.workflowStepName,
|
||||
resumedFromStatus: target.status,
|
||||
reason,
|
||||
},
|
||||
});
|
||||
|
||||
await this.atomicWriteTaskJson(dir, task);
|
||||
if (this.isWatching) this.taskCache.set(id, { ...task });
|
||||
|
||||
this.emit("task:updated", task);
|
||||
return task;
|
||||
});
|
||||
}
|
||||
/*
|
||||
FNXC:WorkflowEvents 2026-08-01-06:11:
|
||||
One seam decorates every TaskStore `task:updated` emission, including hot synchronous paths, with a
|
||||
|
||||
@@ -358,6 +358,22 @@ export interface WorkflowStepResult {
|
||||
* Reset when a superseded planning episode is replaced.
|
||||
*/
|
||||
planReviewAttemptCount?: number;
|
||||
/*
|
||||
* FNXC:StepResume 2026-07-24-13:00:
|
||||
* STAS-032: A stuck `pending` pre-merge workflow step (caused by
|
||||
* Runfusion/Fusion#1946 dispatched prompt node verdict callback never
|
||||
* received) can be resumed to `failed` via `fn_workflow_step_resume`.
|
||||
* These fields are stamped as audit trail — they do NOT participate in
|
||||
* merge-blocking (getTaskMergeBlocker).
|
||||
*/
|
||||
/** Operator identity that performed the resume, if this result was resumed from pending. */
|
||||
resumedBy?: string;
|
||||
/** ISO-8601 timestamp when the resume was applied. */
|
||||
resumedAt?: string;
|
||||
/** Mandatory operator-supplied justification for resuming this pending step. */
|
||||
resumeReason?: string;
|
||||
/** The `status` this result carried immediately before the resume rewrote it (always `"pending"`). */
|
||||
resumedFromStatus?: WorkflowStepResult["status"];
|
||||
/*
|
||||
* FNXC:WorkflowStepResults 2026-07-09-00:10:
|
||||
* FN-7727: self-healing recovery re-runs a failed pre-merge review node
|
||||
|
||||
@@ -497,6 +497,7 @@ export type DatabaseMutationType =
|
||||
| "task:release"
|
||||
| "task:pause"
|
||||
| "task:unpause"
|
||||
| "task:resume-step"
|
||||
| "task:dependency:add"
|
||||
| "merge:request-enqueued"
|
||||
| "merge:dependency-parity-diff"
|
||||
|
||||
Reference in New Issue
Block a user