diff --git a/.changeset/workflow-review-followups.md b/.changeset/workflow-review-followups.md new file mode 100644 index 0000000000..26f1fa4dc4 --- /dev/null +++ b/.changeset/workflow-review-followups.md @@ -0,0 +1,7 @@ +--- +"@runfusion/fusion": patch +--- + +summary: Preserve workflow lifecycle state and start execution steps only after worktree creation. +category: fix +dev: Adds shared active-state semantics, lifecycle records, and worktree-first graph step projection. diff --git a/packages/core/src/__tests__/builtin-coding-ideas-workflow-ir.test.ts b/packages/core/src/__tests__/builtin-coding-ideas-workflow-ir.test.ts index 4c6f9bf7b1..4c67df9d08 100644 --- a/packages/core/src/__tests__/builtin-coding-ideas-workflow-ir.test.ts +++ b/packages/core/src/__tests__/builtin-coding-ideas-workflow-ir.test.ts @@ -100,9 +100,13 @@ describe("builtin coding-ideas workflow ir", () => { expect(nodeColumn("parse")).toBe("in-progress"); expect(nodeColumn("steps")).toBe("in-progress"); expect(nodeColumn("code-review")).toBe("in-review"); + expect(nodeColumn("code-review-remediation")).toBe("in-progress"); expect(nodeColumn("completion-summary")).toBe("in-review"); - expect(nodeColumn("merge-gate")).toBe("in-review"); + for (const node of ir.nodes.filter((candidate) => candidate.id.startsWith("merge-"))) { + expect(node.column, `${node.id} should remain in review`).toBe("in-review"); + } expect(ir.nodes.some((node) => node.id === "browser-verification")).toBe(false); + expect(ir.nodes.some((node) => node.id === "browser-verification-remediation")).toBe(false); expect(ir.nodes.some((node) => node.id === "post-merge-verification")).toBe(false); }); diff --git a/packages/core/src/builtin-coding-ideas-workflow-ir.ts b/packages/core/src/builtin-coding-ideas-workflow-ir.ts index 68b06cf8b0..aa5caaf013 100644 --- a/packages/core/src/builtin-coding-ideas-workflow-ir.ts +++ b/packages/core/src/builtin-coding-ideas-workflow-ir.ts @@ -94,9 +94,12 @@ const RAW_BUILTIN_CODING_IDEAS_WORKFLOW_IR: WorkflowIr = (() => { } } + // FNXC:CodingIdeasWorkflow 2026-07-21-12:20: // Keep this preset intentionally small: planning + plan review in Todo, // implementation in In progress, then code review and merge in In review. - // Browser and post-merge verification remain available in richer workflows. + // Browser and post-merge verification remain available in richer workflows; + // direct success edges replace the removed nodes so the reduced preset keeps + // one continuous executable path. const removedNodeIds = new Set([ "browser-verification", "browser-verification-remediation", diff --git a/packages/core/src/index.gate.ts b/packages/core/src/index.gate.ts index c303d415c2..dfec191174 100644 --- a/packages/core/src/index.gate.ts +++ b/packages/core/src/index.gate.ts @@ -2247,3 +2247,4 @@ export { evaluateTransitionInvariants, evaluateMergeBlockerPostcondition, evalua export { StaleBinarySchemaError, assertBinaryNotOlderThanDatabase } from "./postgres/schema-applier.js"; export { promoteResearchFinding } from "./research-feature-promotion.js"; export type { ResearchFeaturePromotionInput } from "./research-feature-promotion.js"; +export { ACTIVE_WORKFLOW_WORK_ITEM_STATES } from "./types.js"; diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index a8dc189868..63d430d569 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -2609,3 +2609,4 @@ export type { LanguageFamily, DetectedContentLanguage } from "./detect-content-l export { promoteResearchFinding } from "./research-feature-promotion.js"; export type { ResearchFeaturePromotionInput } from "./research-feature-promotion.js"; export { getTotalAgentActiveMs, startPlanningSegment, finalizePlanningSegment } from "./task-timing.js"; +export { ACTIVE_WORKFLOW_WORK_ITEM_STATES } from "./types.js"; diff --git a/packages/core/src/types.ts b/packages/core/src/types.ts index 1a8ef7b519..2d8d2bc99f 100644 --- a/packages/core/src/types.ts +++ b/packages/core/src/types.ts @@ -78,6 +78,7 @@ export type { ThinkingLevel, Column, ColumnId, TaskPriority }; import { MERGE_REQUEST_STATES, + ACTIVE_WORKFLOW_WORK_ITEM_STATES, WORKFLOW_WORK_ITEM_KINDS, WORKFLOW_WORK_ITEM_STATES, } from "./types/merge-queue.js"; @@ -101,6 +102,7 @@ import type { } from "./types/merge-queue.js"; export { MERGE_REQUEST_STATES, + ACTIVE_WORKFLOW_WORK_ITEM_STATES, WORKFLOW_WORK_ITEM_KINDS, WORKFLOW_WORK_ITEM_STATES, }; diff --git a/packages/core/src/types/merge-queue.ts b/packages/core/src/types/merge-queue.ts index b1838721d7..99d7ac77a3 100644 --- a/packages/core/src/types/merge-queue.ts +++ b/packages/core/src/types/merge-queue.ts @@ -43,6 +43,15 @@ export const WORKFLOW_WORK_ITEM_STATES = [ export type WorkflowWorkItemState = (typeof WORKFLOW_WORK_ITEM_STATES)[number]; +/** FNXC:WorkflowContinuations 2026-07-21-12:30: + * States that keep a workflow work item eligible for continuation ownership. */ +export const ACTIVE_WORKFLOW_WORK_ITEM_STATES: readonly WorkflowWorkItemState[] = [ + "runnable", + "running", + "held", + "retrying", +]; + export interface WorkflowWorkItem { id: string; runId: string; diff --git a/packages/core/src/workflow-ir.ts b/packages/core/src/workflow-ir.ts index 9e0ec3f2e4..4a7e8a1c2f 100644 --- a/packages/core/src/workflow-ir.ts +++ b/packages/core/src/workflow-ir.ts @@ -1553,6 +1553,7 @@ function validateColumnAgent(column: WorkflowIrColumn): void { function validateV2(ir: WorkflowIrV2): void { validateColumns(ir); + // FNXC:WorkflowValidation 2026-07-21-12:20: // Capacity holds must have somewhere the scheduler can actually release // them. Failing authoring here avoids durable continuations that can never // become runnable. diff --git a/packages/engine/src/__tests__/executor-fast-mode-workflows.test.ts b/packages/engine/src/__tests__/executor-fast-mode-workflows.test.ts index 812b93b723..ed19c4b387 100644 --- a/packages/engine/src/__tests__/executor-fast-mode-workflows.test.ts +++ b/packages/engine/src/__tests__/executor-fast-mode-workflows.test.ts @@ -10,6 +10,7 @@ import "./executor-test-helpers.js"; import { getBuiltinWorkflow } from "@fusion/core"; import { TaskExecutor } from "../executor.js"; import { WorkflowGraphTaskRunner } from "../workflow-graph-task-runner.js"; +import { FOREACH_ACTIVE_CONTEXT_KEY } from "../workflow-node-handlers.js"; import { createMockStore, mockedCreateFnAgent, @@ -187,6 +188,99 @@ describe("fast mode workflow/runtime invariants", () => { }); }); + it("does not project a fresh graph step or capture its baseline before the executor creates its worktree", async () => { + let liveTask = task({ + steps: [{ name: "Preflight", status: "pending" }], + worktree: undefined, + branch: undefined, + baseCommitSha: undefined, + }); + const store = createMockStore(); + store.getTask.mockImplementation(async () => liveTask); + const executor = new TaskExecutor(store, "/tmp/project-root"); + const runGraphTaskStep = vi.spyOn(executor as any, "runGraphTaskStep").mockImplementation(async () => { + expect(store.updateStep).not.toHaveBeenCalled(); + liveTask = { + ...liveTask, + worktree: "/tmp/project-root/.worktrees/fresh-step", + branch: "fusion/fn-6226", + baseCommitSha: "fresh-worktree-base", + steps: [{ name: "Preflight", status: "done" }], + }; + return { success: true }; + }); + + const result = await (executor as any) + .createAuthoritativeWorkflowPrimitives({ experimentalFeatures: { workflowGraphExecutor: true } }) + .runTaskStep( + { + run: { taskId: liveTask.id }, + node: { + node: { id: "steps#0:step-execute" }, + context: { + [FOREACH_ACTIVE_CONTEXT_KEY]: { + foreachNodeId: "steps", + stepIndex: 0, + instanceId: "steps#0", + }, + }, + }, + }, + liveTask, + 0, + ); + + expect(runGraphTaskStep).toHaveBeenCalledTimes(1); + expect(store.updateStep).not.toHaveBeenCalled(); + expect(result).toEqual({ + outcome: "success", + baselineSha: "fresh-worktree-base", + checkpointId: undefined, + }); + }); + + it("applies fresh-worktree step ordering through the legacy graph seam", async () => { + let liveTask = task({ + steps: [{ name: "Preflight", status: "pending" }], + worktree: undefined, + baseCommitSha: undefined, + }); + const store = createMockStore(); + store.getTask.mockImplementation(async () => liveTask); + const executor = new TaskExecutor(store, "/tmp/project-root"); + vi.spyOn(executor as any, "runGraphTaskStep").mockImplementation(async () => { + expect(store.updateStep).not.toHaveBeenCalled(); + liveTask = { + ...liveTask, + worktree: "/tmp/project-root/.worktrees/fresh-step", + baseCommitSha: "fresh-worktree-base", + steps: [{ name: "Preflight", status: "done" }], + }; + return { success: true }; + }); + const active = { + foreachNodeId: "steps", + stepIndex: 0, + instanceId: "steps#0", + }; + + const result = await executor.createAuthoritativeWorkflowSeams({} as any).stepExecute?.( + liveTask, + { [FOREACH_ACTIVE_CONTEXT_KEY]: active }, + ); + + expect(store.updateStep).not.toHaveBeenCalled(); + expect(result).toMatchObject({ + outcome: "success", + contextPatch: { + [FOREACH_ACTIVE_CONTEXT_KEY]: { + baselineSha: "fresh-worktree-base", + checkpointId: undefined, + }, + }, + }); + }); + it("fast builtin:coding still parses and executes steps while disabled optional groups stay inert", async () => { const calls: string[] = []; const prompt = "# Task\n\n## Steps\n\n### Step 1: Do the work\n- [ ] edit files"; diff --git a/packages/engine/src/__tests__/workflow-continuation-selection.test.ts b/packages/engine/src/__tests__/workflow-continuation-selection.test.ts new file mode 100644 index 0000000000..c2ece26d1b --- /dev/null +++ b/packages/engine/src/__tests__/workflow-continuation-selection.test.ts @@ -0,0 +1,29 @@ +import { describe, expect, it } from "vitest"; +import type { Task, WorkflowWorkItem } from "@fusion/core"; +import { selectActionablePlanningContinuations } from "../runtimes/in-process-runtime.js"; + +function workItem(id: string, waitReason: WorkflowWorkItem["waitReason"]): WorkflowWorkItem { + return { id, waitReason } as WorkflowWorkItem; +} + +function task(id: string, patch: Partial = {}): Task { + return { id, paused: false, userPaused: false, ...patch } as Task; +} + +describe("selectActionablePlanningContinuations", () => { + it("retains only planning items whose tasks are present and unpaused", () => { + const selected = selectActionablePlanningContinuations([ + { item: workItem("eligible", "planning"), task: task("T-1") }, + { item: workItem("capacity", "capacity"), task: task("T-2") }, + { item: workItem("missing", "planning"), task: undefined }, + { item: workItem("null-task", "planning"), task: null }, + { item: workItem("no-wait-reason", null), task: task("T-5") }, + { item: workItem("paused", "planning"), task: task("T-3", { paused: true }) }, + { item: workItem("user-paused", "planning"), task: task("T-4", { userPaused: true }) }, + ]); + + expect(selected.map(({ item, task: selectedTask }) => [item.id, selectedTask.id])).toEqual([ + ["eligible", "T-1"], + ]); + }); +}); diff --git a/packages/engine/src/executor.ts b/packages/engine/src/executor.ts index 9c0cd65e8d..92db3fa7b7 100644 --- a/packages/engine/src/executor.ts +++ b/packages/engine/src/executor.ts @@ -14,7 +14,7 @@ import { existsSync, lstatSync, realpathSync } from "node:fs"; import { readFile, rm, writeFile } from "node:fs/promises"; import type { TaskStore, Task, TaskDetail, TaskTokenUsage, StepStatus, Settings, WorkflowStep, MissionStore, AsyncMissionStore, Slice, AgentState, AgentCapability, RunMutationContext, AgentHeartbeatConfig, Agent, AgentMemoryInclusionMode, ProjectSettings, MergeResult, WorkflowIrNode, WorkflowIrNodeKind, WorkflowStepResult as CoreWorkflowStepResult, ThinkingLevel } from "@fusion/core"; import { getUnmetSchedulingDependencies } from "./scheduler.js"; -import { RetryStormError, serializeRetryStormError, evaluateCompletedPromotionFailureProvenance, evaluateSkipBypassTaint, resolveWorkflowIrForTask, resolveCompleteColumn, resolveMergeOrchestrationColumn, resolveReboundTarget, resolveColumnAgentBinding, resolveEffectiveAgent, instanceNodeId, getWorkflowExtensionRegistry, getBuiltinWorkflow, parseNoOpCompletionMarker, allowsAutoMergeProcessing, resolveEffectiveAutoMerge, isLiveSharedBranchGroupMemberIntegration, resolveMaxAutoMergeRetries, resolveMaxConsecutiveToolFailureRetries, resolveConsecutiveToolFailureRetryBackoffMs, resolveConsecutiveToolFailureThreshold, resolveExecutorEscalationTarget, resolveOptionalStepRevisionBudget, resolveOptionalReviewRevisionBudget, COMPLETION_SUMMARY_NODE_ID, upsertWorkflowStepResult, AWAITING_APPROVAL_PAUSE_REASON, THINKING_LEVELS, AgentStore, resolveExecutorFallbackModel } from "@fusion/core"; +import { RetryStormError, serializeRetryStormError, evaluateCompletedPromotionFailureProvenance, evaluateSkipBypassTaint, resolveWorkflowIrForTask, resolveCompleteColumn, resolveMergeOrchestrationColumn, resolveReboundTarget, resolveColumnAgentBinding, resolveEffectiveAgent, instanceNodeId, getWorkflowExtensionRegistry, getBuiltinWorkflow, parseNoOpCompletionMarker, allowsAutoMergeProcessing, resolveEffectiveAutoMerge, isLiveSharedBranchGroupMemberIntegration, resolveMaxAutoMergeRetries, resolveMaxConsecutiveToolFailureRetries, resolveConsecutiveToolFailureRetryBackoffMs, resolveConsecutiveToolFailureThreshold, resolveExecutorEscalationTarget, resolveOptionalStepRevisionBudget, resolveOptionalReviewRevisionBudget, COMPLETION_SUMMARY_NODE_ID, upsertWorkflowStepResult, AWAITING_APPROVAL_PAUSE_REASON, THINKING_LEVELS, ACTIVE_WORKFLOW_WORK_ITEM_STATES, AgentStore, resolveExecutorFallbackModel } from "@fusion/core"; import { finalizeProvenAutoMergeTask } from "./auto-merge-finalization.js"; import { mergeEffectiveSettings } from "./effective-settings.js"; import { generateFeatureVideo, type GenerateFeatureVideoOptions } from "./review-artifacts/feature-video.js"; @@ -191,7 +191,7 @@ import type { StuckTaskDetector, StuckTaskEvent } from "./stuck-task-detector.js import type { PluginRunner } from "./plugin-runner.js"; import { isContextLimitError } from "./context-limit-detector.js"; import { StepSessionExecutor } from "./step-session-executor.js"; -import { makeAncestryBlastRadiusGuard, resetStepToBaseline, runTaskStep } from "./step-runner.js"; +import { makeAncestryBlastRadiusGuard, resetStepToBaseline, runTaskStep, type RunTaskStepResult } from "./step-runner.js"; // FNXC:MergerUnification 2026-06-21-19:05: the foundation branch imported `acquireWorkspaceRepoWorktree` here but never used it in executor.ts (the agent tool wraps it via agent-tools.ts), which fails lint on the inherited base. Removed until master-plan U1 re-adds it together with its per-repo acquisition usage. import { acquireTaskWorktree, type AcquireTaskWorktreeResult } from "./worktree-acquisition.js"; import { resolveCapturedBaseCommitSha } from "./base-commit-capture.js"; @@ -5659,7 +5659,7 @@ export class TaskExecutor { const workItems = await this.store.listWorkflowWorkItemsForTask?.(task.id, { kinds: ["task"] }) ?? []; for (let index = workItems.length - 1; index >= 0; index -= 1) { const candidate = workItems[index]; - if (["held", "runnable", "running", "retrying"].includes(candidate.state)) { + if (ACTIVE_WORKFLOW_WORK_ITEM_STATES.includes(candidate.state)) { continuation = candidate; break; } @@ -5956,7 +5956,7 @@ export class TaskExecutor { clearPin: pinPersistence.clearPin, onSuspend: async (suspension) => { const items = await this.store.listWorkflowWorkItemsForTask(task.id, { kinds: ["task"] }); - const live = items.filter((item) => ["held", "runnable", "running", "retrying"].includes(item.state)); + const live = items.filter((item) => ACTIVE_WORKFLOW_WORK_ITEM_STATES.includes(item.state)); if (live.some((item) => item.nodeId === suspension.nodeId)) return; await this.store.replaceActiveTaskWorkflowContinuation({ runId: `${workflowRunId ?? `${task.id}:workflow`}:continuation:${suspension.nodeId}:${items.length}`, @@ -5996,7 +5996,7 @@ export class TaskExecutor { target: event.taskId, metadata: event.type === "task:column-transition" - ? { taskId: event.taskId, workflowId: event.workflowId, fromColumn: event.fromColumn, toColumn: event.toColumn, nodeId: event.nodeId } + ? { taskId: event.taskId, workflowId: event.workflowId, fromColumn: event.fromColumn, toColumn: event.toColumn, nodeId: event.nodeId, irHash: event.irHash } : { taskId: event.taskId, workflowId: event.workflowId, pinnedNodeId: event.pinnedNodeId, reason: event.reason }, }); }, @@ -6688,6 +6688,57 @@ export class TaskExecutor { return only; } + /** + * Project a graph-owned step only after it has a real worktree. + * + * A fresh task has no worktree until the authoritative implementation pass + * acquires one. Projecting before that pass produces a false "step started" + * event and captures the baseline from the project root. In that fresh path, + * let the implementation pass own the first projection and reuse the base SHA + * it captures during worktree acquisition. Resumed and isolated-step runs + * already have a worktree, so they keep the normal per-step projection and + * pre-work baseline behavior. + */ + private async runProjectedGraphTaskStep( + task: Task, + live: TaskDetail, + stepIndex: number, + active: ForeachActiveContext, + governingNodeId?: string, + thinkingLevel?: ThinkingLevel, + ): Promise { + const worktreePath = active.worktreePath || live.worktree; + const runStep = (idx: number) => + this.runGraphTaskStep( + task, + idx, + active.instanceId, + governingNodeId, + thinkingLevel, + ); + + if (!worktreePath) { + const result = await runStep(stepIndex); + const refreshed = await this.store.getTask(task.id).catch(() => live); + return { + outcome: result.success ? "success" : "failure", + baselineSha: refreshed.baseCommitSha, + checkpointId: undefined, + }; + } + + return runTaskStep( + { + store: this.store, + worktreePath, + runStep, + }, + { id: task.id, steps: live.steps }, + stepIndex, + { markDoneOnSuccess: active.deferDoneToReview !== true, projectionSource: "graph" }, + ); + } + /** Public authoritative-driver seam factory: exposes the same real lifecycle * seams the internal graph runner uses, without changing legacy behavior. */ public createAuthoritativeWorkflowPrimitives(settings: Settings): WorkflowRuntimePrimitives { @@ -6790,24 +6841,14 @@ export class TaskExecutor { data: { status: liveStatus }, }; } - const worktreePath = active.worktreePath || live.worktree || this.rootDir; this.graphStepActiveContext.set(this.graphActiveContextKey(task.id, active.instanceId), active); const stepGoverningNodeId = context[SEAM_GOVERNING_NODE_CONTEXT_KEY]; - return await runTaskStep( - { - store: this.store, - worktreePath, - runStep: (idx) => - this.runGraphTaskStep( - task, - idx, - active.instanceId, - typeof stepGoverningNodeId === "string" ? stepGoverningNodeId : undefined, - ), - }, - { id: task.id, steps: live.steps }, + return await this.runProjectedGraphTaskStep( + task, + live, stepIndex, - { markDoneOnSuccess: active.deferDoneToReview !== true, projectionSource: "graph" }, + active, + typeof stepGoverningNodeId === "string" ? stepGoverningNodeId : undefined, ); }, resetTaskStep: async (ctx, task, stepIndex, baselineSha, checkpointId) => { @@ -7390,7 +7431,6 @@ export class TaskExecutor { // worktree when the foreach allocated one; otherwise the task's main // worktree (shared isolation — unchanged). The file-scope guard the session // machinery installs applies to either worktree unchanged (not bypassed). - const worktreePath = active.worktreePath || live.worktree || this.rootDir; // Stamp the active instance so `runGraphTaskStep` can honor // `deferDoneToReview` when judging a non-terminal step (FIX 3). this.graphStepActiveContext.set(this.graphActiveContextKey(seamTask.id, active.instanceId), active); @@ -7405,35 +7445,15 @@ export class TaskExecutor { // foreach (overwrite mid-build, or clear while the shared pass is live). const stepGoverningNodeId = context[SEAM_GOVERNING_NODE_CONTEXT_KEY]; const seamThinkingLevel = context[SEAM_THINKING_LEVEL_CONTEXT_KEY]; - const result: Awaited> = await runTaskStep( - { - store: this.store, - worktreePath, - // U6/U8: graph-owned per-step physics. Per-step-review workflows - // pin StepSessionExecutor inside runGraphTaskStep; final-review coding - // honors runStepsInNewSessions and may reuse one executor session. - // Thread the instanceId so the active-context read is per-instance - // (parallel-foreach safe). - runStep: (stepIndex) => - this.runGraphTaskStep( - seamTask, - stepIndex, - active.instanceId, - typeof stepGoverningNodeId === "string" ? stepGoverningNodeId : undefined, - typeof seamThinkingLevel === "string" && WORKFLOW_THINKING_LEVEL_SET.has(seamThinkingLevel) - ? (seamThinkingLevel as ThinkingLevel) - : undefined, - ), - }, - { id: seamTask.id, steps: live.steps }, + const result = await this.runProjectedGraphTaskStep( + seamTask, + live, active.stepIndex, - { - // Single-authority done-marking (U6/KTD-4): when the foreach template - // has a step-review node, leave the step in-progress so the review's - // APPROVE marks it done (the review is the single done authority). - markDoneOnSuccess: active.deferDoneToReview !== true, - projectionSource: "graph", - }, + active, + typeof stepGoverningNodeId === "string" ? stepGoverningNodeId : undefined, + typeof seamThinkingLevel === "string" && WORKFLOW_THINKING_LEVEL_SET.has(seamThinkingLevel) + ? (seamThinkingLevel as ThinkingLevel) + : undefined, ); // Capture baseline/checkpoint back into the reserved active context so the // foreach sub-walk threads them to later template nodes (step-review/reset). diff --git a/packages/engine/src/hold-release.ts b/packages/engine/src/hold-release.ts index 31d5ef14f4..c723f5a843 100644 --- a/packages/engine/src/hold-release.ts +++ b/packages/engine/src/hold-release.ts @@ -42,6 +42,7 @@ import { resolveColumnFlags, resolveColumnAdjacency, PLAN_REVIEW_GROUP_ID, + ACTIVE_WORKFLOW_WORK_ITEM_STATES, DEFAULT_WORKFLOW_POOL_ID, TransitionRejectionError, resolveWorkflowIrForTask, @@ -142,6 +143,7 @@ function isHeldTask(ir: WorkflowIr, task: Task): boolean { * `promoteHeldTask`, `releaseHeldTaskByEvent`) enforces the same invariant. */ /** + * FNXC:PlanReview 2026-07-21-12:20: * Locate a Plan Review node placed before WIP. Disabled optional groups still * traverse this node, allowing the graph to persist the same generic capacity * continuation without invoking a reviewer. @@ -176,9 +178,7 @@ export async function isUnplannedForExecution(store: TaskStore, task: Task, ir: if (!legacyPassed) { if (typeof store.listWorkflowWorkItemsForTask !== "function") return true; const continuations = await store.listWorkflowWorkItemsForTask(task.id, { kinds: ["task"] }); - const active = continuations.filter((item) => - ["held", "runnable", "running", "retrying"].includes(item.state), - ); + const active = continuations.filter((item) => ACTIVE_WORKFLOW_WORK_ITEM_STATES.includes(item.state)); // Readiness is represented by the graph's durable boundary continuation, // not by a special-case review result. Optional groups that are disabled // are still traversed and therefore reach the same capacity boundary. diff --git a/packages/engine/src/runtimes/in-process-runtime.ts b/packages/engine/src/runtimes/in-process-runtime.ts index fe3458bcbb..1eb7a54a9a 100644 --- a/packages/engine/src/runtimes/in-process-runtime.ts +++ b/packages/engine/src/runtimes/in-process-runtime.ts @@ -14,11 +14,13 @@ import type { GithubIssueAction, CliSession, NotificationPayload, + WorkflowWorkItem, } from "@fusion/core"; import { AsyncCentralClaimStore, ChatStore, computeWorkflowIrPin, + ACTIVE_WORKFLOW_WORK_ITEM_STATES, isEphemeralAgent, resolveWorkflowIrForTask, } from "@fusion/core"; @@ -66,6 +68,26 @@ const yieldEventLoop = (): Promise => new Promise((resolve) => setImmediat export const CLI_AGENT_AWAITING_INPUT_EVENT = "cli-agent-awaiting-input" as const; const TASK_PLANNER_CHAT_AGENT_ID_PREFIX = "task-planner:"; +export interface PlanningContinuationCandidate { + item: WorkflowWorkItem; + task: Task | null | undefined; +} + +/** FNXC:WorkflowScheduling 2026-07-21-12:30: + * Select due planning continuations whose task remains dispatchable. */ +export function selectActionablePlanningContinuations( + candidates: readonly PlanningContinuationCandidate[], +): Array<{ item: WorkflowWorkItem; task: Task }> { + return candidates.filter( + (candidate): candidate is { item: WorkflowWorkItem; task: Task } => + candidate.item.waitReason === "planning" + && candidate.task !== null + && candidate.task !== undefined + && candidate.task.paused !== true + && candidate.task.userPaused !== true, + ); +} + export interface CliAgentAwaitingInputNotificationInfo { sessionId: string; notification: Record | undefined; @@ -996,8 +1018,14 @@ export class InProcessRuntime const planReview = resolvePreReleasePlanReviewNode(ir); if (!planReview || planReview.column !== live.column) return; + /* + FNXC:PlanReview 2026-07-21-12:20: + Specification completion creates a planning continuation only + when the review node belongs to the card's current column and no + active continuation already owns the task. + */ const active = await this.taskStore.listWorkflowWorkItemsForTask(live.id, { kinds: ["task"] }); - if (!active.some((item) => ["held", "runnable", "running", "retrying"].includes(item.state))) { + if (!active.some((item) => ACTIVE_WORKFLOW_WORK_ITEM_STATES.includes(item.state))) { await this.taskStore.replaceActiveTaskWorkflowContinuation({ runId: `${live.id}:planning-continuation:${planReview.id}:${active.length}`, taskId: live.id, @@ -1840,7 +1868,11 @@ export class InProcessRuntime return this.missionExecutionLoop; } - /** Wake the durable task-continuation consumer without nesting execution in triage. */ + /** + * FNXC:WorkflowScheduling 2026-07-21-12:20: + * Wake the durable task-continuation consumer in a microtask so triage can + * release its own execution slot before continuation dispatch begins. + */ private kickWorkflowContinuationProcessor(): void { queueMicrotask(() => { void this.drainWorkflowContinuations().catch((error) => { @@ -1850,6 +1882,11 @@ export class InProcessRuntime } private async drainWorkflowContinuations(): Promise { + /* + FNXC:WorkflowScheduling 2026-07-21-12:20: + A single runtime drain owns selection at a time. Concurrent wakeups collapse + behind this guard and the recurring processor supplies the next bounded pass. + */ if (this.workflowContinuationDrainActive || this.status !== "active") return; this.workflowContinuationDrainActive = true; try { @@ -1858,10 +1895,12 @@ export class InProcessRuntime states: ["runnable", "retrying"], limit: 20, }); + const candidates: PlanningContinuationCandidate[] = []; for (const item of items) { - if (item.waitReason !== "planning") continue; - const task = await this.taskStore.getTask(item.taskId).catch(() => undefined); - if (!task || task.paused || task.userPaused) continue; + const task = await this.taskStore.getTask(item.taskId); + candidates.push({ item, task }); + } + for (const { item, task } of selectActionablePlanningContinuations(candidates)) { void this.executor.execute(task).catch((error) => { runtimeLog.error(`Workflow continuation ${item.id} failed:`, error); });