From d2ce1ba8b55e8b4cc0282bd023cc1a64ea98e065 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Wed, 29 Jul 2026 10:44:12 -0700 Subject: [PATCH] U11: resolve the scheduler's event-handler columns by trait (10 live sites, sync resolution) (#2518) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Based on `main`. Ten live `"todo"` sites in `scheduler.ts` now resolve the column by trait. ## Four groups, converted together They fail **independently**, and a half-conversion is indistinguishable from a working system: | group | sites | failure mode | |---|---:|---| | **Wake triggers** | 4 | **Latency** — snapshot invalidation, mission-failure tracking, engine requeue tracking, move-to-backlog wake. The wake doesn't fire and the card waits up to a poll interval. Exactly why it would go unnoticed indefinitely. | | **Parked wakes** | 2 | Latency — unpause and planning-finished, keyed on hold OR intake. | | **Dependency** | 3 | **Not latency.** After a blocker completes or is soft-deleted, the query returns nothing, so the dependent is *never* unblocked and waits on a blocker that already finished. | | **Agent link** | 1 | `rollbackRunningAgentsForQueuedTodoTask` passes a synthetic `{ column: "todo" }`. Wrong here **drops a running agent's task link** — the worse direction of that safeguard. Resolved `parkedColumns` is now passed through too, rather than letting the helper fall back to its legacy default. | ## Resolution is synchronous, deliberately — the part worth reading My first cut used the async resolver and made the `task:updated` listener `async` to suit it. **That broke 5 pre-existing tests, and the tests were right:** introducing a new `await` *before* a listener's existing synchronous work defers everything after it to a microtask and reorders handlers relative to a synchronous emitter. A conversion must not change event ordering. It now uses the store's sync IR path (`resolveTaskWorkflowIrSync`), so **no new suspension point is introduced anywhere**. That's the fifth time in this program a change that looked like a move quietly altered behavior — and the first time the existing suite caught it before review. ## Verification - **Mutation-verified:** forcing the resolver back to the literals fails **4 of the 6** new tests - 110 tests green across all 8 scheduler suites (6 new) - Fail-soft to the legacy pair: an unresolvable workflow behaves exactly as before rather than losing the wake - merge gate green (309 + 10 + 71), tsc clean, lint clean ## Measured 10 of my unit's 68 remaining code sites converted. `scheduler.ts` now has **one** `"todo"` literal left in live code: `isRunnableQueuedOverlapCandidate`, which is **exported but has no production caller** — its only consumer was the legacy dispatcher deleted in #2505. That's a **deletion, not a conversion**, so it is deliberately not in this PR. No changeset: `@fusion/engine` is private. 🤖 Generated with [Claude Code](https://claude.com/claude-code) ## Summary by CodeRabbit * **Bug Fixes** * Scheduling now correctly recognizes workflow-specific hold and intake columns, including renamed columns. * Tasks entering a hold column reliably trigger scheduling and wake-up behavior. * Dependency recovery now finds blocked tasks in renamed hold columns. * Planning, unpausing, task completion, deletion, and requeue flows now respect each workflow’s configured parked columns. * Prevented unnecessary scheduling for moves between unrelated workflow columns. * **Tests** * Added coverage for renamed hold-column scheduling, wake-up, and dependency-unblocking scenarios. --- .../scheduler-renamed-hold-events.test.ts | 286 ++++++++++++++++++ packages/engine/src/scheduler.ts | 78 ++++- 2 files changed, 353 insertions(+), 11 deletions(-) create mode 100644 packages/engine/src/__tests__/scheduler-renamed-hold-events.test.ts diff --git a/packages/engine/src/__tests__/scheduler-renamed-hold-events.test.ts b/packages/engine/src/__tests__/scheduler-renamed-hold-events.test.ts new file mode 100644 index 0000000000..876e450c2b --- /dev/null +++ b/packages/engine/src/__tests__/scheduler-renamed-hold-events.test.ts @@ -0,0 +1,286 @@ +/* +FNXC:WorkflowLifecycleColumns 2026-07-28-11:20 (U11 conversion — scheduler live sites): + +The scheduler's event handlers decide "is this the backlog column?" by comparing +against the literal `"todo"`. For a workflow that names its hold column anything +else, each of these silently stops firing — and after U11 deletes `todo` from the +builtins, they stop firing for EVERY workflow. + +Four groups, all covered here because they fail independently: + + WAKE TRIGGERS a move into/out of the hold column should wake the scheduler so a + freed slot is used immediately instead of waiting a poll interval. + Failure mode is SLOW, not wrong — up to one poll interval of + latency per affected move — which is exactly why it would go + unnoticed indefinitely. + + PARKED WAKES unpause and planning-finished wakes fire for a card resting in + hold OR intake. Same latency failure. + + DEPENDENCY after a blocker completes or is soft-deleted, dependents resting + in the hold column are unblocked. This one is NOT latency: a + dependent never gets unblocked, so it waits on a blocker that is + already done. + + AGENT LINK the parked-agent-link evaluation passes a synthetic + `{ column: "todo" }`, which decides whether a running agent's task + link survives. Wrong here means a live agent's link is dropped. + +Written against the literal implementation and observed FAILING first. +*/ +import { describe, expect, it, vi } from "vitest"; +import type { TaskStore, WorkflowIr } from "@fusion/core"; +import { Scheduler } from "../scheduler.js"; +import { evaluateParkedAgentTaskLink } from "../task-agent-sync.js"; + +const WF = "custom:wf"; + +/** Hold is `drafting`, intake is `inbox` — no `todo` column exists. */ +function renamedIr(): WorkflowIr { + return { + version: "v2", + id: WF, + nodes: [], + edges: [], + columns: [ + { id: "inbox", name: "inbox", traits: [{ trait: "intake" }] }, + { id: "drafting", name: "drafting", traits: [{ trait: "hold", config: { release: "capacity" } }] }, + { id: "building", name: "building", traits: [{ trait: "wip", config: { limitSetting: "maxConcurrent" } }] }, + { id: "shipped", name: "shipped", traits: [{ trait: "complete" }] }, + ], + } as unknown as WorkflowIr; +} + +function createStore(tasks: Record[] = []) { + const listeners = new Map void)[]>(); + const selection = { workflowId: WF, stepIds: [] }; + const listTasks = vi.fn(async (opts?: { column?: string }) => + opts?.column ? tasks.filter((t) => t.column === opts.column) : tasks, + ); + const store = { + on: vi.fn((event: string, listener: (payload: unknown) => void) => { + const existing = listeners.get(event) ?? []; + existing.push(listener); + listeners.set(event, existing); + }), + off: vi.fn(), + getRootDir: vi.fn().mockReturnValue("/test/project"), + getSettings: vi.fn().mockResolvedValue({ globalPause: false, enginePaused: false }), + listTasks, + getTask: vi.fn(async (id: string) => tasks.find((t) => t.id === id) ?? null), + updateTask: vi.fn().mockResolvedValue(undefined), + logEntry: vi.fn().mockResolvedValue(undefined), + getCompletionHandoffAcceptedMarker: vi.fn().mockResolvedValue(null), + getTaskWorkflowSelection: vi.fn(() => selection), + getTaskWorkflowSelectionAsync: vi.fn(async () => selection), + getWorkflowDefinition: vi.fn(async () => ({ ir: renamedIr() })), + resolveTaskWorkflowIrSync: vi.fn(() => renamedIr()), + } as unknown as TaskStore; + + return { + store, + listTasks, + emit: async (event: string, payload: unknown) => { + for (const l of listeners.get(event) ?? []) await l(payload); + }, + }; +} + +function task(over: Record = {}) { + return { + id: "FN-1", + column: "drafting", + status: null, + paused: false, + userPaused: false, + assignedAgentId: null, + checkedOutBy: null, + deletedAt: null, + dependencies: [], + blockedBy: null, + columnMovedAt: "2026-01-01T00:00:00.000Z", + createdAt: "2026-01-01T00:00:00.000Z", + updatedAt: "2026-01-01T00:00:00.000Z", + ...over, + }; +} + +/* +FNXC:WorkflowLifecycleColumns 2026-07-29-17:20 (PR #2518 review — coderabbit): +Agent-link coverage needs a real `agentStore` plus the live-execution signal, so +the harness takes both. `hasActiveAgentExecution` is the ONLY difference between +the preserve and clear cases below — the safeguard must turn on live proof, not +on the column vocabulary. +*/ +function createAgentStore(agents: Record[], freshRun: unknown = null) { + const updateAgentState = vi.fn().mockResolvedValue(undefined); + const syncExecutionTaskLink = vi.fn().mockResolvedValue(undefined); + return { + agentStore: { + listAgents: vi.fn(async () => agents), + getActiveHeartbeatRun: vi.fn(async () => freshRun), + updateAgentState, + syncExecutionTaskLink, + }, + updateAgentState, + syncExecutionTaskLink, + }; +} + +function createScheduler( + tasks: Record[] = [], + options: Record = {}, +) { + const { store, emit, listTasks } = createStore(tasks); + const scheduler = new Scheduler(store, options as never); + const schedule = vi.spyOn(scheduler, "schedule").mockResolvedValue(undefined); + (scheduler as unknown as { running: boolean }).running = true; + return { scheduler, emit, schedule, store, listTasks }; +} + +describe("scheduler event handlers under a renamed hold column", () => { + describe("wake triggers (failure mode is latency, which is why it hides)", () => { + it("wakes when a card moves INTO the renamed hold column", async () => { + const { emit, schedule } = createScheduler(); + await emit("task:moved", { task: task(), from: "building", to: "drafting", source: "engine" }); + expect(schedule).toHaveBeenCalled(); + }); + + it("does NOT wake for a move between two non-hold columns", async () => { + /* The negative half: converting must not turn every move into a wake. */ + const { emit, schedule } = createScheduler(); + await emit("task:moved", { task: task({ column: "building" }), from: "inbox", to: "building", source: "user" }); + expect(schedule).not.toHaveBeenCalled(); + }); + }); + + describe("parked wakes (hold OR intake)", () => { + it("wakes when a card unpauses in the renamed HOLD column", async () => { + const { emit, schedule } = createScheduler(); + await emit("task:updated", task({ paused: true })); + await emit("task:updated", task({ paused: false })); + expect(schedule).toHaveBeenCalled(); + }); + + it("wakes when planning finishes in the renamed INTAKE column", async () => { + const { emit, schedule } = createScheduler(); + await emit("task:updated", task({ column: "inbox", status: "planning" })); + await emit("task:updated", task({ column: "inbox", status: null })); + expect(schedule).toHaveBeenCalled(); + }); + + it("does NOT wake for a card resting in a wip column", async () => { + const { emit, schedule } = createScheduler(); + await emit("task:updated", task({ column: "building", status: "planning" })); + await emit("task:updated", task({ column: "building", status: null })); + expect(schedule).not.toHaveBeenCalled(); + }); + }); + + describe("dependency unblocking (failure mode is a card that waits forever)", () => { + it("finds dependents resting in the renamed hold column when a blocker completes", async () => { + /* + Not a latency bug: if the query returns nothing, the dependent is never + unblocked and waits on a blocker that already finished. + */ + const dependent = task({ id: "FN-DEP", column: "drafting", dependencies: ["FN-BLOCK"], blockedBy: "FN-BLOCK" }); + const blocker = task({ id: "FN-BLOCK", column: "shipped" }); + const { emit, listTasks } = createScheduler([dependent, blocker]); + + await emit("task:moved", { task: blocker, from: "building", to: "done", source: "engine" }); + + const queried = listTasks.mock.calls.map((c) => (c[0] as { column?: string } | undefined)?.column); + expect(queried).not.toContain("todo"); + expect(queried).toContain("drafting"); + }); + }); + + describe("agent link (wrong here DROPS a live agent's task link)", () => { + /* + CORRECTED after mutation testing. The first version of these tests asserted + that the literal synthetic column dropped a live agent's link — and reverting + the conversion did NOT fail them, because the literal passed `{column:"todo"}` + together with the helper's LEGACY default parked list. The pair was + self-consistent, so it read as parked either way: this site's conversion is + behaviour-NEUTRAL. + + What is actually load-bearing is that the synthetic column and `parkedColumns` + travel TOGETHER. Drift between them — a resolved column checked against the + legacy list, or the reverse — reads as unparked and clears a live agent's link. + These tests pin the live-proof safeguard and that consistency invariant, which + is what the site really depends on. + */ + const runningAgent = { id: "AG-1", taskId: "FN-1", state: "running" }; + + it("PRESERVES a live agent's link under the renamed hold column", async () => { + const { agentStore, updateAgentState, syncExecutionTaskLink } = createAgentStore([runningAgent]); + const { scheduler } = createScheduler([task()], { + agentStore, + hasActiveAgentExecution: () => true, + }); + + await (scheduler as unknown as { + rollbackRunningAgentsForQueuedTodoTask: (id: string) => Promise; + }).rollbackRunningAgentsForQueuedTodoTask("FN-1"); + + expect(updateAgentState).not.toHaveBeenCalled(); + expect(syncExecutionTaskLink).not.toHaveBeenCalled(); + }); + + it("CLEARS the link when nothing proves the execution is live", async () => { + /* The negative half. Without it the test above passes for a scheduler that + preserves every link unconditionally, which would be a different bug. */ + const { agentStore, updateAgentState, syncExecutionTaskLink } = createAgentStore([runningAgent]); + const { scheduler } = createScheduler([task()], { + agentStore, + hasActiveAgentExecution: () => false, + }); + + await (scheduler as unknown as { + rollbackRunningAgentsForQueuedTodoTask: (id: string) => Promise; + }).rollbackRunningAgentsForQueuedTodoTask("FN-1"); + + expect(updateAgentState).toHaveBeenCalledWith("AG-1", "active"); + expect(syncExecutionTaskLink).toHaveBeenCalledWith("AG-1", undefined); + }); + + it("leaves agents linked to OTHER tasks untouched", async () => { + const { agentStore, updateAgentState } = createAgentStore([ + { id: "AG-2", taskId: "FN-OTHER", state: "running" }, + ]); + const { scheduler } = createScheduler([task()], { + agentStore, + hasActiveAgentExecution: () => false, + }); + + await (scheduler as unknown as { + rollbackRunningAgentsForQueuedTodoTask: (id: string) => Promise; + }).rollbackRunningAgentsForQueuedTodoTask("FN-1"); + + expect(updateAgentState).not.toHaveBeenCalled(); + }); + + it("drops the link when the synthetic column DRIFTS from parkedColumns", () => { + /* + The real defect this site can suffer, asserted against the helper directly + because the scheduler passes the pair consistently by construction. If a + future edit resolves one side and not the other, a live agent loses its task. + */ + const drifted = evaluateParkedAgentTaskLink({ + agent: { id: "AG-1", taskId: "FN-1" }, + linkedTask: { column: "drafting" }, + hasActiveAgentExecution: () => true, + parkedColumns: ["todo", "triage"], + }); + expect(drifted.shouldPreserveParkedLink).toBe(false); + + const consistent = evaluateParkedAgentTaskLink({ + agent: { id: "AG-1", taskId: "FN-1" }, + linkedTask: { column: "drafting" }, + hasActiveAgentExecution: () => true, + parkedColumns: ["drafting", "inbox"], + }); + expect(consistent.shouldPreserveParkedLink).toBe(true); + }); + }); +}); diff --git a/packages/engine/src/scheduler.ts b/packages/engine/src/scheduler.ts index 34cb87ed7b..afe136cebc 100644 --- a/packages/engine/src/scheduler.ts +++ b/packages/engine/src/scheduler.ts @@ -39,7 +39,7 @@ import { StaleTaskReporter } from "./stale-task-reporter.js"; import { BacklogPressureReporter } from "./backlog-pressure-reporter.js"; import { UnlinkedMissionsAdvisoryReporter } from "./unlinked-missions-advisory-reporter.js"; import { createRunAuditor, generateSyntheticRunId } from "./run-audit.js"; -import { resolveWorkflowIrForTask, resolveWorkflowIrById, resolveColumnFlags, resolveWorktreeCapacityLimit } from "@fusion/core"; +import { resolveWorkflowIrForTask, resolveWorkflowIrById, resolveColumnFlags, resolveWorktreeCapacityLimit, resolveLifecycleColumns } from "@fusion/core"; import type { WorkflowIr, WorkflowIrV2 } from "@fusion/core"; import { runHoldReleaseSweep, isUnplannedForExecution, type SlotReservation } from "./hold-release.js"; import { moveTaskToReplanColumn } from "./replan-target.js"; @@ -287,6 +287,39 @@ export function getUnmetSchedulingDependencies( } +/* +FNXC:WorkflowLifecycleColumns 2026-07-28-11:35 (U11 conversion): +A task's HOLD and INTAKE columns, resolved from its own workflow. + +The scheduler's event handlers all ask a variant of "is this the backlog?" and +answered it with the literal `"todo"`. That silently stops matching for a renamed +workflow, and stops matching for EVERY workflow once U11 deletes `todo` from the +builtins. Most of these failures are LATENCY rather than incorrectness — a wake +that does not fire costs up to one poll interval — which is precisely why they +would go unnoticed. + +Resolution is per task because a board spans workflows. Cost matches existing +practice in this file, which already resolves per task at the dispatch-diagnostic +and worktree-capacity sites; these handlers fire per move/update event, not per +card in a sweep. + +Fail-soft to the legacy pair: an unresolvable workflow behaves exactly as before +this conversion rather than losing the wake entirely. +*/ +/* SYNCHRONOUS on purpose. These run inside `task:moved` / `task:updated` + listeners; introducing a new `await` before their existing synchronous work + defers everything after it to a microtask and reorders handlers relative to a + synchronous emitter. A file split must not change event ordering, so the + resolution uses the store's sync IR path. */ +function resolveTaskParkedColumnsSync(store: TaskStore, taskId: string): { hold: string; intake: string } { + try { + const lifecycle = resolveLifecycleColumns(store.resolveTaskWorkflowIrSync(taskId)); + return { hold: lifecycle?.hold ?? "todo", intake: lifecycle?.intake ?? "triage" }; + } catch { + return { hold: "todo", intake: "triage" }; + } +} + export function shouldHoldActiveFileScopeLease( task: Task, tasks: Task[], @@ -753,7 +786,8 @@ export class Scheduler { */ this.store.on("task:moved", async ({ task, from, to, source }) => { this.lastAutoClaimFingerprint.set(task.id, computeAutoClaimFingerprint(task)); - if (from === "todo" || to === "todo") { + const parked = resolveTaskParkedColumnsSync(this.store, task.id); + if (from === parked.hold || to === parked.hold) { this.options.snapshotManager?.invalidate(`task:moved:${from}->${to}`); } // PR Monitoring @@ -793,7 +827,7 @@ export class Scheduler { // Mission failure tracking: status/error are cleared during moveTask(in-progress → todo), // so we pair this with failedTaskIds captured from task:updated events. - if (task.sliceId && to === "todo" && this.options.onTaskFailed) { + if (task.sliceId && to === parked.hold && this.options.onTaskFailed) { if (task.status === "failed" || this.failedTaskIds.has(task.id)) { this.failedTaskIds.delete(task.id); void Promise.resolve(this.options.onTaskFailed(task.id)).catch((err) => { @@ -809,7 +843,7 @@ export class Scheduler { try { const settings = await this.store.getSettings(); if (!settings.globalPause && !settings.enginePaused) { - const todoTasks = await this.store.listTasks({ column: "todo", slim: true }); + const todoTasks = await this.store.listTasks({ column: parked.hold, slim: true }); for (const dependent of todoTasks) { const mentionsCompletedTask = dependent.dependencies.includes(task.id); const currentlyBlockedByCompletedTask = dependent.blockedBy === task.id; @@ -868,7 +902,7 @@ export class Scheduler { } } - if (from === "in-progress" && to === "todo") { + if (from === "in-progress" && to === parked.hold) { if (source === "engine") { this.recentEngineTodoRequeues.set(task.id, task.columnMovedAt ?? new Date().toISOString()); } else { @@ -891,7 +925,7 @@ export class Scheduler { // Event-driven scheduling: when a task moves to "done" (completion) or "todo" (retry/manual move), // trigger scheduling immediately so waiting tasks can start without waiting // for the next poll interval (up to 15 seconds). - if (to === "done" || to === "todo") { + if (to === "done" || to === parked.hold) { schedulerLog.log(`Task moved to ${to} — triggering scheduling`); this.schedule(); } @@ -943,7 +977,8 @@ export class Scheduler { schedulerLog.warn(`Failed to reset dispatch oscillation state for ${task.id} on unpause: ${error instanceof Error ? error.message : String(error)}`); }); } - if (this.running && (task.column === "todo" || task.column === "triage")) { + const unpausedParked = resolveTaskParkedColumnsSync(this.store, task.id); + if (this.running && (task.column === unpausedParked.hold || task.column === unpausedParked.intake)) { schedulerLog.log(`Task ${task.id} unpaused — triggering scheduling`); this.schedule(); } @@ -969,12 +1004,13 @@ export class Scheduler { this.planningTaskIds.add(task.id); } else if (this.planningTaskIds.has(task.id)) { this.planningTaskIds.delete(task.id); + const planningParked = resolveTaskParkedColumnsSync(this.store, task.id); if ( this.running && !task.status && !task.paused && !task.userPaused - && (task.column === "todo" || task.column === "triage") + && (task.column === planningParked.hold || task.column === planningParked.intake) ) { schedulerLog.log(`Task ${task.id} finished planning — triggering scheduling`); this.schedule(); @@ -1019,7 +1055,8 @@ export class Scheduler { return; } - const todoTasks = await this.store.listTasks({ column: "todo", slim: true }); + const deletedParked = resolveTaskParkedColumnsSync(this.store, task.id); + const todoTasks = await this.store.listTasks({ column: deletedParked.hold, slim: true }); const inProgressTasks = await this.store.listTasks({ column: "in-progress", slim: true }); const dependents = [...todoTasks, ...inProgressTasks]; @@ -1058,7 +1095,7 @@ export class Scheduler { dependent.id, `Auto-reblocked (FN-5496): unresolved dependency ${nextBlocker} remains after blocker ${task.id} was soft-deleted`, ); - } else if (dependent.column === "todo") { + } else if (dependent.column === deletedParked.hold) { await this.store.updateTask(dependent.id, { blockedBy: null, status: null }); await this.store.logEntry(dependent.id, `Auto-unblocked (FN-5496): blocker ${task.id} was soft-deleted`); } else { @@ -1283,11 +1320,30 @@ export class Scheduler { for (const agent of linkedAgents) { const activeRun = await agentStore.getActiveHeartbeatRun?.(agent.id); + /* + FNXC:WorkflowLifecycleColumns 2026-07-29-17:45 (PR #2518 self-review — CORRECTED): + The synthetic column here stands for "this task is parked in the backlog", + which is what `evaluateParkedAgentTaskLink` gates on via + `isParkedTaskColumn(linkedTask, parkedColumns)`. + + An earlier version of this note claimed the literal made a renamed workflow + read as UNPARKED and drop a live agent's link. THAT WAS WRONG, and the + mutation test proved it: the literal passed `{column:"todo"}` together with + the helper's LEGACY default parked list, so the pair was self-consistent and + read as parked either way. This conversion is behaviour-NEUTRAL here. + + What is load-bearing is the pair travelling TOGETHER. The dangerous state is + drift — a resolved column checked against the legacy list (or the reverse), + which reads as unparked and clears a live agent's link. That invariant, not + the conversion, is what the agent-link tests pin. + */ + const rollbackParked = resolveTaskParkedColumnsSync(this.store, taskId); const proof = evaluateParkedAgentTaskLink({ agent, - linkedTask: { column: "todo" } as Pick, + linkedTask: { column: rollbackParked.hold } as Pick, activeRun, hasActiveAgentExecution: this.options.hasActiveAgentExecution, + parkedColumns: [rollbackParked.hold, rollbackParked.intake], }); if (proof.shouldPreserveParkedLink) { schedulerLog.log(