From 0cb54dd8c4faf4dc9522e61dfc114a99582e0999 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Tue, 9 Jun 2026 11:45:29 -0700 Subject: [PATCH 1/5] refactor(workflow): map merge policy ownership --- ...kflow-owned-merge-retry-scheduling-plan.md | 279 ++++++++++++++++++ docs/workflow-policy-ownership-map.md | 78 +++++ .../workflow-policy-ownership-map.test.ts | 73 +++++ packages/engine/vitest.config.ts | 1 + 4 files changed, 431 insertions(+) create mode 100644 docs/plans/2026-06-09-002-refactor-workflow-owned-merge-retry-scheduling-plan.md create mode 100644 docs/workflow-policy-ownership-map.md create mode 100644 packages/engine/src/__tests__/workflow-policy-ownership-map.test.ts diff --git a/docs/plans/2026-06-09-002-refactor-workflow-owned-merge-retry-scheduling-plan.md b/docs/plans/2026-06-09-002-refactor-workflow-owned-merge-retry-scheduling-plan.md new file mode 100644 index 0000000000..bf6beaac8a --- /dev/null +++ b/docs/plans/2026-06-09-002-refactor-workflow-owned-merge-retry-scheduling-plan.md @@ -0,0 +1,279 @@ +--- +title: "refactor: Workflow-owned merge, retry, and scheduling policy" +type: refactor +status: active +date: 2026-06-09 +depth: deep +origin: none (solo planning bootstrap; focuses docs/plans/2026-06-09-001-refactor-big-bang-workflow-native-execution-plan.md on merge, retry, and scheduling ownership) +--- + +# refactor: Workflow-owned merge, retry, and scheduling policy + +## Summary + +Move Fusion's merge policy, retry policy, task scheduling decisions, and git operation ownership into workflow IR/runtime instead of keeping them as hidden engine behavior. The engine remains the substrate for durable storage, leases, capacity accounting, process supervision, timers, routing, and audit plumbing. Workflow nodes own the git and merge capability modules they invoke, including checkout preparation, branch integration, conflict handling, squash/finalize flows, retry routing, and manual holds. + +This plan does not require rewriting the existing merger algorithms up front. The first cut relocates the merger and git helpers behind workflow node capabilities while preserving the same guard rails. The shipped target is that production lifecycle and git/merge policy is authored in built-in workflow graphs and custom workflows, not in `ProjectEngine`, `Scheduler`, self-healing sweeps, or merge queue special cases. + +--- + +## Problem Frame + +Fusion now has a workflow runtime, but merge, retry, and scheduling still leak through engine-owned control paths: + +- The scheduler knows task lifecycle concepts and merge-specific eligibility instead of only claiming runnable workflow work. +- `ProjectEngine` and self-healing sweeps directly mutate task lifecycle or re-enqueue tasks for merge based on engine-side interpretations. +- The merge queue is a separate procedural control plane, so workflow state and merge state can disagree. +- Retry behavior is spread across retry helpers, task counters, rate-limit handling, manual retry reset, transient merge classification, and self-healing recovery. +- Dashboard badges can reflect stale engine classifications because the workflow run is not the single source of truth for waiting/retry/merge states. + +The desired architecture is simpler: a workflow run owns task policy and git/merge operation flow. The engine supplies reliable execution mechanics and non-bypassable guard services. + +--- + +## Requirements + +- R1. Built-in workflow IR expresses default merge, retry, and scheduling policy explicitly. +- R2. Custom workflows can model their own waiting, retry, git, and merge gates using the same runtime state model and guarded node capabilities. +- R3. The engine keeps only substrate responsibilities: durable queues/work items, leases, capacity limits, timers, process supervision, persistence, routing, storage, transition guard services, and audit plumbing. +- R4. The scheduler dispatches generic runnable workflow work items. It does not decide task lifecycle policy, merge eligibility, retry routing, or self-healing outcomes. +- R5. Merge work is represented as workflow work or workflow node state, not as an independent hidden merge queue with separate lifecycle semantics. +- R6. Retry budgets are scoped to workflow nodes/runs and surfaced through workflow runtime state. Legacy task-level retry fields may remain as compatibility summaries, not policy authority. +- R7. Self-healing and restart recovery emit typed workflow recovery events or wake workflow nodes. They do not directly requeue, fail, pause, unpause, or merge tasks except through guarded workflow primitives. +- R8. Existing invariants remain non-configurable: `autoMerge:false` is terminal-until-human-merged except the shared-branch member integration exception, `moveTask(in-progress -> todo)` is a hard cancel, file-scope/squash guards remain authoritative, branch-group target rules remain intact, and user pauses are respected. +- R9. Dashboard/API/CLI surfaces show workflow-native reasons for queued, blocked, retrying, merging, manually held, stalled, failed, and recovered states. +- R10. Tests assert the invariant across default coding, stepwise coding, custom workflows, PR workflows, branch groups, auto-merge off, manual retry, pause/cancel, restart recovery, transient merge errors, and stale work recovery. +- R11. Git operations are workflow node capabilities. Engine code may supervise child processes and provide guard services, but it must not own checkout, branch integration, conflict resolution, squash, finalize, or recovery policy. + +--- + +## Scope Boundaries + +### In Scope + +- Extract scheduler policy into workflow-owned runnable work item selection. +- Convert merge queue/merge request behavior into workflow run/node/work-item state. +- Add built-in merge subgraph nodes for merge gates, merge attempts, manual holds, retry branches, post-merge finalization, and recovery routing. +- Move retry budgets and retry-after decisions into workflow node policies. +- Convert self-healing from lifecycle mutation to workflow event publication and wakeup. +- Update dashboard/API/CLI state derivation to read workflow runtime state first. +- Move merger and git operation ownership into workflow node capability modules while preserving transition and repository safety guards as shared guard services. + +### Out of Scope + +- Rewriting the low-level git merge algorithm. +- Replacing SQLite or the task identity model. +- Removing branch groups, PR workflows, workflow steps, or custom workflow authoring. +- Redesigning the dashboard beyond the state display needed for workflow-native merge/retry/scheduling. +- Changing release mechanics. + +--- + +## Key Technical Decisions + +- KTD-1. Workflow owns policy and git operation flow; engine owns execution mechanics. A workflow node may run `prepareCheckout`, `integrateBranch`, `attemptMerge`, `finalizeSquash`, `runAgentSession`, or `scheduleRetry`, but the choice to call them and the route after success/failure belongs to workflow IR/runtime. +- KTD-2. Git and merge become workflow node capabilities. Existing files like `merger.ts`, `merger-ai.ts`, `merger-integration-worktree.ts`, `worktree-acquisition.ts`, and base-commit helpers should move behind workflow node capability modules rather than remain engine lifecycle primitives. +- KTD-3. Scheduling becomes generic runnable-work claiming. The scheduler reads workflow work items, leases one, checks capacity/routing, starts the workflow runtime, and records audit. It does not inspect task statuses to infer merge or retry policy. +- KTD-4. Retry state is node-scoped. Attempts, retry-after, transient/permanent classification, exhausted budgets, and manual retry resets live on workflow node/run/work-item state. Task-level retry summaries are projections. +- KTD-5. Recovery is event-driven. Restart recovery, stale detection, and self-healing write typed events such as `run-stale`, `merge-work-stale`, `agent-session-lost`, `retry-after-expired`, or `already-landed`; workflow recovery nodes consume them. +- KTD-6. Manual merge and `autoMerge:false` are workflow holds. The hold is visible, durable, and terminal until a human action releases or completes it. +- KTD-7. Branch groups are workflow subgraphs. Member-to-shared-branch integration and shared-branch-to-default promotion are separate merge nodes with separate auto-merge gates. +- KTD-8. Migration is staged internally, but the final shipped state has no production fallback where old engine merge/retry/scheduling policy can race the workflow runtime. +- KTD-9. Repository safety remains centralized as guard services. File-scope checks, squash overlap checks, worktree ownership checks, and branch target validation are not optional workflow author logic; nodes call guard services before mutating git state. + +--- + +## Target Architecture + +```mermaid +flowchart TB + Event[task event / timer / user action / recovery sweep] --> WorkItem[workflow work item] + WorkItem --> Scheduler[generic scheduler claim] + Scheduler --> Capacity[capacity + routing + lease] + Capacity --> Runtime[WorkflowTaskRuntime] + Runtime --> Graph[WorkflowGraphExecutor] + Graph --> Policy[workflow nodes own policy] + Policy --> MergeGate[merge/manual hold node] + Policy --> Retry[retry/backoff node policy] + Policy --> Recovery[recovery router node] + Policy --> Wait[hold/wait/capacity node] + MergeGate --> GitNodes[git/merge node capabilities] + Retry --> WorkItem + Recovery --> WorkItem + Wait --> WorkItem + GitNodes --> Guards[repository guard services] + GitNodes --> Store[(TaskStore / DB)] + GitNodes --> Git[git/worktrees] + GitNodes --> Audit[run audit] +``` + +The scheduler should be boring. The interesting state transitions are encoded in the workflow graph and persisted as workflow runtime state. + +--- + +## Implementation Units + +### U1. Inventory And Ownership Map + +- **Goal:** Produce a concrete source map of every merge, retry, and scheduling policy branch that must move. +- **Requirements:** R1, R3, R4, R5, R6, R7. +- **Files:** `packages/engine/src/project-engine.ts`, `packages/engine/src/scheduler.ts`, `packages/engine/src/self-healing.ts`, `packages/engine/src/merger.ts`, `packages/engine/src/group-merge-coordinator.ts`, `packages/engine/src/transient-merge-error-classifier.ts`, `packages/engine/src/retry-with-backoff.ts`, `packages/engine/src/rate-limit-retry.ts`, `packages/core/src/store.ts`, `packages/core/src/task-merge.ts`, `packages/core/src/retry-summary.ts`, `packages/core/src/manual-retry-reset.ts`, `docs/architecture.md`, `docs/workflow-steps.md`. +- **Approach:** Classify each branch as substrate, workflow policy, compatibility projection, or delete. Record non-bypassable guards separately from policy. This map becomes the checklist for later deletion gates. +- **Test scenarios:** Add a narrow characterization/search test or documented checklist that fails review if a known policy branch is left unclassified. +- **Verification:** Every existing merge queue recovery, self-healing merge requeue, scheduler retry, transient retry, manual retry, branch-group merge, and auto-merge branch has a target workflow node capability, workflow policy node, or shared guard service. + +### U2. Workflow Work Items For Scheduling, Merge, Retry, And Recovery + +- **Goal:** Add a durable workflow work-item model that represents runnable, held, retrying, merge, and recovery work generically. +- **Requirements:** R2, R4, R5, R6, R7, R9. +- **Files:** `packages/core/src/db.ts`, `packages/core/src/store.ts`, `packages/core/src/types.ts`, `packages/engine/src/workflow-task-runtime.ts`, `packages/engine/src/workflow-graph-executor.ts`, `packages/core/src/__tests__/central-db.test.ts`, `packages/core/src/__tests__/store-workflow-runtime.test.ts` (new), `packages/core/src/__tests__/merge-request-record.test.ts`. +- **Approach:** Introduce or consolidate a work-item table keyed by workflow run, task, node, and kind. Minimum fields should cover `kind`, `state`, `runId`, `taskId`, `nodeId`, `attempt`, `retryAfter`, `lease`, `lastError`, `blockedReason`, `createdAt`, and `updatedAt`. Existing merge request records can be migrated into or projected from this model during cutover. +- **Test scenarios:** A coding completion creates merge work; a transient merge error creates retrying merge work with `retryAfter`; `autoMerge:false` creates manual hold work; recovery events create recovery work; duplicate wakeups are idempotent; expired leases can be reclaimed; completed work cannot be re-enqueued by self-healing. +- **Verification:** Store tests prove the engine can find runnable work without inspecting task lifecycle policy. + +### U3. Generic Scheduler Substrate + +- **Goal:** Convert `Scheduler` into a generic workflow work dispatcher. +- **Requirements:** R3, R4, R8, R10. +- **Files:** `packages/engine/src/scheduler.ts`, `packages/engine/src/project-engine.ts`, `packages/engine/src/workflow-task-runtime.ts`, `packages/engine/src/workflow-authoritative-driver.ts`, `packages/engine/src/workflow-parity-observer.ts`, `packages/engine/src/__tests__/scheduler.test.ts`, `packages/engine/src/__tests__/workflow-work-engine-dispatch.test.ts` (new). +- **Approach:** Scheduler polling should select runnable workflow work items, apply capacity/agent routing/lease checks, and invoke the runtime. Remove special cases that decide whether `todo`, `in-progress`, `in-review`, merge-queued, retrying, or failed tasks should advance. Those decisions are represented by work-item state and workflow node outcomes. +- **Test scenarios:** Scheduler claims only due work items; capacity blocks produce held work state instead of task mutation; retry-after items are skipped until due; user-paused tasks are not claimed; hard-cancelled work is aborted and parked; engine restart reclaims stale leases; no merge-specific scheduler branch is needed. +- **Verification:** Scheduler tests use workflow work items as inputs and do not construct merge queue policy directly. + +### U4. Git And Merge Node Capabilities + +- **Goal:** Move checkout, branch integration, merge, squash, finalize, and conflict-handling operation ownership into explicit workflow node capability modules. +- **Requirements:** R1, R2, R5, R8, R10, R11. +- **Files:** `packages/engine/src/merger.ts`, `packages/engine/src/merger-ai.ts`, `packages/engine/src/merger-integration-worktree.ts`, `packages/engine/src/group-merge-coordinator.ts`, `packages/engine/src/merge-trait.ts`, `packages/engine/src/workflow-node-handlers.ts`, `packages/engine/src/workflow-merge-nodes.ts` (new), `packages/core/src/builtin-coding-workflow-ir.ts`, `packages/core/src/builtin-pr-workflow-ir.ts`, `packages/engine/src/__tests__/interpreter-merge-seam.test.ts`, `packages/engine/src/__tests__/dual-observe-merge-seam.test.ts`, branch-group merge tests. +- **Approach:** Define node capabilities for checkout preparation, base/fork-point capture, branch integration, merge eligibility, manual hold, merge attempt, conflict/revision routing, post-squash audit, finalize, and already-landed recovery. Move git operation orchestration out of engine lifecycle modules and into these capabilities. The capability modules call shared guard services for file-scope checks, squash checks, worktree ownership, branch target validation, and audit correlation before mutating repository state. +- **Test scenarios:** Completed implementation enters merge node; checkout preparation is driven by a workflow node; `autoMerge:false` routes to manual hold; file-scope violation fails through workflow outcome; already-on-main routes to finalize; transient merge failure routes to retry; non-transient conflict routes to revision or manual hold; branch-group member integration uses member auto-merge; group promotion uses group auto-merge; no engine lifecycle loop can perform branch integration directly. +- **Verification:** No production caller starts a merge attempt except a workflow merge node or an explicit human/manual API that records the equivalent workflow event. + +### U5. Workflow-Owned Retry Policies + +- **Goal:** Move retry attempts, budgets, backoff, manual retry reset, and retry exhaustion into workflow runtime state. +- **Requirements:** R2, R6, R7, R9, R10. +- **Files:** `packages/engine/src/workflow-graph-executor.ts`, `packages/engine/src/workflow-node-handlers.ts`, `packages/engine/src/retry-with-backoff.ts`, `packages/engine/src/rate-limit-retry.ts`, `packages/engine/src/transient-merge-error-classifier.ts`, `packages/core/src/retry-summary.ts`, `packages/core/src/manual-retry-reset.ts`, `packages/engine/src/__tests__/workflow-graph-executor-retry-coding-workflow.test.ts`, `packages/engine/src/__tests__/workflow-graph-step-rerun.test.ts`, `packages/engine/src/__tests__/executor-retry-storm.test.ts`, `packages/engine/src/__tests__/workflow-node-retry-policy.test.ts` (new). +- **Approach:** Attach retry policy to node definitions or built-in node configs. Persist attempt count, last error, classification, retry-after, and exhaustion on workflow node/work-item records. Manual retry clears the relevant node/run retry state and emits a workflow wake event. Keep task-level retry summary as a derived dashboard field. +- **Test scenarios:** Coding node transient failure retries within budget; merge transient failure retries merge node only; rate-limit retry uses due time; exhausted retry routes to workflow failure/manual hold; manual retry resets exactly the failed node; retry state survives engine restart; retry storm protection remains enforced by runtime substrate. +- **Verification:** Tests prove that no retry branch is controlled solely by task status/counters. + +### U6. Self-Healing As Workflow Events + +- **Goal:** Remove lifecycle-mutating self-healing decisions and replace them with typed workflow recovery events. +- **Requirements:** R7, R8, R9, R10. +- **Files:** `packages/engine/src/self-healing.ts`, `packages/engine/src/restart-recovery-coordinator.ts`, `packages/engine/src/recovery-policy.ts`, `packages/engine/src/workflow-task-runtime.ts`, `packages/engine/src/workflow-node-handlers.ts`, `packages/engine/src/__tests__/self-healing.test.ts`, `packages/engine/src/__tests__/reliability-interaction-backstops.test.ts`, `packages/engine/src/__tests__/workflow-recovery-events.test.ts` (new). +- **Approach:** Sweeps detect facts and publish events. Recovery nodes decide routes. Example facts: stale lease, missing session, merge work stale, already landed, no active work item, cancelled worktree, manual hold still valid, auto-merge disabled. Self-healing should no-op when workflow state already explains the task. +- **Test scenarios:** In-review merge work is not re-enqueued repeatedly; `autoMerge:false` in-review tasks remain terminal; stale running node wakes recovery node; already landed task finalizes; user-paused task is not mutated; duplicate recovery events are deduped; completed/held workflow work is not marked stalled. +- **Verification:** Existing false recovery strings like "Auto-recovered: eligible in-review task re-enqueued for merge" are replaced by workflow event audit when applicable and disappear for valid held/queued states. + +### U7. Built-In Workflow Migration + +- **Goal:** Encode default coding, stepwise coding, and PR workflows with explicit scheduling, retry, merge, and recovery regions. +- **Requirements:** R1, R2, R8, R10. +- **Files:** `packages/core/src/builtin-coding-workflow-ir.ts`, `packages/core/src/builtin-stepwise-coding-workflow-ir.ts`, `packages/core/src/builtin-pr-workflow-ir.ts`, `packages/core/src/builtin-workflows.ts`, `packages/core/src/workflow-ir-types.ts`, `packages/core/src/__tests__/builtin-coding-workflow-ir.test.ts`, `packages/core/src/__tests__/builtin-stepwise-coding-workflow-ir.test.ts`, `packages/core/src/__tests__/builtin-pr-workflow-ir.test.ts`. +- **Approach:** Add explicit graph regions for queue/hold, implementation retry, review, merge gate, merge retry, manual hold, branch-group integration, finalization, and recovery. Keep compatibility with existing workflow column/trait semantics. +- **Test scenarios:** Built-in workflows validate; every legacy lifecycle phase has a node; fast execution mode still preserves required post-merge checks; PR response workflow routes review/fix/merge correctly; stepwise workflow retries per-step without retrying the entire task when possible. +- **Verification:** Built-in workflow fixtures become the source of truth for default lifecycle behavior. + +### U8. Branch Group And Shared Branch Workflows + +- **Goal:** Model branch-group member integration and group promotion as workflow-owned merge subgraphs. +- **Requirements:** R5, R7, R8, R10. +- **Files:** `packages/engine/src/group-merge-coordinator.ts`, `packages/engine/src/merge-trait.ts`, `packages/engine/src/merger-integration-worktree.ts`, `packages/core/src/builtin-coding-workflow-ir.ts`, branch-group tests under `packages/engine/src/__tests__/`. +- **Approach:** Split merge target resolution into workflow node capability calls guarded by repository safety services, then let workflow nodes route member-to-group and group-to-default promotion. Preserve the scoped `autoMerge:false` exception for shared-branch group members. +- **Test scenarios:** Shared member integrates to group branch while global auto-merge is off; group promotion remains blocked when group/global auto-merge is off; conflicting member integration routes to recovery/revision; final group promotion uses file-scope and squash guards; group merge work is visible as workflow work. +- **Verification:** No branch-group coordinator loop owns task lifecycle independent of workflow runtime. + +### U9. Dashboard, API, And CLI State Projection + +- **Goal:** Show workflow-native merge, retry, waiting, and stalled reasons everywhere users inspect tasks. +- **Requirements:** R6, R7, R9. +- **Files:** `packages/dashboard/app/components/TaskCard.tsx`, task detail components, reliability views, task API routes, CLI task output files, `packages/core/src/retry-summary.ts`, `packages/core/src/task-merge.ts`, dashboard tests for task cards/reliability. +- **Approach:** Derive badges and details from workflow run/work-item state first. Keep compatibility projections for older rows during migration, but do not let stale merge queue classifications override valid workflow holds or queued merge work. +- **Test scenarios:** Merge queued shows queued/merge work state, not stalled; retrying shows attempt and retry-after; manual hold shows human action required; recovery event shows event reason; completed work hides stale stalled badges; branch-group merge work identifies target branch. +- **Verification:** UI tests cover task card and detail surfaces for queued, retrying, manual hold, merge failed, and recovered states. + +### U10. Cutover And Deletion Gates + +- **Goal:** Remove production engine-owned policy paths after workflow parity is proven. +- **Requirements:** R3, R4, R5, R6, R7, R10. +- **Files:** `packages/engine/src/project-engine.ts`, `packages/engine/src/scheduler.ts`, `packages/engine/src/self-healing.ts`, `packages/engine/src/merger.ts`, `packages/core/src/store.ts`, focused search tests under `packages/engine/src/__tests__/`. +- **Approach:** Delete or demote engine branches that directly requeue for merge, classify merge lifecycle, schedule task lifecycle by status, mutate retry state, or self-heal by setting terminal statuses. Add search/structure tests for forbidden production patterns where practical. +- **Test scenarios:** Search tests fail on direct merge queue lifecycle mutation from self-healing; scheduler tests fail if task-status policy reappears; merge attempts require workflow node context; retry writes require workflow run/node id; manual retry emits workflow wake event. +- **Verification:** Final branch has one lifecycle control plane: workflow runtime. + +### U11. Documentation And Release Notes + +- **Goal:** Update architecture and user-facing docs to match the new ownership model. +- **Requirements:** R1, R3, R9. +- **Files:** `docs/architecture.md`, `docs/workflow-steps.md`, `docs/dashboard-guide.md`, `docs/settings-reference.md`, `CONCEPTS.md`, `.changeset/.md`. +- **Approach:** Document the workflow/substrate boundary, workflow work items, merge nodes, retry policy, recovery events, dashboard state meanings, and compatibility projections. Add a patch changeset if behavior changes affect published `@runfusion/fusion`. +- **Verification:** Docs mention the same state names used in API/UI tests. + +--- + +## Acceptance Examples + +- AE1. A task finishing implementation creates/continues a workflow merge node. No engine merge queue loop independently decides to merge it. +- AE2. A transient merge failure records retry state on the merge node/work item with `retryAfter`; the scheduler only wakes it when due. +- AE3. `autoMerge:false` routes the task to a visible manual merge hold and self-healing leaves it there. +- AE4. A valid queued merge item is never repeatedly marked "Auto-recovered: eligible in-review task re-enqueued for merge." +- AE5. Moving an active task from `in-progress` to `todo` aborts active workflow work and parks the workflow according to hard-cancel semantics. +- AE6. Branch-group member integration can run while shared-branch assembly is allowed, but shared-branch promotion to default remains gated by group/global auto-merge. +- AE7. A manual retry clears the failed workflow node's retry state and creates a due work item without resetting unrelated workflow progress. +- AE8. Dashboard cards and detail views show queued, retrying, manual hold, failed, and recovery states from workflow runtime state. + +--- + +## Rollout Sequence + +1. **Characterize:** Land U1 and focused tests around current merge/retry/scheduler behavior. +2. **State model:** Land U2 without changing production routing; project existing merge request state into workflow work items. +3. **Generic dispatch:** Convert scheduler to claim workflow work items while still producing equivalent behavior. +4. **Git/merge nodes:** Route built-in checkout, branch integration, merge, squash, and finalize behavior through workflow node capabilities. +5. **Retry nodes:** Move retry budgets and manual retry reset to workflow node/run state. +6. **Recovery events:** Convert self-healing to facts/events plus workflow recovery nodes. +7. **Dashboard projection:** Switch UI/API/CLI to workflow-native state. +8. **Deletion gate:** Remove old engine-owned merge/retry/scheduling policy paths and add regression/search tests. +9. **Docs and changeset:** Update docs and add a changeset if published behavior changed. + +--- + +## Risks And Mitigations + +- **Hidden merger invariants:** Start by moving existing merger code behind workflow node capabilities and characterize behavior before routing changes. +- **Queue starvation:** Make workflow work-item selection explicit and test retry-after, capacity, and lease ordering. +- **Double execution during migration:** Use idempotent work-item creation and explicit final deletion gates; do not leave two production controllers active. +- **Custom workflow expressiveness gaps:** Add built-in node kinds for non-authorable primitives rather than forcing users to script around core merge/retry mechanics. +- **State table bloat:** Add retention/pruning rules for terminal workflow work items while preserving audit. +- **Manual hold confusion:** Surface hold reason and release action consistently in dashboard/API/CLI. +- **Branch-group regressions:** Treat member integration and group promotion as separate acceptance surfaces with separate auto-merge tests. + +--- + +## Verification Plan + +- `pnpm test:gate` +- `pnpm lint` +- `pnpm build` +- Targeted engine/core suites for scheduler, workflow runtime, workflow graph executor, merge nodes, branch groups, self-healing, retry policies, manual retry reset, and dashboard task state projection. +- Manual verification with a local Fusion project: + - one ordinary auto-merge task, + - one `autoMerge:false` task, + - one transient merge failure, + - one manual retry, + - one branch-group member integration, + - one engine restart during queued merge work. + +--- + +## Done Criteria + +- Built-in workflows explicitly model git, merge, retry, and scheduling policy. +- Scheduler is a generic workflow work dispatcher. +- Checkout preparation, branch integration, merge attempts, squash, and finalize operations are invoked by workflow nodes or explicit human actions recorded as workflow events. +- Retry state is node/run scoped and visible through workflow runtime state. +- Self-healing emits workflow recovery events instead of directly mutating lifecycle. +- UI/API/CLI state projections come from workflow state, with compatibility fallback only for old rows. +- Production engine code no longer contains independent git/merge/retry/scheduling policy paths that can race workflow runtime. diff --git a/docs/workflow-policy-ownership-map.md b/docs/workflow-policy-ownership-map.md new file mode 100644 index 0000000000..ada7625224 --- /dev/null +++ b/docs/workflow-policy-ownership-map.md @@ -0,0 +1,78 @@ +# Workflow Policy Ownership Map + +## Purpose + +This map is the U1 characterization artifact for moving merge, retry, scheduling, +and recovery policy into workflow IR/runtime. It classifies current production +branches before code is deleted or moved so later cutover work can prove that no +legacy engine control path was left unowned. + +## Ownership Categories + +- `substrate`: engine/core mechanics that remain below workflow policy. +- `workflow-policy`: decisions that must be represented by workflow nodes, + workflow node state, or workflow recovery events. +- `capability`: operations invoked by workflow nodes while still using shared + guard services. +- `compat-projection`: legacy task fields or records that may remain as + derived summaries during migration. +- `delete-after-cutover`: branches that should disappear once workflow parity is + authoritative. + +## Catalog + +| Surface | Current source | Current owner | Target owner | Disposition | +|---|---|---|---|---| +| Auto-merge queue enqueue and dequeue | `packages/engine/src/project-engine.ts` | `ProjectEngine` merge queue | workflow merge work items and merge-gate nodes | `workflow-policy`, `delete-after-cutover` | +| In-review handoff delay and startup sweep | `packages/engine/src/project-engine.ts` | `task:moved` listener plus in-review scan | workflow completion handoff node creates merge work | `workflow-policy` | +| Manual `onMerge` requests | `packages/engine/src/project-engine.ts` | engine public merge queue entry point | explicit human/manual workflow event that wakes merge node | `workflow-policy`, `capability` | +| Merge request shadow contract | `packages/core/src/store.ts`, `packages/engine/src/project-engine.ts`, `packages/engine/src/merger.ts` | store record plus shadow parity branches | workflow work-item state or compatibility projection | `compat-projection` | +| Merge checkout, integration, conflict resolution, squash, finalize | `packages/engine/src/merger.ts`, `packages/engine/src/merger-ai.ts`, `packages/engine/src/merger-integration-worktree.ts` | merger lifecycle procedures | workflow merge node capabilities calling guard services | `capability` | +| Branch-group member integration and group promotion | `packages/engine/src/group-merge-coordinator.ts`, `packages/engine/src/merge-trait.ts`, `packages/engine/src/merger-integration-worktree.ts` | group coordinator and merger helpers | branch-group workflow subgraph with separate member and promotion nodes | `workflow-policy`, `capability` | +| Merge target and auto-merge eligibility guards | `packages/core/src/task-merge.ts` | shared helper used by engine paths | shared guard service called by workflow nodes | `substrate` | +| Dependency satisfaction treats `in-review` as satisfied | `packages/engine/src/scheduler.ts`, `packages/core/src/task-merge.ts` | scheduler/task helper lifecycle interpretation | workflow completion handoff state and compatibility projection | `workflow-policy`, `compat-projection` | +| Active scope leases include unmerged `in-review` worktrees | `packages/engine/src/scheduler.ts` | scheduler overlap policy | workflow work leases plus repository guard services | `workflow-policy`, `substrate` | +| PR monitor starts/stops from `in-review` transitions | `packages/engine/src/scheduler.ts` | scheduler task-move listener | workflow PR/watch nodes or workflow events | `workflow-policy` | +| Generic agent capacity, routing, claim, and lease mechanics | `packages/engine/src/scheduler.ts` | scheduler | scheduler substrate claiming runnable workflow work | `substrate` | +| Executor retry storm cap | `packages/engine/src/__tests__/executor-retry-storm.test.ts`, `packages/engine/src/project-engine.ts` | engine retry counters and execution loop | workflow node retry policy plus runtime substrate guard | `workflow-policy`, `substrate` | +| Generic backoff helpers | `packages/engine/src/retry-with-backoff.ts`, `packages/engine/src/rate-limit-retry.ts` | helper functions | reusable substrate helper called by retry nodes | `substrate` | +| Transient merge error classification | `packages/engine/src/transient-merge-error-classifier.ts` | helper used by merger/self-healing | merge-node classification input, not route owner | `substrate` | +| Task-level retry summary fields | `packages/core/src/retry-summary.ts`, `packages/core/src/manual-retry-reset.ts` | task metadata and reset patch | compatibility projection from workflow node/run retry state | `compat-projection` | +| Manual retry reset | `packages/core/src/manual-retry-reset.ts`, dashboard/API callers | task metadata patch | workflow event clearing targeted failed node retry state | `workflow-policy`, `compat-projection` | +| Recover mergeable in-review tasks | `packages/engine/src/self-healing.ts` | self-healing directly re-enqueues merge | workflow recovery event wakes merge node | `workflow-policy`, `delete-after-cutover` | +| Completion handoff limbo recovery | `packages/engine/src/self-healing.ts` | self-healing re-emits auto-merge handoff | workflow recovery event or idempotent handoff node wake | `workflow-policy` | +| Transient merge failure recovery | `packages/engine/src/self-healing.ts` | self-healing resets merge retries and re-enqueues | merge-node retry policy and retry-after work item | `workflow-policy`, `delete-after-cutover` | +| Stale merge status recovery | `packages/engine/src/self-healing.ts` | self-healing clears status and may enqueue merge | workflow recovery event plus merge work reconciliation | `workflow-policy` | +| Already-landed and no-op finalization | `packages/engine/src/self-healing.ts`, `packages/engine/src/merger.ts` | self-healing/merger lifecycle paths | workflow recovery/finalize nodes with repository guard services | `workflow-policy`, `capability` | +| Backward in-review recovery paths | `packages/engine/src/self-healing.ts`, `docs/self-healing-backward-move-audit.md` | proof-gated self-healing mutations | workflow recovery nodes; engine only emits facts | `workflow-policy`, `delete-after-cutover` | +| Workflow runtime execution facade | `packages/engine/src/workflow-task-runtime.ts`, `packages/engine/src/workflow-graph-executor.ts` | runtime executes graph nodes | remains workflow runtime owner | `substrate`, `workflow-policy` | +| Built-in default workflow definitions | `packages/core/src/builtin-coding-workflow-ir.ts`, `packages/core/src/builtin-stepwise-coding-workflow-ir.ts`, `packages/core/src/builtin-pr-workflow-ir.ts` | partial lifecycle expression | authoritative source for default scheduling, retry, merge, and recovery regions | `workflow-policy` | +| Dashboard task-card merge/retry/stall badges | `packages/dashboard/app/components/TaskCard.tsx` | task fields and legacy classifications | workflow run/work-item projection first, legacy fields second | `compat-projection` | +| Reliability and diagnostics surfaces | `docs/diagnostics.md`, dashboard reliability views | self-healing and engine status strings | workflow-native recovery and held-work reasons | `compat-projection` | + +## Non-Bypassable Guard Services + +These remain centralized and are called by workflow node capabilities before +mutating git state: + +- File-scope and squash overlap checks. +- Branch target and branch-group target validation. +- Worktree ownership and lease checks. +- Auto-merge processing gate, including `autoMerge:false` terminal-until-human + semantics and the shared-branch member integration exception. +- Run-audit correlation for git operations and recovery facts. + +## Deletion Gates + +- No production caller may start checkout, branch integration, squash, or finalize + except a workflow merge node or an explicit human/manual API that records an + equivalent workflow event. +- `Scheduler` may claim runnable workflow work, apply capacity/routing/leases, + and monitor PR/watch substrate events; it must not infer merge eligibility, + retry routing, or task lifecycle advancement from task columns. +- `SelfHealingManager` may publish typed recovery facts and reconcile metadata; + it must not directly requeue, pause, fail, unpause, or move merge/retry tasks + except through guarded workflow primitives. +- Task-level retry and merge fields are compatibility summaries. Workflow + run/node/work-item state is the policy authority. + diff --git a/packages/engine/src/__tests__/workflow-policy-ownership-map.test.ts b/packages/engine/src/__tests__/workflow-policy-ownership-map.test.ts new file mode 100644 index 0000000000..00e3e7622d --- /dev/null +++ b/packages/engine/src/__tests__/workflow-policy-ownership-map.test.ts @@ -0,0 +1,73 @@ +import { readFileSync } from "node:fs"; +import { resolve } from "node:path"; +import { describe, expect, it } from "vitest"; + +const DOC_PATH = resolve(process.cwd(), "../../docs/workflow-policy-ownership-map.md"); + +const REQUIRED_SOURCE_FILES = [ + "packages/engine/src/project-engine.ts", + "packages/engine/src/scheduler.ts", + "packages/engine/src/self-healing.ts", + "packages/engine/src/merger.ts", + "packages/engine/src/merger-ai.ts", + "packages/engine/src/merger-integration-worktree.ts", + "packages/engine/src/group-merge-coordinator.ts", + "packages/engine/src/transient-merge-error-classifier.ts", + "packages/engine/src/retry-with-backoff.ts", + "packages/engine/src/rate-limit-retry.ts", + "packages/core/src/store.ts", + "packages/core/src/task-merge.ts", + "packages/core/src/retry-summary.ts", + "packages/core/src/manual-retry-reset.ts", + "packages/core/src/builtin-coding-workflow-ir.ts", + "packages/core/src/builtin-stepwise-coding-workflow-ir.ts", + "packages/core/src/builtin-pr-workflow-ir.ts", + "packages/dashboard/app/components/TaskCard.tsx", +] as const; + +const REQUIRED_POLICY_SURFACES = [ + "Auto-merge queue enqueue and dequeue", + "Merge checkout, integration, conflict resolution, squash, finalize", + "Branch-group member integration and group promotion", + "Dependency satisfaction treats `in-review` as satisfied", + "Active scope leases include unmerged `in-review` worktrees", + "Manual retry reset", + "Recover mergeable in-review tasks", + "Completion handoff limbo recovery", + "Transient merge failure recovery", + "Already-landed and no-op finalization", + "Built-in default workflow definitions", + "Dashboard task-card merge/retry/stall badges", +] as const; + +describe("workflow policy ownership map", () => { + const doc = readFileSync(DOC_PATH, "utf-8"); + + it("classifies every required policy surface from the workflow-owned merge plan", () => { + for (const surface of REQUIRED_POLICY_SURFACES) { + expect(doc, `missing ownership surface: ${surface}`).toContain(surface); + } + }); + + it("anchors the map to the production source files that own merge, retry, scheduling, and projection today", () => { + for (const file of REQUIRED_SOURCE_FILES) { + expect(doc, `missing source file: ${file}`).toContain(file); + } + }); + + it("records the migration dispositions needed for later deletion gates", () => { + for (const disposition of [ + "`substrate`", + "`workflow-policy`", + "`capability`", + "`compat-projection`", + "`delete-after-cutover`", + ]) { + expect(doc).toContain(disposition); + } + + expect(doc).toContain("## Deletion Gates"); + expect(doc).toContain("No production caller may start checkout, branch integration, squash, or finalize"); + expect(doc).toContain("Task-level retry and merge fields are compatibility summaries"); + }); +}); diff --git a/packages/engine/vitest.config.ts b/packages/engine/vitest.config.ts index 3478d16eda..81afd0a195 100644 --- a/packages/engine/vitest.config.ts +++ b/packages/engine/vitest.config.ts @@ -84,6 +84,7 @@ export default defineConfig({ "src/__tests__/self-healing.test.ts", "src/__tests__/heartbeat-monitor.test.ts", "src/__tests__/workflow-node-handlers.test.ts", + "src/__tests__/workflow-policy-ownership-map.test.ts", ], exclude: ["node_modules/**", "dist/**"], }, From f16b038fab02387b24fb39119ee2d7ab9d823208 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Tue, 9 Jun 2026 12:41:29 -0700 Subject: [PATCH 2/5] refactor(workflow): add workflow work-item store slice --- .changeset/workflow-work-items.md | 5 + ...-owned-merge-full-migration-slices-plan.md | 537 ++++++++++++++++++ .../core/src/__tests__/db-migrate.test.ts | 50 +- packages/core/src/__tests__/db.test.ts | 84 +-- .../core/src/__tests__/goals-schema.test.ts | 2 +- .../core/src/__tests__/insight-store.test.ts | 10 +- .../__tests__/merge-request-record.test.ts | 2 +- .../core/src/__tests__/mission-store.test.ts | 2 +- packages/core/src/__tests__/run-audit.test.ts | 2 +- .../src/__tests__/store-merge-queue.test.ts | 2 +- .../__tests__/store-workflow-runtime.test.ts | 232 ++++++++ .../core/src/__tests__/task-documents.test.ts | 2 +- packages/core/src/db.ts | 57 +- packages/core/src/index.ts | 4 +- packages/core/src/store.ts | 256 ++++++++- packages/core/src/types.ts | 74 +++ 16 files changed, 1239 insertions(+), 82 deletions(-) create mode 100644 .changeset/workflow-work-items.md create mode 100644 docs/plans/2026-06-09-003-refactor-workflow-owned-merge-full-migration-slices-plan.md create mode 100644 packages/core/src/__tests__/store-workflow-runtime.test.ts diff --git a/.changeset/workflow-work-items.md b/.changeset/workflow-work-items.md new file mode 100644 index 0000000000..bfa8b27ff1 --- /dev/null +++ b/.changeset/workflow-work-items.md @@ -0,0 +1,5 @@ +--- +"@runfusion/fusion": patch +--- + +Add workflow work-item storage primitives for workflow-owned merge migration. diff --git a/docs/plans/2026-06-09-003-refactor-workflow-owned-merge-full-migration-slices-plan.md b/docs/plans/2026-06-09-003-refactor-workflow-owned-merge-full-migration-slices-plan.md new file mode 100644 index 0000000000..50254bac85 --- /dev/null +++ b/docs/plans/2026-06-09-003-refactor-workflow-owned-merge-full-migration-slices-plan.md @@ -0,0 +1,537 @@ +--- +title: "refactor: Workflow-owned merge full migration slices" +type: refactor +status: active +date: 2026-06-09 +depth: deep +origin: docs/plans/2026-06-09-002-refactor-workflow-owned-merge-retry-scheduling-plan.md +--- + +# refactor: Workflow-owned merge full migration slices + +## Summary + +This plan turns the workflow-owned merge/retry/scheduling architecture into a +sequence of PR-sized migration slices. The target state is unchanged from the +origin plan: workflow IR/runtime owns merge policy, retry policy, scheduling +policy, recovery routing, and git operation flow; the engine keeps substrate +responsibilities such as storage, leases, timers, process supervision, routing, +capacity, guard services, and audit plumbing. + +The migration should ship in independently reviewable slices, but the final +cutover must not leave two production control planes. Compatibility projections +are allowed while slices are in flight. Production fallback paths are removed at +the deletion gates. + +## Requirements Trace + +- R1. Built-in workflow IR expresses default merge, retry, scheduling, and + recovery policy explicitly. +- R2. Workflow work state replaces hidden merge queue and retry routing as the + policy authority. +- R3. Scheduler claims generic workflow work; it does not infer task lifecycle + advancement from columns. +- R4. Git and merge operations are workflow node capabilities guarded by shared + repository safety services. +- R5. Retry state is node/run scoped; task retry fields are compatibility + projections only. +- R6. Self-healing publishes typed workflow recovery facts and wakes recovery + nodes; it does not directly mutate merge/retry lifecycle. +- R7. Dashboard/API/CLI state derives from workflow state first. +- R8. Existing invariants remain non-configurable: `autoMerge:false`, hard + cancel on `in-progress -> todo`, file-scope/squash guards, branch-group target + rules, and user pauses. +- R9. Branch-group member integration and group promotion are workflow-owned + subgraphs with separate gates. +- R10. Final deletion tests fail if engine-owned merge/retry/scheduling policy + reappears. + +## Current Baseline + +The starting checkpoint is `docs/workflow-policy-ownership-map.md`. It classifies +today's ownership of merge queue enqueue/dequeue, scheduler in-review policy, +merge request shadow state, git/merge procedures, retry helpers, manual retry, +self-healing recovery, built-in workflow IR, and dashboard projections. + +Keep that map updated through the migration. A slice is not complete if it moves +policy without updating the map or adding the corresponding deletion gate. + +## Slice Strategy + +- Keep each slice mergeable and behavior-preserving unless the slice is an + explicit cutover gate. +- Prefer characterization-first on legacy policy before moving it. +- Introduce workflow-native state and projections before changing production + routing. +- Route one ownership surface at a time, then delete the old owner. +- Run the merge gate for every slice: `pnpm test:gate`. +- Add `pnpm lint` and `pnpm build` for every behavior-bearing slice. +- Add a changeset only when a slice changes published `@runfusion/fusion` + behavior. + +## Migration Slices + +### S0. Ownership Map And Guard + +- **Goal:** Keep the migration inventory explicit and enforced. +- **Status:** Done by the origin PR. +- **Files:** `docs/workflow-policy-ownership-map.md`, + `packages/engine/src/__tests__/workflow-policy-ownership-map.test.ts`, + `packages/engine/vitest.config.ts`. +- **Tests:** `packages/engine/src/__tests__/workflow-policy-ownership-map.test.ts`. +- **Exit gate:** Every known current policy surface is classified as + `substrate`, `workflow-policy`, `capability`, `compat-projection`, or + `delete-after-cutover`. + +### S1. Workflow Work-Item Schema And Store API + +- **Goal:** Add durable workflow work items that can represent runnable, held, + retrying, merge, manual-hold, and recovery work without changing production + routing yet. +- **Depends on:** S0. +- **Files:** `packages/core/src/db.ts`, `packages/core/src/store.ts`, + `packages/core/src/types.ts`, `packages/core/src/index.ts`, + `packages/core/src/__tests__/central-db.test.ts`, + `packages/core/src/__tests__/store-workflow-runtime.test.ts` (new), + `packages/core/src/__tests__/merge-request-record.test.ts`. +- **Decisions:** Work items are keyed by workflow run, task, node, and kind. + Minimum fields: `id`, `runId`, `taskId`, `nodeId`, `kind`, `state`, + `attempt`, `retryAfter`, `leaseOwner`, `leaseExpiresAt`, `lastError`, + `blockedReason`, `createdAt`, `updatedAt`. +- **Test scenarios:** create runnable work; create merge work; transition to + held/retrying/manual-required/succeeded/cancelled/exhausted; reclaim expired + lease; duplicate wakeups are idempotent; completed work cannot be requeued. +- **Exit gate:** Store can find due runnable work without reading task columns + for merge/retry policy. + +### S2. Merge Request Projection Onto Work Items + +- **Goal:** Project existing merge request records into workflow work-item state + so dashboards and schedulers can dual-read before cutover. +- **Depends on:** S1. +- **Files:** `packages/core/src/store.ts`, `packages/core/src/task-merge.ts`, + `packages/core/src/types.ts`, + `packages/core/src/__tests__/merge-request-record.test.ts`, + `packages/core/src/__tests__/store-workflow-runtime.test.ts`, + `packages/engine/src/__tests__/dual-observe-merge-seam.test.ts`. +- **Decisions:** Existing `mergeRequestContractShadowEnabled` remains a + compatibility switch during this slice. Work-item state is the new shape; + merge request rows remain the old projection. +- **Test scenarios:** queued/running/retrying/manual-required/succeeded rows + project to equivalent work items; task hard-cancel cancels active merge work; + exhausted merge request maps to terminal failed work; projection is + idempotent across restart. +- **Exit gate:** Every merge request state has a lossless workflow work-item + equivalent. + +### S3. Generic Scheduler Claim Path + +- **Goal:** Teach `Scheduler` to claim due workflow work items while preserving + existing task dispatch behavior. +- **Depends on:** S1. +- **Files:** `packages/engine/src/scheduler.ts`, + `packages/engine/src/workflow-task-runtime.ts`, + `packages/engine/src/project-engine.ts`, + `packages/engine/src/__tests__/scheduler.test.ts`, + `packages/engine/src/__tests__/scheduler-node-routing.test.ts`, + `packages/engine/src/__tests__/workflow-work-engine-dispatch.test.ts` (new). +- **Decisions:** Scheduler remains substrate. It may apply capacity, routing, + leases, global pause, engine pause, and remote-node dispatch. It must not own + merge eligibility, retry routing, or recovery outcome. +- **Test scenarios:** claim only due runnable work; skip `retryAfter` until due; + hold on capacity without task mutation; user-paused work is not claimed; stale + leases are reclaimable; remote node receives workflow runtime work. +- **Exit gate:** A workflow work item can be dispatched end to end in tests + without constructing a merge queue branch. + +### S4. Built-In Merge/Retry/Recovery IR Regions + +- **Goal:** Add explicit merge, retry, manual hold, branch-group, and recovery + regions to built-in workflow IR. +- **Depends on:** S1, S2. +- **Files:** `packages/core/src/builtin-coding-workflow-ir.ts`, + `packages/core/src/builtin-stepwise-coding-workflow-ir.ts`, + `packages/core/src/builtin-pr-workflow-ir.ts`, + `packages/core/src/builtin-workflows.ts`, + `packages/core/src/workflow-ir-types.ts`, + `packages/core/src/__tests__/builtin-coding-workflow-ir.test.ts`, + `packages/core/src/__tests__/builtin-stepwise-coding-workflow-ir.test.ts`, + `packages/core/src/__tests__/builtin-pr-workflow-ir.test.ts`. +- **Decisions:** Use built-in node kinds for non-authorable primitives: + merge gate, merge attempt, manual merge hold, retry/backoff, branch-group + member integration, group promotion, finalize, and recovery router. +- **Test scenarios:** built-in workflows validate; default coding has a merge + gate; stepwise coding has per-step retry plus merge retry; PR workflow routes + review/fix/merge; `autoMerge:false` routes to manual hold; branch-group member + integration and group promotion are separate nodes. +- **Exit gate:** Built-in IR is the source of truth for all default + merge/retry/recovery policy, even if production handlers are not wired yet. + +### S5. Runtime Work-Item Driver + +- **Goal:** Let `WorkflowTaskRuntime` start from a workflow work item and persist + node/work-item outcomes. +- **Depends on:** S1, S3, S4. +- **Files:** `packages/engine/src/workflow-task-runtime.ts`, + `packages/engine/src/workflow-graph-executor.ts`, + `packages/engine/src/workflow-node-handlers.ts`, + `packages/engine/src/__tests__/workflow-task-runtime.test.ts`, + `packages/engine/src/__tests__/workflow-graph-executor-retry-coding-workflow.test.ts`, + `packages/engine/src/__tests__/workflow-node-handlers.test.ts`. +- **Decisions:** Runtime receives `{ workItemId, runId, taskId, nodeId }` and + returns a typed outcome that updates work item state. Task column updates are + side effects of workflow primitives, not scheduler policy. +- **Test scenarios:** runnable work completes; failing node creates retrying + work; manual hold node creates held work; runtime restart resumes from stored + work; duplicate start of same work item is refused by lease. +- **Exit gate:** Runtime can progress workflow work without old merge queue + callbacks. + +### S6. Git And Merge Capability Extraction + +- **Goal:** Put checkout preparation, branch integration, merge attempt, squash, + finalize, and conflict classification behind workflow node capability modules. +- **Depends on:** S4, S5. +- **Files:** `packages/engine/src/merger.ts`, + `packages/engine/src/merger-ai.ts`, + `packages/engine/src/merger-integration-worktree.ts`, + `packages/engine/src/workflow-merge-nodes.ts` (new), + `packages/engine/src/workflow-node-handlers.ts`, + `packages/engine/src/merge-trait.ts`, + `packages/engine/src/__tests__/interpreter-merge-seam.test.ts`, + `packages/engine/src/__tests__/dual-observe-merge-seam.test.ts`, + `packages/engine/src/__tests__/workflow-merge-nodes.test.ts` (new). +- **Decisions:** This slice does not rewrite low-level merge algorithms. It + extracts orchestration boundaries so workflow nodes call existing guarded + operations. +- **Test scenarios:** merge node calls checkout preparation; file-scope + violation returns workflow failure; already-on-main routes to finalize; + transient merge error returns retry outcome; non-transient conflict routes to + revision/manual hold; no production caller can bypass guard service in tests. +- **Exit gate:** A merge attempt can be driven by a workflow node capability in + tests with the same guard behavior as `merger.ts`. + +### S7. Completion Handoff Creates Merge Work + +- **Goal:** Replace task-moved `in-review` auto-enqueue as the policy authority + with workflow completion handoff creating merge work. +- **Depends on:** S2, S5, S6. +- **Files:** `packages/engine/src/project-engine.ts`, + `packages/engine/src/merger.ts`, + `packages/core/src/store.ts`, + `packages/engine/src/__tests__/workflow-interpreter-cutover.test.ts`, + `packages/engine/src/__tests__/completion-fanout-x-self-healing.test.ts`, + `packages/engine/src/__tests__/merge-reuse-task-worktree.slow.test.ts`. +- **Decisions:** During this slice the old queue can remain as a projection, but + merge work creation happens through workflow handoff. `autoMerge:false` creates + a manual hold work item. +- **Test scenarios:** coding completion creates merge work; `autoMerge:false` + creates manual hold and does not enqueue merge; duplicate handoff is + idempotent; soft-deleted task cancels handoff; startup projection does not + create duplicate merge work. +- **Exit gate:** New task completions produce workflow merge work before any old + queue processing path runs. + +### S8. Workflow-Owned Merge Queue Processing + +- **Goal:** Process merge work items through workflow runtime instead of + `ProjectEngine`'s in-memory merge queue loop. +- **Depends on:** S3, S6, S7. +- **Files:** `packages/engine/src/project-engine.ts`, + `packages/engine/src/scheduler.ts`, + `packages/engine/src/merger.ts`, + `packages/core/src/store.ts`, + `packages/engine/src/__tests__/merger-merge-lifecycle.test.ts`, + `packages/engine/src/__tests__/merger-post-merge.test.ts`, + `packages/engine/src/__tests__/workflow-work-engine-dispatch.test.ts`, + `packages/engine/src/__tests__/workflow-merge-nodes.test.ts`. +- **Decisions:** Keep queue fairness and serialization as substrate leases. The + policy route after success/failure belongs to workflow node outcomes. +- **Test scenarios:** queued merge work claims one at a time; successful merge + finalizes task; transient failure schedules retrying merge work; permanent + conflict routes to revision/manual hold; active merge lease blocks duplicate + processing; hard cancel cancels running merge work. +- **Exit gate:** Production merge processing no longer depends on a hidden + `mergeQueue` dequeue loop. + +### S9. Workflow-Owned Retry State + +- **Goal:** Move retry attempts, budgets, backoff, retry-after, exhaustion, and + manual retry reset into workflow node/work-item state. +- **Depends on:** S5, S8. +- **Files:** `packages/engine/src/workflow-graph-executor.ts`, + `packages/engine/src/workflow-node-handlers.ts`, + `packages/engine/src/retry-with-backoff.ts`, + `packages/engine/src/rate-limit-retry.ts`, + `packages/engine/src/transient-merge-error-classifier.ts`, + `packages/core/src/retry-summary.ts`, + `packages/core/src/manual-retry-reset.ts`, + `packages/engine/src/__tests__/workflow-node-retry-policy.test.ts` (new), + `packages/core/src/__tests__/manual-retry-reset.test.ts`. +- **Decisions:** Task retry fields remain as derived display summaries until + deletion. Manual retry emits a workflow wake and clears only targeted failed + node state. +- **Test scenarios:** implementation node retry stays within budget; merge node + retry does not reset implementation progress; rate-limit error persists due + time; exhausted retry routes to failure/manual hold; manual retry clears only + failed node; retry state survives restart. +- **Exit gate:** No retry branch is controlled solely by task counters. + +### S10. Self-Healing Recovery Events + +- **Goal:** Convert self-healing merge/retry lifecycle mutations into typed + workflow recovery events and node wakes. +- **Depends on:** S5, S8, S9. +- **Files:** `packages/engine/src/self-healing.ts`, + `packages/engine/src/restart-recovery-coordinator.ts`, + `packages/engine/src/recovery-policy.ts`, + `packages/engine/src/workflow-task-runtime.ts`, + `packages/engine/src/__tests__/self-healing.test.ts`, + `packages/engine/src/__tests__/workflow-recovery-events.test.ts` (new), + `packages/engine/src/__tests__/reliability-interactions/in-review-automerge-off.test.ts`, + `packages/engine/src/__tests__/reliability-interactions/workflow-interpreter-cutover.test.ts`. +- **Decisions:** Sweeps detect facts. Recovery nodes decide routes. Non-task + agent/heartbeat cleanup may remain engine-owned when it is not task lifecycle + policy. +- **Test scenarios:** mergeable in-review task gets recovery event, not direct + requeue; stale merge status emits event; transient merge failure emits retry + event; already landed emits finalize event; `autoMerge:false` remains terminal; + duplicate recovery events are deduped. +- **Exit gate:** Self-healing no longer directly requeues, pauses, fails, + unpauses, or moves merge/retry tasks except through guarded workflow + primitives. + +### S11. Branch Group Workflow Subgraphs + +- **Goal:** Move branch-group member integration and group promotion into + workflow-owned merge subgraphs. +- **Depends on:** S6, S8, S10. +- **Files:** `packages/engine/src/group-merge-coordinator.ts`, + `packages/engine/src/merge-trait.ts`, + `packages/engine/src/merger-integration-worktree.ts`, + `packages/core/src/builtin-coding-workflow-ir.ts`, + `packages/engine/src/__tests__/reliability-interactions/shared-branch-group-lifecycle.slow.test.ts`, + `packages/engine/src/__tests__/workflow-branch-group-merge.test.ts` (new). +- **Decisions:** Member-to-shared-branch integration and shared-branch-to-default + promotion are distinct workflow nodes with distinct auto-merge gates. +- **Test scenarios:** shared member integrates while global auto-merge is off + under the scoped exception; group promotion remains blocked when group/global + auto-merge is off; conflicting member integration routes to recovery/revision; + final group promotion runs file-scope and squash guards. +- **Exit gate:** Branch-group coordinator no longer owns task lifecycle + independent of workflow runtime. + +### S12. Dashboard/API/CLI Workflow Projection + +- **Goal:** Surface workflow-native queued, retrying, merging, manual-hold, + failed, stalled, and recovered reasons across user inspection surfaces. +- **Depends on:** S1, S2, S7, S9, S10. +- **Files:** `packages/dashboard/app/components/TaskCard.tsx`, + task detail components, reliability views, task API routes, + CLI task output files, `packages/core/src/retry-summary.ts`, + `packages/core/src/task-merge.ts`, + `packages/dashboard/app/components/__tests__/TaskCard.test.tsx`, + reliability/dashboard API tests. +- **Decisions:** Workflow state wins over stale task fields. Legacy task fields + remain fallback for old rows only. +- **Test scenarios:** merge queued shows workflow merge work, not stalled; + retrying shows attempt and due time; manual hold shows human action required; + recovery event shows reason; completed work hides stale stalled badges; + branch-group merge work identifies target branch. +- **Exit gate:** UI/API/CLI tests prove workflow state is the first projection + source. + +### S13. Scheduler Policy Deletion + +- **Goal:** Delete scheduler branches that infer lifecycle, merge eligibility, + retry routing, or in-review dependency behavior from task columns. +- **Depends on:** S3, S7, S8, S12. +- **Files:** `packages/engine/src/scheduler.ts`, + `packages/core/src/task-merge.ts`, + `packages/engine/src/__tests__/scheduler.test.ts`, + `packages/engine/src/__tests__/scheduler-overlap-requeue.test.ts`, + `packages/engine/src/__tests__/workflow-scheduler-policy-deletion.test.ts` (new). +- **Decisions:** Dependency satisfaction should use completion handoff/workflow + state. In-review scope leases are replaced by workflow work leases and guard + services. +- **Test scenarios:** scheduler cannot satisfy dependency only because a task is + `in-review`; retry due time comes from work item; overlap lease comes from + workflow work; PR monitor behavior remains as watch substrate, not lifecycle + owner. +- **Exit gate:** Search/structure test fails if scheduler reintroduces + task-column merge/retry policy. + +### S14. ProjectEngine Merge Queue Deletion + +- **Goal:** Remove production `ProjectEngine` merge queue policy and retain only + explicit human/manual event entry points plus substrate helpers. +- **Depends on:** S8, S11, S13. +- **Files:** `packages/engine/src/project-engine.ts`, + `packages/engine/src/runtimes/in-process-runtime.ts`, + `packages/core/src/store.ts`, + `packages/engine/src/__tests__/merger-merge-lifecycle.test.ts`, + `packages/engine/src/__tests__/workflow-merge-policy-deletion.test.ts` (new). +- **Decisions:** Manual merge APIs record a workflow event or create due workflow + work; they do not enqueue hidden engine work. +- **Test scenarios:** no startup in-review scan enqueues hidden merge work; + unpause wakes workflow work; manual merge event wakes merge node; stale + `mergeActive` state cannot block workflow work; old queue APIs are absent or + compatibility-only. +- **Exit gate:** No production caller starts merge processing outside workflow + runtime. + +### S15. Self-Healing Policy Deletion + +- **Goal:** Delete self-healing direct lifecycle mutations for merge/retry tasks + after recovery events cover all cases. +- **Depends on:** S10, S11, S14. +- **Files:** `packages/engine/src/self-healing.ts`, + `docs/self-healing-backward-move-audit.md`, + `packages/engine/src/__tests__/self-healing.test.ts`, + `packages/engine/src/__tests__/workflow-recovery-events.test.ts`, + `packages/engine/src/__tests__/workflow-self-healing-policy-deletion.test.ts` (new). +- **Decisions:** Metadata reconciliation and non-task agent cleanup can remain. + Task lifecycle repair becomes recovery events plus workflow node outcomes. +- **Test scenarios:** direct calls to `moveTask(..., "todo")`, + `updateTask({ paused: true })`, merge requeue callbacks, and merge retry resets + are absent for merge/retry surfaces; valid held states are no-ops; recovery + facts carry audit context. +- **Exit gate:** Search tests fail on direct self-healing merge/retry lifecycle + mutation patterns. + +### S16. Legacy Retry Field Demotion + +- **Goal:** Demote task-level retry/merge counters to projections and remove + policy reads that still treat them as authority. +- **Depends on:** S9, S12, S15. +- **Files:** `packages/core/src/types.ts`, `packages/core/src/retry-summary.ts`, + `packages/core/src/manual-retry-reset.ts`, `packages/core/src/store.ts`, + `packages/engine/src/project-engine.ts`, `packages/engine/src/self-healing.ts`, + `packages/core/src/__tests__/manual-retry-reset.test.ts`, + `packages/engine/src/__tests__/workflow-node-retry-policy.test.ts`. +- **Decisions:** Do not remove fields until all compatibility surfaces can read + workflow projections. Removal can be a later cleanup; this slice removes policy + authority. +- **Test scenarios:** retry summaries derive from workflow node/work state; + manual retry emits workflow wake; old task fields changing alone cannot cause + scheduler/recovery/merge action. +- **Exit gate:** Task retry fields are display-only compatibility data. + +### S17. End-To-End Cutover Matrix + +- **Goal:** Prove the full workflow-owned invariant across all known production + surfaces before removing dual-read compatibility. +- **Depends on:** S13, S14, S15, S16. +- **Files:** focused tests across `packages/engine/src/__tests__/`, + reliability interactions under + `packages/engine/src/__tests__/reliability-interactions/`, core store tests, + dashboard projection tests, `docs/testing.md`. +- **Test matrix:** default coding auto-merge; stepwise coding; custom workflow; + PR workflow; plugin workflow extension; `autoMerge:false`; manual retry; + user hard cancel; engine restart during merge work; transient merge failure; + permanent conflict; branch-group member integration; branch-group promotion; + stale recovery; already-landed finalization; dashboard task card/detail; + CLI task output. +- **Exit gate:** `pnpm test:gate`, `pnpm lint`, `pnpm build`, and targeted matrix + suites pass. No old engine merge/retry/scheduling policy path can race workflow + runtime in production. + +### S18. Documentation, Settings, And Release Notes + +- **Goal:** Update architecture, settings, dashboard, CLI, and testing docs for + workflow-owned policy and compatibility projections. +- **Depends on:** S17. +- **Files:** `docs/architecture.md`, `docs/workflow-steps.md`, + `docs/dashboard-guide.md`, `docs/settings-reference.md`, `docs/testing.md`, + `CONCEPTS.md`, `.changeset/.md`. +- **Decisions:** Document the new source of truth, remaining compatibility fields, + recovery event vocabulary, manual hold behavior, branch-group routing, and + deletion gates. +- **Test scenarios:** docs inventory/search tests if applicable; lazy view + inventory unchanged unless dashboard imports change. +- **Exit gate:** User-facing docs use the same state names as API/UI tests, and a + patch changeset exists if published `@runfusion/fusion` behavior changed. + +## Dependency Graph + +```mermaid +flowchart TB + S0 --> S1 + S1 --> S2 + S1 --> S3 + S2 --> S4 + S3 --> S5 + S4 --> S5 + S5 --> S6 + S6 --> S7 + S7 --> S8 + S8 --> S9 + S9 --> S10 + S8 --> S11 + S10 --> S11 + S7 --> S12 + S9 --> S12 + S10 --> S12 + S12 --> S13 + S13 --> S14 + S11 --> S14 + S14 --> S15 + S15 --> S16 + S16 --> S17 + S17 --> S18 +``` + +## Release And Merge Strategy + +- **Preferred PR count:** 18 slices, one PR per slice. +- **Can combine:** S1+S2 if schema and projection are small; S13+S14 if deletion + is purely mechanical after S8. +- **Do not combine:** S8 with S14, or S10 with S15. Route through workflow first, + then delete old owner in a separate reviewable PR. +- **Branch policy:** Each slice branches from current `main`, not from a stale + feature stack. Drop duplicate commits before merging. +- **Changesets:** Add only when published CLI behavior changes. Internal docs, + CI config, and behavior-preserving refactors do not require changesets. + +## Cutover Safety Gates + +- Gate A after S4: built-in IR expresses all planned policy regions. +- Gate B after S8: workflow runtime can process merge work without hidden queue + ownership. +- Gate C after S10: self-healing emits recovery events for merge/retry surfaces. +- Gate D after S12: dashboard/API/CLI read workflow state first. +- Gate E after S17: deletion tests and end-to-end matrix prove no production + legacy control plane remains. + +## Rollback Strategy + +- Before S13, rollback is disabling workflow work dispatch and relying on legacy + projections. +- After S13, rollback is revert-by-slice, not runtime fallback. Do not ship a + production dual-controller fallback after deletion gates begin. +- Keep old fields as compatibility projections through S17 so data downgrade is + not required for ordinary slice rollback. + +## Verification Commands + +- `pnpm test:gate` +- `pnpm lint` +- `pnpm build` +- Targeted suites named in each slice. +- `pnpm test:full` only for explicit final matrix verification or release-adjacent + confidence, not as the normal merge gate. + +## Done Criteria + +- Workflow work items are the durable source of runnable, held, retrying, merge, + and recovery work. +- Built-in workflows express default merge, retry, scheduling, branch-group, and + recovery policy. +- Scheduler dispatches generic workflow work only. +- Git/merge operations are invoked by workflow nodes or explicit human/manual + events recorded as workflow events. +- Retry budgets and manual retry reset are node/run scoped. +- Self-healing publishes recovery facts and wakes workflow recovery nodes. +- Dashboard/API/CLI projections read workflow state first. +- Deletion tests prevent reintroducing engine-owned merge/retry/scheduling + policy. diff --git a/packages/core/src/__tests__/db-migrate.test.ts b/packages/core/src/__tests__/db-migrate.test.ts index fd69f5c1b0..3b06532f27 100644 --- a/packages/core/src/__tests__/db-migrate.test.ts +++ b/packages/core/src/__tests__/db-migrate.test.ts @@ -715,8 +715,8 @@ describe("schema migration", () => { const row = db.prepare("SELECT deletedAt FROM tasks WHERE id = 'FN-legacy'").get() as { deletedAt: string | null }; expect(row.deletedAt).toBeNull(); - expect(db.getSchemaVersion()).toBe(114); - expect(db.getSchemaVersion()).toBe(114); + expect(db.getSchemaVersion()).toBe(115); + expect(db.getSchemaVersion()).toBe(115); db.close(); }); @@ -749,8 +749,8 @@ describe("schema migration", () => { { id: "WS-001", mode: "prompt", gateMode: "advisory" }, { id: "WS-002", mode: "script", gateMode: "advisory" }, ]); - expect(db.getSchemaVersion()).toBe(114); - expect(db.getSchemaVersion()).toBe(114); + expect(db.getSchemaVersion()).toBe(115); + expect(db.getSchemaVersion()).toBe(115); db.close(); }); @@ -800,8 +800,8 @@ describe("schema migration", () => { reviewerContextRetryCount: 0, reviewerFallbackRetryCount: 0, }); - expect(db.getSchemaVersion()).toBe(114); - expect(db.getSchemaVersion()).toBe(114); + expect(db.getSchemaVersion()).toBe(115); + expect(db.getSchemaVersion()).toBe(115); db.close(); }); @@ -830,8 +830,8 @@ describe("schema migration", () => { const columns = db.prepare("PRAGMA table_info(milestones)").all() as Array<{ name: string }>; expect(columns.map((column) => column.name)).toContain("acceptanceCriteria"); - expect(db.getSchemaVersion()).toBe(114); - expect(db.getSchemaVersion()).toBe(114); + expect(db.getSchemaVersion()).toBe(115); + expect(db.getSchemaVersion()).toBe(115); db.close(); }); @@ -872,8 +872,8 @@ describe("schema migration", () => { const missionColumns = db.prepare("PRAGMA table_info(missions)").all() as Array<{ name: string }>; expect(missionColumns.map((column) => column.name)).toContain("autoMerge"); - expect(db.getSchemaVersion()).toBe(114); - expect(db.getSchemaVersion()).toBe(114); + expect(db.getSchemaVersion()).toBe(115); + expect(db.getSchemaVersion()).toBe(115); db.close(); }); @@ -907,8 +907,8 @@ describe("schema migration", () => { { id: "WS-002", mode: "script", enabled: 1, gateMode: "advisory" }, { id: "WS-003", mode: "prompt", enabled: 0, gateMode: "advisory" }, ]); - expect(db.getSchemaVersion()).toBe(114); - expect(db.getSchemaVersion()).toBe(114); + expect(db.getSchemaVersion()).toBe(115); + expect(db.getSchemaVersion()).toBe(115); db.close(); }); @@ -945,8 +945,8 @@ describe("schema migration", () => { const indexes = db.prepare("PRAGMA index_list(mission_goals)").all() as Array<{ name: string }>; expect(indexes.some((index) => index.name === "idxMissionGoalsGoalId")).toBe(true); - expect(db.getSchemaVersion()).toBe(114); - expect(db.getSchemaVersion()).toBe(114); + expect(db.getSchemaVersion()).toBe(115); + expect(db.getSchemaVersion()).toBe(115); db.close(); }); @@ -1007,7 +1007,7 @@ describe("schema migration", () => { expect(customFieldsColumn).toBeDefined(); expect(customFieldsColumn?.dflt_value).toBe("'{}'"); - expect(db.getSchemaVersion()).toBe(114); + expect(db.getSchemaVersion()).toBe(115); db.close(); }); @@ -1045,7 +1045,7 @@ describe("schema migration", () => { const indexes = db.prepare("PRAGMA index_list(workflow_settings)").all() as Array<{ name: string }>; expect(indexes.some((index) => index.name === "idx_workflow_settings_project")).toBe(true); - expect(db.getSchemaVersion()).toBe(114); + expect(db.getSchemaVersion()).toBe(115); db.close(); }); @@ -1127,7 +1127,7 @@ describe("schema migration", () => { expect(indexNames).toContain("idx_cli_sessions_chatSessionId"); expect(indexNames).toContain("idx_cli_sessions_project_state"); - expect(db.getSchemaVersion()).toBe(114); + expect(db.getSchemaVersion()).toBe(115); db.close(); }); @@ -1159,7 +1159,7 @@ describe("schema migration", () => { .all() as Array<{ name: string }>; expect(columns.map((column) => column.name)).toContain("cliExecutorAdapterId"); - expect(db.getSchemaVersion()).toBe(114); + expect(db.getSchemaVersion()).toBe(115); db.close(); }); @@ -1169,7 +1169,7 @@ describe("schema migration", () => { const tables = db.prepare("SELECT name FROM sqlite_master WHERE type='table'").all() as Array<{ name: string }>; expect(tables.map((row) => row.name)).toContain("cli_sessions"); - expect(db.getSchemaVersion()).toBe(114); + expect(db.getSchemaVersion()).toBe(115); db.close(); }); @@ -1226,23 +1226,23 @@ describe("schema migration", () => { .get() as { migrated_fragment_id: string | null }; expect(stepRow.migrated_fragment_id).toBeNull(); - expect(db.getSchemaVersion()).toBe(114); - expect(db.getSchemaVersion()).toBe(114); + expect(db.getSchemaVersion()).toBe(115); + expect(db.getSchemaVersion()).toBe(115); db.close(); }); it("migration 109 is idempotent on re-init", () => { const db = new Database(fusionDir); db.init(); - expect(db.getSchemaVersion()).toBe(114); - expect(db.getSchemaVersion()).toBe(114); + expect(db.getSchemaVersion()).toBe(115); + expect(db.getSchemaVersion()).toBe(115); db.close(); // Re-open the same on-disk DB: already at 109, the 109 block must be a no-op. const reopened = new Database(fusionDir); reopened.init(); - expect(reopened.getSchemaVersion()).toBe(114); - expect(reopened.getSchemaVersion()).toBe(114); + expect(reopened.getSchemaVersion()).toBe(115); + expect(reopened.getSchemaVersion()).toBe(115); const workflowColumns = reopened.prepare("PRAGMA table_info(workflows)").all() as Array<{ name: string }>; expect(workflowColumns.filter((c) => c.name === "kind")).toHaveLength(1); const stepColumns = reopened.prepare("PRAGMA table_info(workflow_steps)").all() as Array<{ name: string }>; diff --git a/packages/core/src/__tests__/db.test.ts b/packages/core/src/__tests__/db.test.ts index 23d1f377b7..a4f89a3caa 100644 --- a/packages/core/src/__tests__/db.test.ts +++ b/packages/core/src/__tests__/db.test.ts @@ -334,8 +334,8 @@ describe("Database", () => { }); it("seeds schema version", () => { - expect(db.getSchemaVersion()).toBe(114); - expect(db.getSchemaVersion()).toBe(114); + expect(db.getSchemaVersion()).toBe(115); + expect(db.getSchemaVersion()).toBe(115); }); it("includes tokenUsageCacheWriteTokens on freshly initialized tasks table", () => { @@ -394,8 +394,8 @@ describe("Database", () => { it("is idempotent - calling init() twice does not fail", () => { expect(() => db.init()).not.toThrow(); - expect(db.getSchemaVersion()).toBe(114); - expect(db.getSchemaVersion()).toBe(114); + expect(db.getSchemaVersion()).toBe(115); + expect(db.getSchemaVersion()).toBe(115); }); it("does not overwrite existing config on re-init", () => { // Update the config @@ -1465,8 +1465,8 @@ describe("schema migrations", () => { db.init(); // Verify version bumped to 29 (includes v1→v2 through v26→v29) - expect(db.getSchemaVersion()).toBe(114); - expect(db.getSchemaVersion()).toBe(114); + expect(db.getSchemaVersion()).toBe(115); + expect(db.getSchemaVersion()).toBe(115); // Verify new columns exist and existing data is intact const cols = db.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>; @@ -1491,16 +1491,16 @@ describe("schema migrations", () => { const db = new Database(fusionDir); db.init(); - expect(db.getSchemaVersion()).toBe(114); + expect(db.getSchemaVersion()).toBe(115); // Re-init should not fail db.init(); - expect(db.getSchemaVersion()).toBe(114); - expect(db.getSchemaVersion()).toBe(114); + expect(db.getSchemaVersion()).toBe(115); + expect(db.getSchemaVersion()).toBe(115); // Re-init should not fail db.init(); - expect(db.getSchemaVersion()).toBe(114); + expect(db.getSchemaVersion()).toBe(115); db.close(); }); @@ -1535,8 +1535,8 @@ describe("schema migrations", () => { db.init(); - expect(db.getSchemaVersion()).toBe(114); - expect(db.getSchemaVersion()).toBe(114); + expect(db.getSchemaVersion()).toBe(115); + expect(db.getSchemaVersion()).toBe(115); const cols = db.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>; expect(cols.map((col) => col.name)).toContain("priority"); @@ -1577,8 +1577,8 @@ describe("schema migrations", () => { db.init(); - expect(db.getSchemaVersion()).toBe(114); - expect(db.getSchemaVersion()).toBe(114); + expect(db.getSchemaVersion()).toBe(115); + expect(db.getSchemaVersion()).toBe(115); const cols = db.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>; const colNames = cols.map((col) => col.name); @@ -1650,8 +1650,8 @@ describe("schema migrations", () => { db.init(); - expect(db.getSchemaVersion()).toBe(114); - expect(db.getSchemaVersion()).toBe(114); + expect(db.getSchemaVersion()).toBe(115); + expect(db.getSchemaVersion()).toBe(115); const cols = db.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>; const colNames = cols.map((col) => col.name); @@ -1891,8 +1891,8 @@ describe("schema migrations", () => { db.init(); - expect(db.getSchemaVersion()).toBe(114); - expect(db.getSchemaVersion()).toBe(114); + expect(db.getSchemaVersion()).toBe(115); + expect(db.getSchemaVersion()).toBe(115); const cols = db.prepare("PRAGMA table_info(chat_messages)").all() as Array<{ name: string }>; expect(cols.map((col) => col.name)).toContain("attachments"); @@ -1966,8 +1966,8 @@ describe("schema migrations", () => { db.init(); - expect(db.getSchemaVersion()).toBe(114); - expect(db.getSchemaVersion()).toBe(114); + expect(db.getSchemaVersion()).toBe(115); + expect(db.getSchemaVersion()).toBe(115); const tables = db.prepare("SELECT name FROM sqlite_master WHERE type='table' AND name = 'agentRatings'").all() as Array<{ name: string }>; expect(tables).toEqual([{ name: "agentRatings" }]); @@ -1991,8 +1991,8 @@ describe("schema migrations", () => { db.init(); - expect(db.getSchemaVersion()).toBe(114); - expect(db.getSchemaVersion()).toBe(114); + expect(db.getSchemaVersion()).toBe(115); + expect(db.getSchemaVersion()).toBe(115); const tables = db.prepare("SELECT name FROM sqlite_master WHERE type='table' AND name = 'mission_events'").all() as Array<{ name: string }>; expect(tables).toEqual([{ name: "mission_events" }]); @@ -2096,8 +2096,8 @@ describe("schema migrations", () => { db.init(); // Verify version bumped to 29 - expect(db.getSchemaVersion()).toBe(114); - expect(db.getSchemaVersion()).toBe(114); + expect(db.getSchemaVersion()).toBe(115); + expect(db.getSchemaVersion()).toBe(115); // Verify new columns exist and existing data is intact const cols = db.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>; @@ -2316,8 +2316,8 @@ describe("schema migrations", () => { localDb.init(); - expect(localDb.getSchemaVersion()).toBe(114); - expect(localDb.getSchemaVersion()).toBe(114); + expect(localDb.getSchemaVersion()).toBe(115); + expect(localDb.getSchemaVersion()).toBe(115); const columns = localDb.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>; expect(columns.map((column) => column.name)).toContain("tokenUsageCacheWriteTokens"); @@ -2628,8 +2628,8 @@ describe("createDatabase factory", () => { const db = createDatabase(fusionDir); db.init(); - expect(db.getSchemaVersion()).toBe(114); - expect(db.getSchemaVersion()).toBe(114); + expect(db.getSchemaVersion()).toBe(115); + expect(db.getSchemaVersion()).toBe(115); expect(db.getLastModified()).toBeGreaterThan(0); db.close(); @@ -2783,8 +2783,8 @@ describe("migration v77 task token budget columns", () => { migrated = new Database(fusion); migrated.init(); - expect(migrated.getSchemaVersion()).toBe(114); - expect(migrated.getSchemaVersion()).toBe(114); + expect(migrated.getSchemaVersion()).toBe(115); + expect(migrated.getSchemaVersion()).toBe(115); const rows = migrated.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>; const names = new Set(rows.map((row) => row.name)); expect(names.has("tokenBudgetSoftAlertedAt")).toBe(true); @@ -2815,8 +2815,8 @@ describe("migration v106 adds tasks.transitionPending (FN-1417)", () => { const fresh = new Database(fusion); try { fresh.init(); - expect(fresh.getSchemaVersion()).toBe(114); - expect(fresh.getSchemaVersion()).toBe(114); + expect(fresh.getSchemaVersion()).toBe(115); + expect(fresh.getSchemaVersion()).toBe(115); const names = new Set( (fresh.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>).map((r) => r.name), ); @@ -2844,8 +2844,8 @@ describe("migration v106 adds tasks.transitionPending (FN-1417)", () => { migrated = new Database(fusion); migrated.init(); - expect(migrated.getSchemaVersion()).toBe(114); - expect(migrated.getSchemaVersion()).toBe(114); + expect(migrated.getSchemaVersion()).toBe(115); + expect(migrated.getSchemaVersion()).toBe(115); const names = new Set( (migrated.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>).map((r) => r.name), ); @@ -2871,8 +2871,8 @@ describe("migration v107 adds workflow_run_branches + index (FN-1417)", () => { const fresh = new Database(fusion); try { fresh.init(); - expect(fresh.getSchemaVersion()).toBe(114); - expect(fresh.getSchemaVersion()).toBe(114); + expect(fresh.getSchemaVersion()).toBe(115); + expect(fresh.getSchemaVersion()).toBe(115); const table = fresh .prepare("SELECT name FROM sqlite_master WHERE type='table' AND name = 'workflow_run_branches'") .get() as { name: string } | undefined; @@ -2906,8 +2906,8 @@ describe("migration v107 adds workflow_run_branches + index (FN-1417)", () => { migrated = new Database(fusion); migrated.init(); - expect(migrated.getSchemaVersion()).toBe(114); - expect(migrated.getSchemaVersion()).toBe(114); + expect(migrated.getSchemaVersion()).toBe(115); + expect(migrated.getSchemaVersion()).toBe(115); const table = migrated .prepare("SELECT name FROM sqlite_master WHERE type='table' AND name = 'workflow_run_branches'") .get() as { name: string } | undefined; @@ -2948,8 +2948,8 @@ describe("migration v67 drops orphan project auth tables", () => { migrated = new Database(fusion); migrated.init(); - expect(migrated.getSchemaVersion()).toBe(114); - expect(migrated.getSchemaVersion()).toBe(114); + expect(migrated.getSchemaVersion()).toBe(115); + expect(migrated.getSchemaVersion()).toBe(115); const tables = migrated .prepare("SELECT name FROM sqlite_master WHERE type='table' AND name LIKE 'project_auth_%'") .all() as Array<{ name: string }>; @@ -2976,8 +2976,8 @@ describe("migration v67 drops orphan project auth tables", () => { try { fresh.init(); - expect(fresh.getSchemaVersion()).toBe(114); - expect(fresh.getSchemaVersion()).toBe(114); + expect(fresh.getSchemaVersion()).toBe(115); + expect(fresh.getSchemaVersion()).toBe(115); const tables = fresh .prepare("SELECT name FROM sqlite_master WHERE type='table' AND name LIKE 'project_auth_%'") .all() as Array<{ name: string }>; diff --git a/packages/core/src/__tests__/goals-schema.test.ts b/packages/core/src/__tests__/goals-schema.test.ts index d1859c8df3..5ad25567d1 100644 --- a/packages/core/src/__tests__/goals-schema.test.ts +++ b/packages/core/src/__tests__/goals-schema.test.ts @@ -91,6 +91,6 @@ describe("goals schema", () => { }); it("reports schema version 101", () => { - expect(db.getSchemaVersion()).toBe(114); + expect(db.getSchemaVersion()).toBe(115); }); }); diff --git a/packages/core/src/__tests__/insight-store.test.ts b/packages/core/src/__tests__/insight-store.test.ts index cc42db9ec3..cd834d81a9 100644 --- a/packages/core/src/__tests__/insight-store.test.ts +++ b/packages/core/src/__tests__/insight-store.test.ts @@ -1000,7 +1000,7 @@ describe("Migration: pre-33 DB upgrade", () => { // Step 1: Create a fresh database at v33 (runs all migrations up to 33) const db1 = createDatabase(legacyDir); db1.init(); - expect(db1.getSchemaVersion()).toBe(114); + expect(db1.getSchemaVersion()).toBe(115); db1.close(); // Step 2: Manually downgrade to version 32 and drop insight tables @@ -1035,7 +1035,7 @@ describe("Migration: pre-33 DB upgrade", () => { expect(tableNamesBefore).not.toContain("project_insight_runs"); // Now run init — this triggers the v32→v33 migration db3.init(); - expect(db3.getSchemaVersion()).toBe(114); + expect(db3.getSchemaVersion()).toBe(115); // Step 4: Verify insight tables exist after migration const tablesAfter = db3.prepare( @@ -1066,12 +1066,12 @@ describe("Migration: pre-33 DB upgrade", () => { try { const db1 = createDatabase(testDir); db1.init(); - expect(db1.getSchemaVersion()).toBe(114); + expect(db1.getSchemaVersion()).toBe(115); db1.close(); const db2 = createDatabase(testDir); expect(() => db2.init()).not.toThrow(); - expect(db2.getSchemaVersion()).toBe(114); + expect(db2.getSchemaVersion()).toBe(115); db2.close(); } finally { rmSync(testDir, { recursive: true, force: true }); @@ -1085,7 +1085,7 @@ describe("Migration: pre-33 DB upgrade", () => { // Step 1: Create a fresh DB and run migrations const db1 = createDatabase(compatDir); db1.init(); - expect(db1.getSchemaVersion()).toBe(114); + expect(db1.getSchemaVersion()).toBe(115); // Step 2: Strip lifecycle and cancelledAt columns by recreating the // table without them. This simulates a DB that was created before the diff --git a/packages/core/src/__tests__/merge-request-record.test.ts b/packages/core/src/__tests__/merge-request-record.test.ts index 364dbdeac8..018b79c15e 100644 --- a/packages/core/src/__tests__/merge-request-record.test.ts +++ b/packages/core/src/__tests__/merge-request-record.test.ts @@ -38,7 +38,7 @@ describe("TaskStore merge request record + completion handoff marker", () => { .all() as Array<{ name: string }>; expect(tableRows).toEqual([{ name: "completion_handoff_markers" }, { name: "merge_requests" }]); - expect(db.getSchemaVersion()).toBe(114); + expect(db.getSchemaVersion()).toBe(115); }); it("upserts merge request records", async () => { diff --git a/packages/core/src/__tests__/mission-store.test.ts b/packages/core/src/__tests__/mission-store.test.ts index 52fdebcac2..0f3d2b09c5 100644 --- a/packages/core/src/__tests__/mission-store.test.ts +++ b/packages/core/src/__tests__/mission-store.test.ts @@ -3746,7 +3746,7 @@ describe("MissionStore", () => { describe("Loop State & Validator Run Schema (v31)", () => { it("schema version is 101 after migration", () => { - expect(db.getSchemaVersion()).toBe(114); + expect(db.getSchemaVersion()).toBe(115); }); it("mission_features table has loop state columns", () => { diff --git a/packages/core/src/__tests__/run-audit.test.ts b/packages/core/src/__tests__/run-audit.test.ts index ce0035b235..087f71d8fc 100644 --- a/packages/core/src/__tests__/run-audit.test.ts +++ b/packages/core/src/__tests__/run-audit.test.ts @@ -584,7 +584,7 @@ describe("Run Audit", () => { }); it("schema version is bumped to 40", () => { - expect(db.getSchemaVersion()).toBe(114); + expect(db.getSchemaVersion()).toBe(115); }); }); }); diff --git a/packages/core/src/__tests__/store-merge-queue.test.ts b/packages/core/src/__tests__/store-merge-queue.test.ts index 6f5fd1551c..2db85dc1d2 100644 --- a/packages/core/src/__tests__/store-merge-queue.test.ts +++ b/packages/core/src/__tests__/store-merge-queue.test.ts @@ -60,7 +60,7 @@ describe("TaskStore merge queue", () => { expect.arrayContaining(["idx_mergeQueue_lease_ready", "idx_mergeQueue_leaseExpiresAt"]), ); - expect(store.getDatabase().getSchemaVersion()).toBe(114); + expect(store.getDatabase().getSchemaVersion()).toBe(115); }); it("migrates a legacy v88 database and preserves task rows", async () => { diff --git a/packages/core/src/__tests__/store-workflow-runtime.test.ts b/packages/core/src/__tests__/store-workflow-runtime.test.ts new file mode 100644 index 0000000000..0bee6ddf5f --- /dev/null +++ b/packages/core/src/__tests__/store-workflow-runtime.test.ts @@ -0,0 +1,232 @@ +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { mkdtempSync } from "node:fs"; +import { rm } from "node:fs/promises"; +import { join } from "node:path"; +import { tmpdir } from "node:os"; +import { SCHEMA_VERSION } from "../db.js"; +import { TaskStore } from "../store.js"; + +function makeTmpDir(): string { + return mkdtempSync(join(tmpdir(), "kb-workflow-runtime-test-")); +} + +describe("TaskStore workflow work items", () => { + let rootDir: string; + let globalDir: string; + let store: TaskStore; + + beforeEach(async () => { + rootDir = makeTmpDir(); + globalDir = join(rootDir, ".fusion-global"); + store = new TaskStore(rootDir, globalDir); + await store.init(); + }); + + afterEach(async () => { + store.close(); + await rm(rootDir, { recursive: true, force: true, maxRetries: 5, retryDelay: 50 }); + }); + + async function createTaskId(): Promise { + const task = await store.createTask({ description: "workflow work item test" }); + return task.id; + } + + it("creates workflow work-item tables on fresh schema", () => { + const db = store.getDatabase(); + const table = db + .prepare("SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'workflow_work_items'") + .get() as { name: string } | undefined; + const indexes = db + .prepare("SELECT name FROM sqlite_master WHERE type = 'index' AND tbl_name = 'workflow_work_items' ORDER BY name") + .all() as Array<{ name: string }>; + + expect(table).toEqual({ name: "workflow_work_items" }); + expect(indexes.map((row) => row.name)).toEqual( + expect.arrayContaining([ + "idx_workflow_work_items_due", + "idx_workflow_work_items_leaseExpiresAt", + "idx_workflow_work_items_task_run", + ]), + ); + expect(db.getSchemaVersion()).toBe(SCHEMA_VERSION); + }); + + it("upserts by run, task, node, and kind without duplicating work", async () => { + const taskId = await createTaskId(); + + const created = store.upsertWorkflowWorkItem({ + runId: "run-1", + taskId, + nodeId: "merge.node", + kind: "merge", + now: "2026-06-09T00:00:00.000Z", + }); + const updated = store.upsertWorkflowWorkItem({ + runId: "run-1", + taskId, + nodeId: "merge.node", + kind: "merge", + state: "held", + blockedReason: "shared branch is assembling", + now: "2026-06-09T00:00:01.000Z", + }); + + expect(updated).toMatchObject({ + id: created.id, + runId: "run-1", + taskId, + nodeId: "merge.node", + kind: "merge", + state: "held", + attempt: 0, + blockedReason: "shared branch is assembling", + }); + + const rows = store + .getDatabase() + .prepare("SELECT COUNT(*) AS count FROM workflow_work_items WHERE runId = ? AND taskId = ?") + .get("run-1", taskId) as { count: number }; + expect(rows.count).toBe(1); + }); + + it("lists due runnable and retrying work independently of task column", async () => { + const taskId = await createTaskId(); + await store.moveTask(taskId, "todo"); + await store.moveTask(taskId, "in-progress"); + await store.moveTask(taskId, "in-review"); + const now = "2026-06-09T00:00:00.000Z"; + + const runnable = store.upsertWorkflowWorkItem({ + runId: "run-1", + taskId, + nodeId: "plan.node", + kind: "task", + state: "runnable", + now, + }); + const futureRetry = store.upsertWorkflowWorkItem({ + runId: "run-1", + taskId, + nodeId: "retry.node", + kind: "retry", + state: "retrying", + retryAfter: "2026-06-09T00:05:00.000Z", + now, + }); + store.upsertWorkflowWorkItem({ + runId: "run-1", + taskId, + nodeId: "hold.node", + kind: "manual-hold", + state: "held", + now, + }); + + expect(store.listDueWorkflowWorkItems({ now }).map((item) => item.id)).toEqual([runnable.id]); + expect(store.listDueWorkflowWorkItems({ now: "2026-06-09T00:05:00.000Z" }).map((item) => item.id)).toEqual([ + runnable.id, + futureRetry.id, + ]); + }); + + it("acquires due leases and exposes expired running leases for reclaim", async () => { + const taskId = await createTaskId(); + const item = store.upsertWorkflowWorkItem({ + runId: "run-lease", + taskId, + nodeId: "merge.node", + kind: "merge", + state: "runnable", + now: "2026-06-09T00:00:00.000Z", + }); + + const leased = store.acquireWorkflowWorkItemLease(item.id, "worker-a", { + now: "2026-06-09T00:00:00.000Z", + leaseDurationMs: 60_000, + }); + expect(leased).toMatchObject({ + id: item.id, + state: "running", + leaseOwner: "worker-a", + leaseExpiresAt: "2026-06-09T00:01:00.000Z", + }); + + expect( + store.acquireWorkflowWorkItemLease(item.id, "worker-b", { + now: "2026-06-09T00:00:30.000Z", + leaseDurationMs: 60_000, + }), + ).toBeNull(); + expect(store.listDueWorkflowWorkItems({ now: "2026-06-09T00:00:30.000Z" })).toEqual([]); + + expect(store.listDueWorkflowWorkItems({ now: "2026-06-09T00:01:00.000Z" }).map((due) => due.id)).toEqual([item.id]); + const reclaimed = store.acquireWorkflowWorkItemLease(item.id, "worker-b", { + now: "2026-06-09T00:01:00.000Z", + leaseDurationMs: 60_000, + }); + expect(reclaimed).toMatchObject({ + id: item.id, + state: "running", + leaseOwner: "worker-b", + leaseExpiresAt: "2026-06-09T00:02:00.000Z", + }); + }); + + it("preserves lease and retry metadata on idempotent duplicate upserts", async () => { + const taskId = await createTaskId(); + const item = store.upsertWorkflowWorkItem({ + runId: "run-idempotent", + taskId, + nodeId: "retry.node", + kind: "retry", + state: "retrying", + retryAfter: "2026-06-09T00:05:00.000Z", + leaseOwner: "worker-a", + leaseExpiresAt: "2026-06-09T00:06:00.000Z", + lastError: "temporary failure", + now: "2026-06-09T00:00:00.000Z", + }); + + const duplicate = store.upsertWorkflowWorkItem({ + runId: "run-idempotent", + taskId, + nodeId: "retry.node", + kind: "retry", + now: "2026-06-09T00:01:00.000Z", + }); + + expect(duplicate).toMatchObject({ + id: item.id, + state: "retrying", + retryAfter: "2026-06-09T00:05:00.000Z", + leaseOwner: "worker-a", + leaseExpiresAt: "2026-06-09T00:06:00.000Z", + lastError: "temporary failure", + updatedAt: "2026-06-09T00:01:00.000Z", + }); + }); + + it("does not requeue terminal work", async () => { + const taskId = await createTaskId(); + const item = store.upsertWorkflowWorkItem({ + runId: "run-terminal", + taskId, + nodeId: "merge.node", + kind: "merge", + state: "runnable", + }); + + store.transitionWorkflowWorkItem(item.id, "succeeded", { now: "2026-06-09T00:00:01.000Z" }); + + expect(() => + store.upsertWorkflowWorkItem({ + runId: "run-terminal", + taskId, + nodeId: "merge.node", + kind: "merge", + state: "runnable", + }), + ).toThrow(/terminal \(succeeded\) and cannot be requeued as runnable/); + }); +}); diff --git a/packages/core/src/__tests__/task-documents.test.ts b/packages/core/src/__tests__/task-documents.test.ts index b7b71047c0..af7c770510 100644 --- a/packages/core/src/__tests__/task-documents.test.ts +++ b/packages/core/src/__tests__/task-documents.test.ts @@ -51,7 +51,7 @@ describe("TaskStore task documents", () => { expect(tableNames.has("task_documents")).toBe(true); expect(tableNames.has("task_document_revisions")).toBe(true); - expect(db.getSchemaVersion()).toBe(114); + expect(db.getSchemaVersion()).toBe(115); const index = db .prepare( diff --git a/packages/core/src/db.ts b/packages/core/src/db.ts index eb50c0ae20..25b640cabe 100644 --- a/packages/core/src/db.ts +++ b/packages/core/src/db.ts @@ -162,7 +162,7 @@ export function isFts5CorruptionError(error: unknown): boolean { // ── Schema Definition ──────────────────────────────────────────────── -const SCHEMA_VERSION = 114; +const SCHEMA_VERSION = 115; const TASKS_FTS_AUTOMERGE = 8; const TASKS_FTS_CRISISMERGE = 16; @@ -602,6 +602,27 @@ CREATE TABLE IF NOT EXISTS completion_handoff_markers ( ); CREATE INDEX IF NOT EXISTS idx_completion_handoff_markers_acceptedAt ON completion_handoff_markers(acceptedAt); +CREATE TABLE IF NOT EXISTS workflow_work_items ( + id TEXT PRIMARY KEY, + runId TEXT NOT NULL, + taskId TEXT NOT NULL REFERENCES tasks(id) ON DELETE CASCADE, + nodeId TEXT NOT NULL, + kind TEXT NOT NULL, + state TEXT NOT NULL, + attempt INTEGER NOT NULL DEFAULT 0, + retryAfter TEXT, + leaseOwner TEXT, + leaseExpiresAt TEXT, + lastError TEXT, + blockedReason TEXT, + createdAt TEXT NOT NULL, + updatedAt TEXT NOT NULL, + UNIQUE(runId, taskId, nodeId, kind) +); +CREATE INDEX IF NOT EXISTS idx_workflow_work_items_due ON workflow_work_items(state, retryAfter, createdAt); +CREATE INDEX IF NOT EXISTS idx_workflow_work_items_leaseExpiresAt ON workflow_work_items(leaseExpiresAt); +CREATE INDEX IF NOT EXISTS idx_workflow_work_items_task_run ON workflow_work_items(taskId, runId); + -- Per-branch run state for concurrent workflow fan-out/join (U13, KTD-11/R21). -- Reconstructible per ADR-0001: a crashed parallel run resumes each branch from -- its persisted node; completed branches are not re-run. Additive-only. @@ -4634,6 +4655,40 @@ export class Database { }); } + // Migration 115: Workflow-owned merge/retry/scheduling S1. + // Adds durable workflow work items so runnable, held, retrying, merge, + // manual-hold, and recovery work can be claimed generically before legacy + // merge queue and retry policy are deleted. + if (version < 115) { + this.applyMigration(115, () => { + this.db.exec(` + CREATE TABLE IF NOT EXISTS workflow_work_items ( + id TEXT PRIMARY KEY, + runId TEXT NOT NULL, + taskId TEXT NOT NULL REFERENCES tasks(id) ON DELETE CASCADE, + nodeId TEXT NOT NULL, + kind TEXT NOT NULL, + state TEXT NOT NULL, + attempt INTEGER NOT NULL DEFAULT 0, + retryAfter TEXT, + leaseOwner TEXT, + leaseExpiresAt TEXT, + lastError TEXT, + blockedReason TEXT, + createdAt TEXT NOT NULL, + updatedAt TEXT NOT NULL, + UNIQUE(runId, taskId, nodeId, kind) + ); + CREATE INDEX IF NOT EXISTS idx_workflow_work_items_due + ON workflow_work_items(state, retryAfter, createdAt); + CREATE INDEX IF NOT EXISTS idx_workflow_work_items_leaseExpiresAt + ON workflow_work_items(leaseExpiresAt); + CREATE INDEX IF NOT EXISTS idx_workflow_work_items_task_run + ON workflow_work_items(taskId, runId); + `); + }); + } + } /** diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 326972e0d8..4b7d074933 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -1,5 +1,5 @@ -export { COLUMNS, DEFAULT_COLUMN, isColumn, normalizeColumn, COLUMN_LABELS, COLUMN_DESCRIPTIONS, VALID_TRANSITIONS, DEFAULT_SETTINGS, DEFAULT_GLOBAL_SETTINGS, DEFAULT_PROJECT_SETTINGS, GLOBAL_SETTINGS_KEYS, PROJECT_SETTINGS_KEYS, isGlobalSettingsKey, isProjectSettingsKey, isMergeRequestContractShadowEnabled, resolvePersistAgentThinkingLog, THINKING_LEVELS, THEME_MODES, COLOR_THEMES, SUPPORTED_LOCALES, DEFAULT_LOCALE, isLocale, WORKFLOW_STEP_TEMPLATES, AGENT_PERMISSIONS, PERMANENT_AGENT_ACTION_CATEGORIES, AGENT_PERMISSION_POLICY_ACTION_CATEGORIES, AGENT_PROVISIONING_APPROVAL_MODES, SANDBOX_PROVISIONING_APPROVAL_MODES, AGENT_PERMISSION_POLICY_PRESET_IDS, LEGACY_AGENT_PERMISSION_POLICY_ACTION_CATEGORY_ALIASES, APPROVAL_REQUEST_STATUSES, APPROVAL_REQUEST_AUDIT_EVENT_TYPES, normalizeApprovalRequestActionCategory, isValidApprovalRequestTransition, agentToConfigSnapshot, diffConfigSnapshots, isEphemeralAgent, hasAgentIdentity, CheckoutConflictError, DEFAULT_HEARTBEAT_PROCEDURE_PATH, getDefaultHeartbeatProcedurePath, EXECUTION_MODES, DEFAULT_EXECUTION_MODE, TASK_PRIORITIES, DEFAULT_TASK_PRIORITY, HIGH_FANOUT_BLOCKER_TODO_THRESHOLD, STALE_HIGH_FANOUT_BLOCKER_AGE_THRESHOLD_MS, DASHBOARD_USER_ID, normalizeMessageParticipant, validateMessageMetadata, validateDockerNodeConfig, sanitizeDockerNodeConfigForResponse, normalizeMergeIntegrationWorktreeMode, normalizeMergeAdvanceAutoSyncMode, MERGE_ADVANCE_AUTO_SYNC_MODES, normalizeMergeConflictStrategy, normalizeMergeStrategyOverlapBehavior, normalizePostMergeAuditMode, POST_MERGE_AUDIT_MODES, normalizeMergeAuditAutoRecovery, MERGE_AUDIT_AUTO_RECOVERY_MODES, normalizeMergerMode, MERGER_MODES, normalizeAutoRecovery, AUTO_RECOVERY_MODES, buildResearchDocumentKey, REPO_OVERRIDE_RE, SHARED_STATE_SNAPSHOT_VERSION, sanitizeCliAgentSettings, sanitizeCliAgentsSettings, CLI_AGENT_ADAPTER_IDS, CLI_AGENT_AUTONOMY_MODES } from "./types.js"; -export type { Column, ColumnId, IssueInfo, IssueState, TaskSourceIssue, PrInfo, PrConflictState, PrConflictDiagnostics, PrCheckState, PrCheckStatus, PrStatus, BranchGroup, BranchGroupCreateInput, BranchGroupUpdate, BranchGroupPrState, Task, TaskTokenUsage, TaskAttachment, TaskComment, TaskCommentInput, TaskDocument, TaskDocumentRevision, TaskDocumentCreateInput, TaskDocumentWithTask, TaskCreateInput, MeshReplicatedTaskCreatePayload, MeshReplicatedTaskApplyResult, TaskSource, SourceType, TaskDetail, RetrySummary, InboxTask, TodoList, TodoItem, TodoListCreateInput, TodoListUpdateInput, TodoItemCreateInput, TodoItemUpdateInput, TodoListWithItems, AgentLogEntry, AgentLogType, AgentRole, BoardConfig, DistributedTaskIdReserveInput, DistributedTaskIdReserveResult, DistributedTaskIdCommitInput, DistributedTaskIdCommitResult, DistributedTaskIdAbortInput, DistributedTaskIdAbortResult, DistributedTaskIdStateInput, DistributedTaskIdStateResult, AutostashOrphanRecord, AutostashOutcome, MergeDetails, MergeResult, MergeIntegrationWorktreeMode, MergeAdvanceAutoSyncMode, MergeConflictStrategy, CanonicalMergeConflictStrategy, MergeStrategyOverlapBehavior, PostMergeAuditMode, MergeAuditAutoRecoveryMode, MergerMode, MergerSettings, AutoRecoveryMode, AutoRecoveryFailureClass, AutoRecoverySettings, DirectMergeCommitStrategy, Settings, GlobalSettings, ProjectSettings, SecretsEnvConfig, WebSearchBackend, ResearchEnabledSources, ResearchGlobalDefaults, ResearchProjectLimits, ResearchProjectSettings, SandboxBackendName, SandboxFailureMode, SandboxPolicy, SandboxProjectSettings, EvalFollowUpPolicy, EvalProjectSettings, ResolvedEvalSettings, SettingsScope, DaemonTokenSettings, TaskStep, StepStatus, TaskLogEntry, RunMutationContext, ActivityLogEntry, ActivityEventType, ThinkingLevel, ThemeMode, ColorTheme, Locale, ExecutionMode, TaskPriority, MergeQueueEntry, MergeQueueEnqueueOptions, MergeQueueAcquireOptions, MergeQueueReleaseOutcome, MergeRequestState, MergeRequestRecord, CompletionHandoffMarker, HandoffEvidence, HandoffToReviewOptions, UnavailableNodePolicy, OwningNodeHandoffPolicy, PlanningQuestion, PlanningSummary, PlanningResponse, PlanningQuestionType, ArchivedTaskEntry, BatchStatusRequest, BatchStatusResponse, BatchStatusEntry, BatchStatusResult, GithubIssueAction, ModelPreset, WorkflowStep, WorkflowStepMode, WorkflowStepGateMode, WorkflowStepPhase, WorkflowStepInput, WorkflowStepResult, WorkflowStepTemplate, Agent, OrgTreeNode, AgentState, AgentDetail, AgentCreateInput, AgentUpdateInput, AgentApiKey, AgentApiKeyCreateResult, AgentCapability, AgentPromptTemplate, AgentPromptsConfig, AgentPermission, PermanentAgentActionCategory, PermanentAgentSensitiveActionCategory, PermanentAgentGatingContext, AgentPermissionPolicy, AgentPermissionPolicyRules, AgentPermissionPolicyActionCategory, AgentProvisioningApprovalMode, SandboxProvisioningApprovalMode, LegacyAgentPermissionPolicyActionCategory, ApprovalRequestActionCategoryInput, ApprovalRequestActionCategory, AgentPermissionPolicyDisposition, AgentPermissionPolicyPresetId, ApprovalRequestStatus, ApprovalRequestAuditEventType, ApprovalRequestActorSnapshot, ApprovalRequestTargetAction, ApprovalRequestAuditEvent, ApprovalRequest, ApprovalRequestCreateInput, ApprovalRequestDecisionInput, ApprovalRequestCompletionInput, ApprovalRequestListInput, TaskAssignSource, AgentAccessState, AgentHeartbeatConfig, AgentBudgetConfig, AgentBudgetStatus, InstructionsBundleConfig, MessageResponseMode, AgentHeartbeatEvent, AgentHeartbeatRun, BlockedStateSnapshot, HeartbeatInvocationSource, AgentTaskSession, AgentRating, AgentRatingSummary, AgentRatingInput, AgentConfigSnapshot, RevisionFieldDiff, AgentConfigRevision, AgentStats, ReflectionTrigger, ReflectionMetrics, AgentReflection, AgentPerformanceSummary, NtfyNotificationEvent, NotificationEvent, NotificationPayload, NotificationProviderConfig, CustomProvider, SteeringComment, ParticipantType, MessageType, Message, MessageCreateInput, MessageFilter, MessageMetadata, MessageReplyReference, Mailbox, CheckoutLease, CheckoutClaimPrecondition, TaskClaimRow, CentralClaimStore, RunAuditDomain, RunAuditEvent, RunAuditEventInput, RunAuditEventFilter, AgentMemoryInclusionMode, HeartbeatPromptTemplate, HeartbeatScopeDisciplineMode, WorktrunkSettings, WorktrunkOnFailure, TaskBranchContext, CliAgentSettings } from "./types.js"; +export { COLUMNS, DEFAULT_COLUMN, isColumn, normalizeColumn, COLUMN_LABELS, COLUMN_DESCRIPTIONS, VALID_TRANSITIONS, DEFAULT_SETTINGS, DEFAULT_GLOBAL_SETTINGS, DEFAULT_PROJECT_SETTINGS, GLOBAL_SETTINGS_KEYS, PROJECT_SETTINGS_KEYS, isGlobalSettingsKey, isProjectSettingsKey, isMergeRequestContractShadowEnabled, resolvePersistAgentThinkingLog, THINKING_LEVELS, THEME_MODES, COLOR_THEMES, SUPPORTED_LOCALES, DEFAULT_LOCALE, isLocale, WORKFLOW_STEP_TEMPLATES, AGENT_PERMISSIONS, PERMANENT_AGENT_ACTION_CATEGORIES, AGENT_PERMISSION_POLICY_ACTION_CATEGORIES, AGENT_PROVISIONING_APPROVAL_MODES, SANDBOX_PROVISIONING_APPROVAL_MODES, AGENT_PERMISSION_POLICY_PRESET_IDS, LEGACY_AGENT_PERMISSION_POLICY_ACTION_CATEGORY_ALIASES, APPROVAL_REQUEST_STATUSES, APPROVAL_REQUEST_AUDIT_EVENT_TYPES, normalizeApprovalRequestActionCategory, isValidApprovalRequestTransition, agentToConfigSnapshot, diffConfigSnapshots, isEphemeralAgent, hasAgentIdentity, CheckoutConflictError, DEFAULT_HEARTBEAT_PROCEDURE_PATH, getDefaultHeartbeatProcedurePath, EXECUTION_MODES, DEFAULT_EXECUTION_MODE, TASK_PRIORITIES, DEFAULT_TASK_PRIORITY, WORKFLOW_WORK_ITEM_KINDS, WORKFLOW_WORK_ITEM_STATES, HIGH_FANOUT_BLOCKER_TODO_THRESHOLD, STALE_HIGH_FANOUT_BLOCKER_AGE_THRESHOLD_MS, DASHBOARD_USER_ID, normalizeMessageParticipant, validateMessageMetadata, validateDockerNodeConfig, sanitizeDockerNodeConfigForResponse, normalizeMergeIntegrationWorktreeMode, normalizeMergeAdvanceAutoSyncMode, MERGE_ADVANCE_AUTO_SYNC_MODES, normalizeMergeConflictStrategy, normalizeMergeStrategyOverlapBehavior, normalizePostMergeAuditMode, POST_MERGE_AUDIT_MODES, normalizeMergeAuditAutoRecovery, MERGE_AUDIT_AUTO_RECOVERY_MODES, normalizeMergerMode, MERGER_MODES, normalizeAutoRecovery, AUTO_RECOVERY_MODES, buildResearchDocumentKey, REPO_OVERRIDE_RE, SHARED_STATE_SNAPSHOT_VERSION, sanitizeCliAgentSettings, sanitizeCliAgentsSettings, CLI_AGENT_ADAPTER_IDS, CLI_AGENT_AUTONOMY_MODES } from "./types.js"; +export type { Column, ColumnId, IssueInfo, IssueState, TaskSourceIssue, PrInfo, PrConflictState, PrConflictDiagnostics, PrCheckState, PrCheckStatus, PrStatus, BranchGroup, BranchGroupCreateInput, BranchGroupUpdate, BranchGroupPrState, Task, TaskTokenUsage, TaskAttachment, TaskComment, TaskCommentInput, TaskDocument, TaskDocumentRevision, TaskDocumentCreateInput, TaskDocumentWithTask, TaskCreateInput, MeshReplicatedTaskCreatePayload, MeshReplicatedTaskApplyResult, TaskSource, SourceType, TaskDetail, RetrySummary, InboxTask, TodoList, TodoItem, TodoListCreateInput, TodoListUpdateInput, TodoItemCreateInput, TodoItemUpdateInput, TodoListWithItems, AgentLogEntry, AgentLogType, AgentRole, BoardConfig, DistributedTaskIdReserveInput, DistributedTaskIdReserveResult, DistributedTaskIdCommitInput, DistributedTaskIdCommitResult, DistributedTaskIdAbortInput, DistributedTaskIdAbortResult, DistributedTaskIdStateInput, DistributedTaskIdStateResult, AutostashOrphanRecord, AutostashOutcome, MergeDetails, MergeResult, MergeIntegrationWorktreeMode, MergeAdvanceAutoSyncMode, MergeConflictStrategy, CanonicalMergeConflictStrategy, MergeStrategyOverlapBehavior, PostMergeAuditMode, MergeAuditAutoRecoveryMode, MergerMode, MergerSettings, AutoRecoveryMode, AutoRecoveryFailureClass, AutoRecoverySettings, DirectMergeCommitStrategy, Settings, GlobalSettings, ProjectSettings, SecretsEnvConfig, WebSearchBackend, ResearchEnabledSources, ResearchGlobalDefaults, ResearchProjectLimits, ResearchProjectSettings, SandboxBackendName, SandboxFailureMode, SandboxPolicy, SandboxProjectSettings, EvalFollowUpPolicy, EvalProjectSettings, ResolvedEvalSettings, SettingsScope, DaemonTokenSettings, TaskStep, StepStatus, TaskLogEntry, RunMutationContext, ActivityLogEntry, ActivityEventType, ThinkingLevel, ThemeMode, ColorTheme, Locale, ExecutionMode, TaskPriority, MergeQueueEntry, MergeQueueEnqueueOptions, MergeQueueAcquireOptions, MergeQueueReleaseOutcome, MergeRequestState, MergeRequestRecord, CompletionHandoffMarker, WorkflowWorkItem, WorkflowWorkItemDueFilter, WorkflowWorkItemKind, WorkflowWorkItemState, WorkflowWorkItemTransitionPatch, WorkflowWorkItemUpsertInput, HandoffEvidence, HandoffToReviewOptions, UnavailableNodePolicy, OwningNodeHandoffPolicy, PlanningQuestion, PlanningSummary, PlanningResponse, PlanningQuestionType, ArchivedTaskEntry, BatchStatusRequest, BatchStatusResponse, BatchStatusEntry, BatchStatusResult, GithubIssueAction, ModelPreset, WorkflowStep, WorkflowStepMode, WorkflowStepGateMode, WorkflowStepPhase, WorkflowStepInput, WorkflowStepResult, WorkflowStepTemplate, Agent, OrgTreeNode, AgentState, AgentDetail, AgentCreateInput, AgentUpdateInput, AgentApiKey, AgentApiKeyCreateResult, AgentCapability, AgentPromptTemplate, AgentPromptsConfig, AgentPermission, PermanentAgentActionCategory, PermanentAgentSensitiveActionCategory, PermanentAgentGatingContext, AgentPermissionPolicy, AgentPermissionPolicyRules, AgentPermissionPolicyActionCategory, AgentProvisioningApprovalMode, SandboxProvisioningApprovalMode, LegacyAgentPermissionPolicyActionCategory, ApprovalRequestActionCategoryInput, ApprovalRequestActionCategory, AgentPermissionPolicyDisposition, AgentPermissionPolicyPresetId, ApprovalRequestStatus, ApprovalRequestAuditEventType, ApprovalRequestActorSnapshot, ApprovalRequestTargetAction, ApprovalRequestAuditEvent, ApprovalRequest, ApprovalRequestCreateInput, ApprovalRequestDecisionInput, ApprovalRequestCompletionInput, ApprovalRequestListInput, TaskAssignSource, AgentAccessState, AgentHeartbeatConfig, AgentBudgetConfig, AgentBudgetStatus, InstructionsBundleConfig, MessageResponseMode, AgentHeartbeatEvent, AgentHeartbeatRun, BlockedStateSnapshot, HeartbeatInvocationSource, AgentTaskSession, AgentRating, AgentRatingSummary, AgentRatingInput, AgentConfigSnapshot, RevisionFieldDiff, AgentConfigRevision, AgentStats, ReflectionTrigger, ReflectionMetrics, AgentReflection, AgentPerformanceSummary, NtfyNotificationEvent, NotificationEvent, NotificationPayload, NotificationProviderConfig, CustomProvider, SteeringComment, ParticipantType, MessageType, Message, MessageCreateInput, MessageFilter, MessageMetadata, MessageReplyReference, Mailbox, CheckoutLease, CheckoutClaimPrecondition, TaskClaimRow, CentralClaimStore, RunAuditDomain, RunAuditEvent, RunAuditEventInput, RunAuditEventFilter, AgentMemoryInclusionMode, HeartbeatPromptTemplate, HeartbeatScopeDisciplineMode, WorktrunkSettings, WorktrunkOnFailure, TaskBranchContext, CliAgentSettings } from "./types.js"; export { AGENT_VALID_TRANSITIONS, DUPLICATE_OF_METADATA_KEY } from "./types.js"; export { resolveEntryPointBranchAssignment, diff --git a/packages/core/src/store.ts b/packages/core/src/store.ts index 7dff4a89dd..a7556d52b4 100644 --- a/packages/core/src/store.ts +++ b/packages/core/src/store.ts @@ -3,7 +3,7 @@ import { randomUUID } from "node:crypto"; import { mkdir, readdir, readFile, writeFile, rename, unlink } from "node:fs/promises"; import { join } from "node:path"; import { existsSync, watch, type FSWatcher } from "node:fs"; -import type { Task, TaskDetail, TaskCreateInput, TaskAttachment, AgentLogEntry, BoardConfig, Column, ColumnId, CheckoutClaimPrecondition, MergeResult, Settings, GlobalSettings, ProjectSettings, ActivityLogEntry, ActivityEventType, TaskDocument, TaskDocumentRevision, TaskDocumentCreateInput, TaskDocumentWithTask, InboxTask, TaskLogEntry, RunMutationContext, RunAuditEvent, RunAuditEventInput, RunAuditEventFilter, ArchivedTaskEntry, ArchiveAgentLogMode, TaskPriority, SourceType, WorkflowStepTemplate, Agent, AutostashOrphanRecord, TaskCommitAssociation, TaskCommitAssociationMatchSource, TaskCommitAssociationConfidence, GithubIssueAction, MergeQueueEntry, MergeQueueEnqueueOptions, MergeQueueAcquireOptions, MergeQueueReleaseOutcome, HandoffToReviewOptions, GoalCitation, GoalCitationFilter, GoalCitationInput, GoalCitationSurface, BranchGroup, BranchGroupCreateInput, BranchGroupUpdate, TaskBranchAssignmentMode, MergeRequestRecord, MergeRequestState, CompletionHandoffMarker, PrEntity, PrEntityCreateInput, PrEntityUpdate, PrEntityState, PrThreadState, PrThreadOutcome, PrConflictState, PrChecksRollup, PrReviewDecision } from "./types.js"; +import type { Task, TaskDetail, TaskCreateInput, TaskAttachment, AgentLogEntry, BoardConfig, Column, ColumnId, CheckoutClaimPrecondition, MergeResult, Settings, GlobalSettings, ProjectSettings, ActivityLogEntry, ActivityEventType, TaskDocument, TaskDocumentRevision, TaskDocumentCreateInput, TaskDocumentWithTask, InboxTask, TaskLogEntry, RunMutationContext, RunAuditEvent, RunAuditEventInput, RunAuditEventFilter, ArchivedTaskEntry, ArchiveAgentLogMode, TaskPriority, SourceType, WorkflowStepTemplate, Agent, AutostashOrphanRecord, TaskCommitAssociation, TaskCommitAssociationMatchSource, TaskCommitAssociationConfidence, GithubIssueAction, MergeQueueEntry, MergeQueueEnqueueOptions, MergeQueueAcquireOptions, MergeQueueReleaseOutcome, HandoffToReviewOptions, GoalCitation, GoalCitationFilter, GoalCitationInput, GoalCitationSurface, BranchGroup, BranchGroupCreateInput, BranchGroupUpdate, TaskBranchAssignmentMode, MergeRequestRecord, MergeRequestState, CompletionHandoffMarker, WorkflowWorkItem, WorkflowWorkItemDueFilter, WorkflowWorkItemKind, WorkflowWorkItemState, WorkflowWorkItemTransitionPatch, WorkflowWorkItemUpsertInput, PrEntity, PrEntityCreateInput, PrEntityUpdate, PrEntityState, PrThreadState, PrThreadOutcome, PrConflictState, PrChecksRollup, PrReviewDecision } from "./types.js"; import { createActivityLogSnapshot, createRunAuditSnapshot, createTaskMetadataSnapshot, toTaskMetadataRecord, validateSnapshotEnvelope, type ActivityLogSnapshot, type RunAuditSnapshot, type TaskMetadataSnapshot } from "./shared-mesh-state.js"; import { VALID_TRANSITIONS, COLUMNS, DEFAULT_SETTINGS, isGlobalOnlySettingsKey, WORKFLOW_STEP_TEMPLATES, validateDocumentKey } from "./types.js"; import { DEFAULT_PROJECT_SETTINGS } from "./settings-schema.js"; @@ -618,6 +618,23 @@ interface CompletionHandoffMarkerRow { source: string; } +interface WorkflowWorkItemRow { + id: string; + runId: string; + taskId: string; + nodeId: string; + kind: string; + state: string; + attempt: number; + retryAfter: string | null; + leaseOwner: string | null; + leaseExpiresAt: string | null; + lastError: string | null; + blockedReason: string | null; + createdAt: string; + updatedAt: string; +} + /** Database row shape for the config table. */ interface ConfigRow { nextId: number; @@ -8793,6 +8810,59 @@ ${TASK_UPSERT_SQL_ASSIGNMENTS} }; } + private normalizeWorkflowWorkItemKind(value: string): WorkflowWorkItemKind { + switch (value) { + case "task": + case "merge": + case "retry": + case "manual-hold": + case "recovery": + return value; + default: + return "task"; + } + } + + private normalizeWorkflowWorkItemState(value: string): WorkflowWorkItemState { + switch (value) { + case "runnable": + case "running": + case "held": + case "retrying": + case "manual-required": + case "succeeded": + case "failed": + case "cancelled": + case "exhausted": + return value; + default: + return "runnable"; + } + } + + private isTerminalWorkflowWorkItemState(state: WorkflowWorkItemState): boolean { + return state === "succeeded" || state === "failed" || state === "cancelled" || state === "exhausted"; + } + + private rowToWorkflowWorkItem(row: WorkflowWorkItemRow): WorkflowWorkItem { + return { + id: row.id, + runId: row.runId, + taskId: row.taskId, + nodeId: row.nodeId, + kind: this.normalizeWorkflowWorkItemKind(row.kind), + state: this.normalizeWorkflowWorkItemState(row.state), + attempt: row.attempt, + retryAfter: row.retryAfter, + leaseOwner: row.leaseOwner, + leaseExpiresAt: row.leaseExpiresAt, + lastError: row.lastError, + blockedReason: row.blockedReason, + createdAt: row.createdAt, + updatedAt: row.updatedAt, + }; + } + private isValidMergeRequestTransition(from: MergeRequestState, to: MergeRequestState): boolean { if (from === to) return true; const allowed: Record> = { @@ -8882,6 +8952,190 @@ ${TASK_UPSERT_SQL_ASSIGNMENTS} return row ? this.rowToMergeRequestRecord(row) : null; } + upsertWorkflowWorkItem(input: WorkflowWorkItemUpsertInput): WorkflowWorkItem { + return this.db.transactionImmediate(() => { + const existing = this.db + .prepare("SELECT * FROM workflow_work_items WHERE runId = ? AND taskId = ? AND nodeId = ? AND kind = ?") + .get(input.runId, input.taskId, input.nodeId, input.kind) as WorkflowWorkItemRow | undefined; + const now = input.now ?? new Date().toISOString(); + const existingState = existing ? this.normalizeWorkflowWorkItemState(existing.state) : null; + const state = input.state ?? existingState ?? "runnable"; + if (existingState && this.isTerminalWorkflowWorkItemState(existingState) && existingState !== state) { + throw new Error( + `Workflow work item ${existing?.id ?? input.id ?? input.nodeId} is terminal (${existingState}) and cannot be requeued as ${state}`, + ); + } + + const id = existing?.id ?? input.id ?? randomUUID(); + this.db + .prepare( + `INSERT INTO workflow_work_items ( + id, runId, taskId, nodeId, kind, state, attempt, retryAfter, + leaseOwner, leaseExpiresAt, lastError, blockedReason, createdAt, updatedAt + ) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(runId, taskId, nodeId, kind) DO UPDATE SET + state = excluded.state, + attempt = excluded.attempt, + retryAfter = excluded.retryAfter, + leaseOwner = excluded.leaseOwner, + leaseExpiresAt = excluded.leaseExpiresAt, + lastError = excluded.lastError, + blockedReason = excluded.blockedReason, + updatedAt = excluded.updatedAt`, + ) + .run( + id, + input.runId, + input.taskId, + input.nodeId, + input.kind, + state, + input.attempt ?? existing?.attempt ?? 0, + input.retryAfter === undefined ? existing?.retryAfter ?? null : input.retryAfter, + input.leaseOwner === undefined ? existing?.leaseOwner ?? null : input.leaseOwner, + input.leaseExpiresAt === undefined ? existing?.leaseExpiresAt ?? null : input.leaseExpiresAt, + input.lastError === undefined ? existing?.lastError ?? null : input.lastError, + input.blockedReason === undefined ? existing?.blockedReason ?? null : input.blockedReason, + existing?.createdAt ?? now, + now, + ); + + const row = this.db.prepare("SELECT * FROM workflow_work_items WHERE id = ?").get(id) as WorkflowWorkItemRow | undefined; + if (!row) throw new Error(`Failed to upsert workflow work item ${id}`); + this.insertRunAuditEventRow({ + taskId: row.taskId, + runId: row.runId, + domain: "database", + mutationType: "workflowWorkItem:upsert", + target: row.id, + metadata: { id: row.id, nodeId: row.nodeId, kind: row.kind, state: row.state, attempt: row.attempt }, + }); + return this.rowToWorkflowWorkItem(row); + }); + } + + transitionWorkflowWorkItem( + id: string, + state: WorkflowWorkItemState, + patch: WorkflowWorkItemTransitionPatch = {}, + ): WorkflowWorkItem { + return this.db.transactionImmediate(() => { + const now = patch.now ?? new Date().toISOString(); + const existing = this.db.prepare("SELECT * FROM workflow_work_items WHERE id = ?").get(id) as WorkflowWorkItemRow | undefined; + if (!existing) throw new Error(`Workflow work item ${id} not found`); + const fromState = this.normalizeWorkflowWorkItemState(existing.state); + if (this.isTerminalWorkflowWorkItemState(fromState) && fromState !== state) { + throw new Error(`Workflow work item ${id} is terminal (${fromState}) and cannot transition to ${state}`); + } + + this.db + .prepare( + `UPDATE workflow_work_items + SET state = ?, + attempt = ?, + retryAfter = ?, + leaseOwner = ?, + leaseExpiresAt = ?, + lastError = ?, + blockedReason = ?, + updatedAt = ? + WHERE id = ?`, + ) + .run( + state, + patch.attempt ?? existing.attempt, + patch.retryAfter === undefined ? existing.retryAfter : patch.retryAfter, + patch.leaseOwner === undefined ? existing.leaseOwner : patch.leaseOwner, + patch.leaseExpiresAt === undefined ? existing.leaseExpiresAt : patch.leaseExpiresAt, + patch.lastError === undefined ? existing.lastError : patch.lastError, + patch.blockedReason === undefined ? existing.blockedReason : patch.blockedReason, + now, + id, + ); + + const updated = this.db.prepare("SELECT * FROM workflow_work_items WHERE id = ?").get(id) as WorkflowWorkItemRow | undefined; + if (!updated) throw new Error(`Workflow work item ${id} disappeared`); + this.insertRunAuditEventRow({ + taskId: updated.taskId, + runId: updated.runId, + domain: "database", + mutationType: "workflowWorkItem:transition", + target: updated.id, + metadata: { id: updated.id, fromState, toState: state, attempt: updated.attempt }, + }); + return this.rowToWorkflowWorkItem(updated); + }); + } + + getWorkflowWorkItem(id: string): WorkflowWorkItem | null { + const row = this.db.prepare("SELECT * FROM workflow_work_items WHERE id = ?").get(id) as WorkflowWorkItemRow | undefined; + return row ? this.rowToWorkflowWorkItem(row) : null; + } + + listDueWorkflowWorkItems(filter: WorkflowWorkItemDueFilter = {}): WorkflowWorkItem[] { + const now = filter.now ?? new Date().toISOString(); + const states = filter.states?.length ? filter.states : ["runnable", "retrying"]; + const conditions = [ + `((state IN (${states.map(() => "?").join(", ")}) AND (leaseExpiresAt IS NULL OR leaseExpiresAt <= ?)) OR (state = 'running' AND leaseExpiresAt IS NOT NULL AND leaseExpiresAt <= ?))`, + "(retryAfter IS NULL OR retryAfter <= ?)", + ]; + const params: unknown[] = [...states, now, now, now]; + if (filter.kinds?.length) { + conditions.push(`kind IN (${filter.kinds.map(() => "?").join(", ")})`); + params.push(...filter.kinds); + } + params.push(filter.limit ?? 100); + + const rows = this.db + .prepare( + `SELECT * + FROM workflow_work_items + WHERE ${conditions.join(" AND ")} + ORDER BY retryAfter IS NOT NULL, retryAfter ASC, createdAt ASC + LIMIT ?`, + ) + .all(...params) as WorkflowWorkItemRow[]; + return rows.map((row) => this.rowToWorkflowWorkItem(row)); + } + + acquireWorkflowWorkItemLease( + id: string, + leaseOwner: string, + opts: { leaseDurationMs: number; now?: string }, + ): WorkflowWorkItem | null { + return this.db.transactionImmediate(() => { + const now = opts.now ?? new Date().toISOString(); + const leaseExpiresAt = new Date(new Date(now).getTime() + opts.leaseDurationMs).toISOString(); + const result = this.db + .prepare( + `UPDATE workflow_work_items + SET state = 'running', + leaseOwner = ?, + leaseExpiresAt = ?, + updatedAt = ? + WHERE id = ? + AND state IN ('runnable', 'retrying', 'running') + AND (retryAfter IS NULL OR retryAfter <= ?) + AND (leaseExpiresAt IS NULL OR leaseExpiresAt <= ?)`, + ) + .run(leaseOwner, leaseExpiresAt, now, id, now, now); + if (result.changes === 0) return null; + + const row = this.db.prepare("SELECT * FROM workflow_work_items WHERE id = ?").get(id) as WorkflowWorkItemRow | undefined; + if (!row) throw new Error(`Workflow work item ${id} disappeared`); + this.insertRunAuditEventRow({ + taskId: row.taskId, + runId: row.runId, + domain: "database", + mutationType: "workflowWorkItem:lease-acquired", + target: row.id, + metadata: { id: row.id, leaseOwner: row.leaseOwner, leaseExpiresAt: row.leaseExpiresAt }, + }); + return this.rowToWorkflowWorkItem(row); + }); + } + setCompletionHandoffAcceptedMarker( taskId: string, opts: { source: string; acceptedAt?: string }, diff --git a/packages/core/src/types.ts b/packages/core/src/types.ts index 9ee6a8c411..a26d15a01a 100644 --- a/packages/core/src/types.ts +++ b/packages/core/src/types.ts @@ -86,6 +86,80 @@ export const MERGE_REQUEST_STATES = [ export type MergeRequestState = (typeof MERGE_REQUEST_STATES)[number]; +export const WORKFLOW_WORK_ITEM_KINDS = [ + "task", + "merge", + "retry", + "manual-hold", + "recovery", +] as const; + +export type WorkflowWorkItemKind = (typeof WORKFLOW_WORK_ITEM_KINDS)[number]; + +export const WORKFLOW_WORK_ITEM_STATES = [ + "runnable", + "running", + "held", + "retrying", + "manual-required", + "succeeded", + "failed", + "cancelled", + "exhausted", +] as const; + +export type WorkflowWorkItemState = (typeof WORKFLOW_WORK_ITEM_STATES)[number]; + +export interface WorkflowWorkItem { + id: string; + runId: string; + taskId: string; + nodeId: string; + kind: WorkflowWorkItemKind; + state: WorkflowWorkItemState; + attempt: number; + retryAfter: string | null; + leaseOwner: string | null; + leaseExpiresAt: string | null; + lastError: string | null; + blockedReason: string | null; + createdAt: string; + updatedAt: string; +} + +export interface WorkflowWorkItemUpsertInput { + id?: string; + runId: string; + taskId: string; + nodeId: string; + kind: WorkflowWorkItemKind; + state?: WorkflowWorkItemState; + attempt?: number; + retryAfter?: string | null; + leaseOwner?: string | null; + leaseExpiresAt?: string | null; + lastError?: string | null; + blockedReason?: string | null; + now?: string; +} + +export interface WorkflowWorkItemTransitionPatch { + attempt?: number; + retryAfter?: string | null; + leaseOwner?: string | null; + leaseExpiresAt?: string | null; + lastError?: string | null; + blockedReason?: string | null; + now?: string; +} + +export interface WorkflowWorkItemDueFilter { + now?: string; + limit?: number; + kinds?: WorkflowWorkItemKind[]; + states?: WorkflowWorkItemState[]; +} + export interface MergeQueueEntry { taskId: string; enqueuedAt: string; From c11f1700b7badd389cd66db38e0bf3e1a7aab9bc Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Tue, 9 Jun 2026 13:27:43 -0700 Subject: [PATCH 3/5] chore(workflow): plan stacked migration PRs --- ...e-workflow-owned-merge-stacked-prs-plan.md | 65 +++++++++++++++++++ 1 file changed, 65 insertions(+) create mode 100644 docs/plans/2026-06-09-004-chore-workflow-owned-merge-stacked-prs-plan.md diff --git a/docs/plans/2026-06-09-004-chore-workflow-owned-merge-stacked-prs-plan.md b/docs/plans/2026-06-09-004-chore-workflow-owned-merge-stacked-prs-plan.md new file mode 100644 index 0000000000..9ba73cf4a3 --- /dev/null +++ b/docs/plans/2026-06-09-004-chore-workflow-owned-merge-stacked-prs-plan.md @@ -0,0 +1,65 @@ +--- +title: "chore: Workflow-owned merge stacked PR creation" +type: chore +status: active +date: 2026-06-09 +depth: shallow +origin: docs/plans/2026-06-09-003-refactor-workflow-owned-merge-full-migration-slices-plan.md +--- + +# chore: Workflow-owned merge stacked PR creation + +## Summary + +Create a linear GitHub PR stack for the remaining workflow-owned merge, +retry, scheduling, recovery, projection, deletion, and release slices. The stack +does not claim future implementation is complete. Each PR carries a durable +slice handoff document and is opened as a draft against the previous slice +branch so reviewers can see ordering, dependency, and milestone intent. + +## Requirements Trace + +- R1. Every migration slice S0-S18 from the origin plan is represented in the + PR stack. +- R2. Existing PR #1571 remains the stack base for S0/S1. +- R3. Remaining slices S2-S18 each receive a dedicated branch and draft PR. +- R4. Each branch has a non-empty, reviewable diff that records the slice goal, + milestone, dependencies, file scope, tests, and exit gate. +- R5. PR bodies link back to the full migration plan and identify their base + branch so the stack is reconstructible. + +## Scope + +In scope: + +- Add `docs/plans/workflow-owned-merge-stack/sXX-*.md` handoff files. +- Create and push one branch per remaining slice. +- Open draft PRs stacked linearly from S2 through S18. +- Update PR #1571 with the complete slice/milestone list when needed. + +Out of scope: + +- Implementing S2-S18 code changes in this turn. +- Merging the stack. +- Rewriting existing PR #1571 commits. + +## Stack Shape + +- S0/S1: existing PR #1571, branch + `feature/workflow-owned-merge-retry-scheduling-plan`, base `main`. +- S2: base S0/S1 branch. +- S3-S18: each branch is based on the immediately preceding slice branch. + +This is intentionally linear even though the origin dependency graph has some +parallelizable edges. A linear stack gives GitHub a straightforward review and +landing path; implementation branches can still be split or rebased later if a +slice needs to move independently. + +## Verification + +- `git status --short --branch` is clean after all branches are pushed. +- `gh pr view` succeeds for each created PR. +- Every created PR body includes the slice number, milestone, dependency, full + plan link, and base branch. +- `gh pr checks` is inspected for the current stack base and any newly opened + PR checks that are immediately available. From cdd86fc3f62f1af361780a6245bebfea8bd7d457 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Tue, 9 Jun 2026 13:37:46 -0700 Subject: [PATCH 4/5] fix(review): address workflow work item feedback --- .../core/src/__tests__/db-migrate.test.ts | 10 ------- packages/core/src/__tests__/db.test.ts | 20 -------------- packages/core/src/__tests__/run-audit.test.ts | 2 +- .../__tests__/store-workflow-runtime.test.ts | 27 +++++++++++++++++++ packages/core/src/store.ts | 15 +++++++++-- 5 files changed, 41 insertions(+), 33 deletions(-) diff --git a/packages/core/src/__tests__/db-migrate.test.ts b/packages/core/src/__tests__/db-migrate.test.ts index 3b06532f27..b2f321eae4 100644 --- a/packages/core/src/__tests__/db-migrate.test.ts +++ b/packages/core/src/__tests__/db-migrate.test.ts @@ -716,7 +716,6 @@ describe("schema migration", () => { const row = db.prepare("SELECT deletedAt FROM tasks WHERE id = 'FN-legacy'").get() as { deletedAt: string | null }; expect(row.deletedAt).toBeNull(); expect(db.getSchemaVersion()).toBe(115); - expect(db.getSchemaVersion()).toBe(115); db.close(); }); @@ -750,7 +749,6 @@ describe("schema migration", () => { { id: "WS-002", mode: "script", gateMode: "advisory" }, ]); expect(db.getSchemaVersion()).toBe(115); - expect(db.getSchemaVersion()).toBe(115); db.close(); }); @@ -801,7 +799,6 @@ describe("schema migration", () => { reviewerFallbackRetryCount: 0, }); expect(db.getSchemaVersion()).toBe(115); - expect(db.getSchemaVersion()).toBe(115); db.close(); }); @@ -831,7 +828,6 @@ describe("schema migration", () => { const columns = db.prepare("PRAGMA table_info(milestones)").all() as Array<{ name: string }>; expect(columns.map((column) => column.name)).toContain("acceptanceCriteria"); expect(db.getSchemaVersion()).toBe(115); - expect(db.getSchemaVersion()).toBe(115); db.close(); }); @@ -872,7 +868,6 @@ describe("schema migration", () => { const missionColumns = db.prepare("PRAGMA table_info(missions)").all() as Array<{ name: string }>; expect(missionColumns.map((column) => column.name)).toContain("autoMerge"); - expect(db.getSchemaVersion()).toBe(115); expect(db.getSchemaVersion()).toBe(115); db.close(); }); @@ -908,7 +903,6 @@ describe("schema migration", () => { { id: "WS-003", mode: "prompt", enabled: 0, gateMode: "advisory" }, ]); expect(db.getSchemaVersion()).toBe(115); - expect(db.getSchemaVersion()).toBe(115); db.close(); }); @@ -946,7 +940,6 @@ describe("schema migration", () => { const indexes = db.prepare("PRAGMA index_list(mission_goals)").all() as Array<{ name: string }>; expect(indexes.some((index) => index.name === "idxMissionGoalsGoalId")).toBe(true); expect(db.getSchemaVersion()).toBe(115); - expect(db.getSchemaVersion()).toBe(115); db.close(); }); @@ -1226,7 +1219,6 @@ describe("schema migration", () => { .get() as { migrated_fragment_id: string | null }; expect(stepRow.migrated_fragment_id).toBeNull(); - expect(db.getSchemaVersion()).toBe(115); expect(db.getSchemaVersion()).toBe(115); db.close(); }); @@ -1235,14 +1227,12 @@ describe("schema migration", () => { const db = new Database(fusionDir); db.init(); expect(db.getSchemaVersion()).toBe(115); - expect(db.getSchemaVersion()).toBe(115); db.close(); // Re-open the same on-disk DB: already at 109, the 109 block must be a no-op. const reopened = new Database(fusionDir); reopened.init(); expect(reopened.getSchemaVersion()).toBe(115); - expect(reopened.getSchemaVersion()).toBe(115); const workflowColumns = reopened.prepare("PRAGMA table_info(workflows)").all() as Array<{ name: string }>; expect(workflowColumns.filter((c) => c.name === "kind")).toHaveLength(1); const stepColumns = reopened.prepare("PRAGMA table_info(workflow_steps)").all() as Array<{ name: string }>; diff --git a/packages/core/src/__tests__/db.test.ts b/packages/core/src/__tests__/db.test.ts index a4f89a3caa..a3fa334ad3 100644 --- a/packages/core/src/__tests__/db.test.ts +++ b/packages/core/src/__tests__/db.test.ts @@ -335,7 +335,6 @@ describe("Database", () => { it("seeds schema version", () => { expect(db.getSchemaVersion()).toBe(115); - expect(db.getSchemaVersion()).toBe(115); }); it("includes tokenUsageCacheWriteTokens on freshly initialized tasks table", () => { @@ -395,7 +394,6 @@ describe("Database", () => { it("is idempotent - calling init() twice does not fail", () => { expect(() => db.init()).not.toThrow(); expect(db.getSchemaVersion()).toBe(115); - expect(db.getSchemaVersion()).toBe(115); }); it("does not overwrite existing config on re-init", () => { // Update the config @@ -1466,7 +1464,6 @@ describe("schema migrations", () => { // Verify version bumped to 29 (includes v1→v2 through v26→v29) expect(db.getSchemaVersion()).toBe(115); - expect(db.getSchemaVersion()).toBe(115); // Verify new columns exist and existing data is intact const cols = db.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>; @@ -1496,7 +1493,6 @@ describe("schema migrations", () => { // Re-init should not fail db.init(); expect(db.getSchemaVersion()).toBe(115); - expect(db.getSchemaVersion()).toBe(115); // Re-init should not fail db.init(); @@ -1535,7 +1531,6 @@ describe("schema migrations", () => { db.init(); - expect(db.getSchemaVersion()).toBe(115); expect(db.getSchemaVersion()).toBe(115); const cols = db.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>; @@ -1577,7 +1572,6 @@ describe("schema migrations", () => { db.init(); - expect(db.getSchemaVersion()).toBe(115); expect(db.getSchemaVersion()).toBe(115); const cols = db.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>; @@ -1650,7 +1644,6 @@ describe("schema migrations", () => { db.init(); - expect(db.getSchemaVersion()).toBe(115); expect(db.getSchemaVersion()).toBe(115); const cols = db.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>; @@ -1891,7 +1884,6 @@ describe("schema migrations", () => { db.init(); - expect(db.getSchemaVersion()).toBe(115); expect(db.getSchemaVersion()).toBe(115); const cols = db.prepare("PRAGMA table_info(chat_messages)").all() as Array<{ name: string }>; @@ -1966,7 +1958,6 @@ describe("schema migrations", () => { db.init(); - expect(db.getSchemaVersion()).toBe(115); expect(db.getSchemaVersion()).toBe(115); const tables = db.prepare("SELECT name FROM sqlite_master WHERE type='table' AND name = 'agentRatings'").all() as Array<{ name: string }>; @@ -1991,7 +1982,6 @@ describe("schema migrations", () => { db.init(); - expect(db.getSchemaVersion()).toBe(115); expect(db.getSchemaVersion()).toBe(115); const tables = db.prepare("SELECT name FROM sqlite_master WHERE type='table' AND name = 'mission_events'").all() as Array<{ name: string }>; @@ -2097,7 +2087,6 @@ describe("schema migrations", () => { // Verify version bumped to 29 expect(db.getSchemaVersion()).toBe(115); - expect(db.getSchemaVersion()).toBe(115); // Verify new columns exist and existing data is intact const cols = db.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>; @@ -2316,7 +2305,6 @@ describe("schema migrations", () => { localDb.init(); - expect(localDb.getSchemaVersion()).toBe(115); expect(localDb.getSchemaVersion()).toBe(115); const columns = localDb.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>; expect(columns.map((column) => column.name)).toContain("tokenUsageCacheWriteTokens"); @@ -2628,7 +2616,6 @@ describe("createDatabase factory", () => { const db = createDatabase(fusionDir); db.init(); - expect(db.getSchemaVersion()).toBe(115); expect(db.getSchemaVersion()).toBe(115); expect(db.getLastModified()).toBeGreaterThan(0); @@ -2784,7 +2771,6 @@ describe("migration v77 task token budget columns", () => { migrated = new Database(fusion); migrated.init(); expect(migrated.getSchemaVersion()).toBe(115); - expect(migrated.getSchemaVersion()).toBe(115); const rows = migrated.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>; const names = new Set(rows.map((row) => row.name)); expect(names.has("tokenBudgetSoftAlertedAt")).toBe(true); @@ -2816,7 +2802,6 @@ describe("migration v106 adds tasks.transitionPending (FN-1417)", () => { try { fresh.init(); expect(fresh.getSchemaVersion()).toBe(115); - expect(fresh.getSchemaVersion()).toBe(115); const names = new Set( (fresh.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>).map((r) => r.name), ); @@ -2845,7 +2830,6 @@ describe("migration v106 adds tasks.transitionPending (FN-1417)", () => { migrated = new Database(fusion); migrated.init(); expect(migrated.getSchemaVersion()).toBe(115); - expect(migrated.getSchemaVersion()).toBe(115); const names = new Set( (migrated.prepare("PRAGMA table_info(tasks)").all() as Array<{ name: string }>).map((r) => r.name), ); @@ -2872,7 +2856,6 @@ describe("migration v107 adds workflow_run_branches + index (FN-1417)", () => { try { fresh.init(); expect(fresh.getSchemaVersion()).toBe(115); - expect(fresh.getSchemaVersion()).toBe(115); const table = fresh .prepare("SELECT name FROM sqlite_master WHERE type='table' AND name = 'workflow_run_branches'") .get() as { name: string } | undefined; @@ -2907,7 +2890,6 @@ describe("migration v107 adds workflow_run_branches + index (FN-1417)", () => { migrated = new Database(fusion); migrated.init(); expect(migrated.getSchemaVersion()).toBe(115); - expect(migrated.getSchemaVersion()).toBe(115); const table = migrated .prepare("SELECT name FROM sqlite_master WHERE type='table' AND name = 'workflow_run_branches'") .get() as { name: string } | undefined; @@ -2949,7 +2931,6 @@ describe("migration v67 drops orphan project auth tables", () => { migrated = new Database(fusion); migrated.init(); expect(migrated.getSchemaVersion()).toBe(115); - expect(migrated.getSchemaVersion()).toBe(115); const tables = migrated .prepare("SELECT name FROM sqlite_master WHERE type='table' AND name LIKE 'project_auth_%'") .all() as Array<{ name: string }>; @@ -2977,7 +2958,6 @@ describe("migration v67 drops orphan project auth tables", () => { try { fresh.init(); expect(fresh.getSchemaVersion()).toBe(115); - expect(fresh.getSchemaVersion()).toBe(115); const tables = fresh .prepare("SELECT name FROM sqlite_master WHERE type='table' AND name LIKE 'project_auth_%'") .all() as Array<{ name: string }>; diff --git a/packages/core/src/__tests__/run-audit.test.ts b/packages/core/src/__tests__/run-audit.test.ts index 087f71d8fc..6e15da28c5 100644 --- a/packages/core/src/__tests__/run-audit.test.ts +++ b/packages/core/src/__tests__/run-audit.test.ts @@ -583,7 +583,7 @@ describe("Run Audit", () => { expect(indexNames).toContain("idxRunAuditEventsTimestamp"); }); - it("schema version is bumped to 40", () => { + it("schema version is bumped to 115", () => { expect(db.getSchemaVersion()).toBe(115); }); }); diff --git a/packages/core/src/__tests__/store-workflow-runtime.test.ts b/packages/core/src/__tests__/store-workflow-runtime.test.ts index 0bee6ddf5f..0849cd6955 100644 --- a/packages/core/src/__tests__/store-workflow-runtime.test.ts +++ b/packages/core/src/__tests__/store-workflow-runtime.test.ts @@ -173,6 +173,33 @@ describe("TaskStore workflow work items", () => { }); }); + it("honors due-list state filters and validates lease duration", async () => { + const taskId = await createTaskId(); + const item = store.upsertWorkflowWorkItem({ + runId: "run-filter", + taskId, + nodeId: "merge.node", + kind: "merge", + state: "runnable", + now: "2026-06-09T00:00:00.000Z", + }); + store.acquireWorkflowWorkItemLease(item.id, "worker-a", { + now: "2026-06-09T00:00:00.000Z", + leaseDurationMs: 60_000, + }); + + expect(store.listDueWorkflowWorkItems({ now: "2026-06-09T00:01:00.000Z", states: ["runnable"] })).toEqual([]); + expect(store.listDueWorkflowWorkItems({ now: "2026-06-09T00:01:00.000Z", states: ["running"] }).map((due) => due.id)).toEqual([ + item.id, + ]); + expect(() => + store.acquireWorkflowWorkItemLease(item.id, "worker-b", { + now: "2026-06-09T00:01:00.000Z", + leaseDurationMs: 0, + }), + ).toThrow("workflow work item leaseDurationMs must be > 0 (received 0)"); + }); + it("preserves lease and retry metadata on idempotent duplicate upserts", async () => { const taskId = await createTaskId(); const item = store.upsertWorkflowWorkItem({ diff --git a/packages/core/src/store.ts b/packages/core/src/store.ts index a7556d52b4..849398ab1b 100644 --- a/packages/core/src/store.ts +++ b/packages/core/src/store.ts @@ -9075,12 +9075,19 @@ ${TASK_UPSERT_SQL_ASSIGNMENTS} listDueWorkflowWorkItems(filter: WorkflowWorkItemDueFilter = {}): WorkflowWorkItem[] { const now = filter.now ?? new Date().toISOString(); + const includeExpiredRunning = !filter.states || filter.states.includes("running"); const states = filter.states?.length ? filter.states : ["runnable", "retrying"]; + const stateConditions = [`(state IN (${states.map(() => "?").join(", ")}) AND (leaseExpiresAt IS NULL OR leaseExpiresAt <= ?))`]; + const params: unknown[] = [...states, now]; + if (includeExpiredRunning) { + stateConditions.push("(state = 'running' AND leaseExpiresAt IS NOT NULL AND leaseExpiresAt <= ?)"); + params.push(now); + } const conditions = [ - `((state IN (${states.map(() => "?").join(", ")}) AND (leaseExpiresAt IS NULL OR leaseExpiresAt <= ?)) OR (state = 'running' AND leaseExpiresAt IS NOT NULL AND leaseExpiresAt <= ?))`, + `(${stateConditions.join(" OR ")})`, "(retryAfter IS NULL OR retryAfter <= ?)", ]; - const params: unknown[] = [...states, now, now, now]; + params.push(now); if (filter.kinds?.length) { conditions.push(`kind IN (${filter.kinds.map(() => "?").join(", ")})`); params.push(...filter.kinds); @@ -9104,6 +9111,10 @@ ${TASK_UPSERT_SQL_ASSIGNMENTS} leaseOwner: string, opts: { leaseDurationMs: number; now?: string }, ): WorkflowWorkItem | null { + if (opts.leaseDurationMs <= 0) { + throw new Error(`workflow work item leaseDurationMs must be > 0 (received ${opts.leaseDurationMs})`); + } + return this.db.transactionImmediate(() => { const now = opts.now ?? new Date().toISOString(); const leaseExpiresAt = new Date(new Date(now).getTime() + opts.leaseDurationMs).toISOString(); From efe7db64fe51c5d70a7fdc53d6cf13358c9e5b00 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Tue, 9 Jun 2026 13:48:06 -0700 Subject: [PATCH 5/5] fix(review): stabilize ownership map test path --- .../src/__tests__/workflow-policy-ownership-map.test.ts | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/packages/engine/src/__tests__/workflow-policy-ownership-map.test.ts b/packages/engine/src/__tests__/workflow-policy-ownership-map.test.ts index 00e3e7622d..75477ec0e8 100644 --- a/packages/engine/src/__tests__/workflow-policy-ownership-map.test.ts +++ b/packages/engine/src/__tests__/workflow-policy-ownership-map.test.ts @@ -1,8 +1,10 @@ import { readFileSync } from "node:fs"; import { resolve } from "node:path"; +import { fileURLToPath } from "node:url"; import { describe, expect, it } from "vitest"; -const DOC_PATH = resolve(process.cwd(), "../../docs/workflow-policy-ownership-map.md"); +const __dirname = fileURLToPath(new URL(".", import.meta.url)); +const DOC_PATH = resolve(__dirname, "../../../../docs/workflow-policy-ownership-map.md"); const REQUIRED_SOURCE_FILES = [ "packages/engine/src/project-engine.ts",