diff --git a/docs/plans/workflow-owned-merge-stack/s08-workflow-owned-merge-processing.md b/docs/plans/workflow-owned-merge-stack/s08-workflow-owned-merge-processing.md new file mode 100644 index 0000000000..92fd4a4c04 --- /dev/null +++ b/docs/plans/workflow-owned-merge-stack/s08-workflow-owned-merge-processing.md @@ -0,0 +1,46 @@ +--- +title: "S08: workflow-owned merge queue processing" +type: refactor +status: draft-stack-handoff +date: 2026-06-09 +slice: S08 +milestone: "Gate B" +origin: docs/plans/2026-06-09-003-refactor-workflow-owned-merge-full-migration-slices-plan.md +stack_base: feature/workflow-owned-merge-s07-completion-handoff-merge-work +--- + +# S08: workflow-owned merge queue processing + +## Stack Role + +This draft PR reserves the S08 review slot in the workflow-owned merge, +retry, scheduling, and recovery migration stack. It is intentionally a handoff +artifact, not the completed implementation for this slice. + +## Milestone + +Gate B + +## Depends On + +S3 scheduler claim path, S6 merge capabilities, and S7 completion handoff. + +## Goal + +Process merge work items through workflow runtime instead of ProjectEngine's in-memory merge queue loop. + +## Expected File Scope + +packages/engine/src/project-engine.ts; packages/engine/src/scheduler.ts; packages/engine/src/merger.ts; packages/core/src/store.ts; merge lifecycle tests. + +## Expected Tests + +Serialized merge claim, successful finalize, transient retry, permanent conflict routing, duplicate lease blocking, hard cancel cancellation. + +## Exit Gate + +Production merge processing no longer depends on a hidden mergeQueue dequeue loop. + +## Full Plan + +See `docs/plans/2026-06-09-003-refactor-workflow-owned-merge-full-migration-slices-plan.md`. diff --git a/packages/engine/src/__tests__/workflow-work-engine-dispatch.test.ts b/packages/engine/src/__tests__/workflow-work-engine-dispatch.test.ts index 42cc9aa1c3..3f190d28f1 100644 --- a/packages/engine/src/__tests__/workflow-work-engine-dispatch.test.ts +++ b/packages/engine/src/__tests__/workflow-work-engine-dispatch.test.ts @@ -1,7 +1,12 @@ // @vitest-environment node -import { afterEach, describe, expect, it, vi } from "vitest"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { mkdtempSync } from "node:fs"; +import { rm } from "node:fs/promises"; +import { join } from "node:path"; +import { tmpdir } from "node:os"; import { + TaskStore, WORKFLOW_EXTENSION_SCHEMA_VERSION, __resetWorkflowExtensionRegistryForTests, getWorkflowExtensionRegistry, @@ -13,6 +18,9 @@ import { } from "@fusion/core"; import { TaskExecutor } from "../executor.js"; import { claimDueWorkflowWorkItem } from "../workflow-work-scheduler.js"; +import { processDueWorkflowWorkItem, workflowMergeWorkKinds } from "../workflow-work-processor.js"; +import { WorkflowTaskRuntime } from "../workflow-task-runtime.js"; +import type { WorkflowRuntimePrimitives } from "../runtime-primitives.js"; describe("workflow work-engine dispatch", () => { afterEach(() => { @@ -152,3 +160,72 @@ describe("workflow work scheduler claims", () => { expect(store.acquireWorkflowWorkItemLease).toHaveBeenCalledTimes(2); }); }); + +describe("workflow work processor", () => { + let rootDir: string; + let store: TaskStore; + + beforeEach(async () => { + rootDir = mkdtempSync(join(tmpdir(), "kb-workflow-work-processor-")); + store = new TaskStore(rootDir, join(rootDir, ".fusion-global")); + await store.init(); + }); + + afterEach(async () => { + store.close(); + await rm(rootDir, { recursive: true, force: true, maxRetries: 5, retryDelay: 50 }); + }); + + function primitives(): WorkflowRuntimePrimitives { + const success = async () => ({ outcome: "success" as const }); + return { + prepareWorktree: async () => ({ outcome: "success", data: { worktreePath: rootDir } }), + readArtifact: async () => undefined, + writeArtifact: async (_ctx, _task, key) => ({ outcome: "success", data: { key } }), + runPlanningSession: success, + runCodingSession: async () => ({ outcome: "success", data: { taskDone: true, modifiedFiles: [] } }), + runTaskStep: success, + resetTaskStep: async () => ({ ok: true }), + runReview: async () => ({ outcome: "success", data: { verdict: "APPROVE" } }), + runVerification: async () => ({ outcome: "success", data: { verdict: "skipped" } }), + runWorkflowStep: success, + updateSteps: async (_ctx, _task, steps) => ({ outcome: "success", data: { count: steps.length } }), + transitionTask: success, + requestMerge: async () => ({ outcome: "success", data: { status: "merged" } }), + abortRun: success, + audit: vi.fn(), + }; + } + + it("claims due merge work and runs it through workflow runtime", async () => { + const task = await store.createTask({ description: "processor task" }); + await store.moveTask(task.id, "todo"); + await store.moveTask(task.id, "in-progress"); + await store.handoffToReview(task.id, { + ownerAgentId: "agent-test", + evidence: { reason: "fn_task_done", runId: "run-processor", agentId: "agent-test" }, + now: "2026-06-09T00:00:00.000Z", + }); + const runtime = new WorkflowTaskRuntime({ + store, + primitives: primitives(), + runCustomNode: async () => ({ outcome: "success" }), + }); + + const result = await processDueWorkflowWorkItem(store, runtime, { experimentalFeatures: {} } as any, { + now: "2026-06-09T00:00:00.000Z", + leaseOwner: "processor-a", + leaseDurationMs: 60_000, + kinds: workflowMergeWorkKinds(), + }); + + expect(result).toMatchObject({ + claimed: true, + taskId: task.id, + runtime: { disposition: "completed" }, + }); + expect(store.listWorkflowWorkItemsForTask(task.id, { kinds: ["merge"] })).toEqual([ + expect.objectContaining({ state: "succeeded", leaseOwner: null, leaseExpiresAt: null }), + ]); + }); +}); diff --git a/packages/engine/src/index.ts b/packages/engine/src/index.ts index 83eec58a61..aad11857b8 100644 --- a/packages/engine/src/index.ts +++ b/packages/engine/src/index.ts @@ -152,6 +152,12 @@ export { runWorkflowMergeAttemptNode, type WorkflowMergeNodeDeps, } from "./workflow-merge-nodes.js"; +export { + processDueWorkflowWorkItem, + workflowMergeWorkKinds, + type WorkflowWorkProcessorOptions, + type WorkflowWorkProcessorResult, +} from "./workflow-work-processor.js"; export { MeshLeaseManager, type MeshLeaseManagerOptions, type LeaseRecoveryContext } from "./mesh-lease-manager.js"; export { MissionAutopilot, type MissionAutopilotOptions } from "./mission-autopilot.js"; export { MissionExecutionLoop, type MissionExecutionLoopOptions, type ValidationResult, loopLog } from "./mission-execution-loop.js"; diff --git a/packages/engine/src/workflow-work-processor.ts b/packages/engine/src/workflow-work-processor.ts new file mode 100644 index 0000000000..4103e99bc6 --- /dev/null +++ b/packages/engine/src/workflow-work-processor.ts @@ -0,0 +1,44 @@ +import type { Settings, WorkflowWorkItemKind } from "@fusion/core"; +import { claimDueWorkflowWorkItem, type WorkflowWorkSchedulerStore } from "./workflow-work-scheduler.js"; +import { WorkflowTaskRuntime, type WorkflowTaskRuntimeResult } from "./workflow-task-runtime.js"; + +export interface WorkflowWorkProcessorOptions { + leaseOwner: string; + leaseDurationMs: number; + now?: string; + kinds?: WorkflowWorkItemKind[]; +} + +export interface WorkflowWorkProcessorResult { + claimed: boolean; + workItemId?: string; + taskId?: string; + runtime?: WorkflowTaskRuntimeResult; +} + +export async function processDueWorkflowWorkItem( + store: WorkflowWorkSchedulerStore, + runtime: WorkflowTaskRuntime, + settings: (Pick & Partial) | undefined, + opts: WorkflowWorkProcessorOptions, +): Promise { + const dispatch = claimDueWorkflowWorkItem(store, { + now: opts.now, + leaseOwner: opts.leaseOwner, + leaseDurationMs: opts.leaseDurationMs, + kinds: opts.kinds, + }); + if (!dispatch) return { claimed: false }; + + const runtimeResult = await runtime.runWorkItem(dispatch.workItem, settings); + return { + claimed: true, + workItemId: dispatch.workItem.id, + taskId: dispatch.taskId, + runtime: runtimeResult, + }; +} + +export function workflowMergeWorkKinds(): WorkflowWorkItemKind[] { + return ["merge", "manual-hold"]; +}