From ebe514c3e495b4e750d6745b9566987aaebec48a Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Sat, 1 Aug 2026 01:15:10 -0700 Subject: [PATCH] FN-8677: propagate task update lanes before conversion Propagate cache-warmed workflow lanes through task updates so synchronous engine consumers support renamed boards. - Add task lane cache and attach resolved lanes to task:updated metadata. - Update scheduler, triage, and notification consumers to use carried lanes with bridge-safe fallbacks. - Cover lane propagation and renamed-lane event behavior with core and engine tests. Files changed: .changeset/fn-8677-manual-merge-hold-lanes.md | 7 ++ .changeset/task-updated-carries-lanes.md | 7 ++ ...orkflow-ir-readers-always-return-the-default.md | 22 +++++ .../sync-workflow-ir-second-blocker.test.ts | 43 +++----- .../core/src/__tests__/task-lane-cache.test.ts | 30 ++++++ .../task-updated-lanes-emit-surfaces.test.ts | 92 ++++++++++++++++++ .../__tests__/task-updated-lanes-payload.test.ts | 42 ++++++++ packages/core/src/index.ts | 1 + packages/core/src/store.ts | 36 ++++++- packages/core/src/task-lane-cache.ts | 63 ++++++++++++ .../core/src/task-store/archive-lifecycle-2.ts | 3 + packages/core/src/task-store/moves.ts | 1 + packages/core/src/task-store/task-artifacts-ops.ts | 1 + packages/core/src/task-store/task-update.ts | 1 + packages/core/src/task-store/update-task-deps.ts | 4 +- .../core/src/task-store/workflow-definitions.ts | 71 +++++--------- .../__tests__/scheduler-task-updated-lanes.test.ts | 108 +++++++++++++++++++++ .../task-updated-lanes-bridge-compat.test.ts | 94 ++++++++++++++++++ ...task-updated-lanes-engine-emit-surfaces.test.ts | 101 +++++++++++++++++++ .../src/__tests__/triage-pause-abort.test.ts | 22 +++++ .../src/__tests__/triage-planning-wake.test.ts | 25 +++++ .../notification-renamed-lifecycle-columns.test.ts | 84 +++++++++++++++- .../__tests__/task-wedge-notification.test.ts | 19 ++++ .../src/notification/notification-service.ts | 56 ++++------- packages/engine/src/scheduler.ts | 62 +++--------- packages/engine/src/triage.ts | 105 ++++++-------------- scripts/lib/inert-sync-lane-baseline.json | 5 +- 27 files changed, 858 insertions(+), 247 deletions(-) Fusion-Task-Id: FN-8677 Fusion-Task-Lineage: d8fef9db-0f88-4dfd-9813-be25e10e3588 Co-authored-by: Fusion (runfusion.ai) --- .changeset/fn-8677-manual-merge-hold-lanes.md | 7 ++ .changeset/task-updated-carries-lanes.md | 7 ++ ...ow-ir-readers-always-return-the-default.md | 22 ++++ .../sync-workflow-ir-second-blocker.test.ts | 41 +++---- .../src/__tests__/task-lane-cache.test.ts | 30 +++++ .../task-updated-lanes-emit-surfaces.test.ts | 92 +++++++++++++++ .../task-updated-lanes-payload.test.ts | 42 +++++++ packages/core/src/index.ts | 1 + packages/core/src/store.ts | 36 +++++- packages/core/src/task-lane-cache.ts | 63 ++++++++++ .../src/task-store/archive-lifecycle-2.ts | 3 + packages/core/src/task-store/moves.ts | 1 + .../core/src/task-store/task-artifacts-ops.ts | 1 + packages/core/src/task-store/task-update.ts | 1 + .../core/src/task-store/update-task-deps.ts | 4 +- .../src/task-store/workflow-definitions.ts | 71 ++++-------- .../scheduler-task-updated-lanes.test.ts | 108 ++++++++++++++++++ .../task-updated-lanes-bridge-compat.test.ts | 94 +++++++++++++++ ...updated-lanes-engine-emit-surfaces.test.ts | 101 ++++++++++++++++ .../src/__tests__/triage-pause-abort.test.ts | 22 ++++ .../__tests__/triage-planning-wake.test.ts | 25 ++++ ...fication-renamed-lifecycle-columns.test.ts | 84 +++++++++++++- .../__tests__/task-wedge-notification.test.ts | 19 +++ .../src/notification/notification-service.ts | 54 ++++----- packages/engine/src/scheduler.ts | 62 ++-------- packages/engine/src/triage.ts | 105 +++++------------ scripts/lib/inert-sync-lane-baseline.json | 5 +- 27 files changed, 856 insertions(+), 245 deletions(-) create mode 100644 .changeset/fn-8677-manual-merge-hold-lanes.md create mode 100644 .changeset/task-updated-carries-lanes.md create mode 100644 packages/core/src/__tests__/task-lane-cache.test.ts create mode 100644 packages/core/src/__tests__/task-updated-lanes-emit-surfaces.test.ts create mode 100644 packages/core/src/__tests__/task-updated-lanes-payload.test.ts create mode 100644 packages/core/src/task-lane-cache.ts create mode 100644 packages/engine/src/__tests__/scheduler-task-updated-lanes.test.ts create mode 100644 packages/engine/src/__tests__/task-updated-lanes-bridge-compat.test.ts create mode 100644 packages/engine/src/__tests__/task-updated-lanes-engine-emit-surfaces.test.ts diff --git a/.changeset/fn-8677-manual-merge-hold-lanes.md b/.changeset/fn-8677-manual-merge-hold-lanes.md new file mode 100644 index 0000000000..78e4c7f02e --- /dev/null +++ b/.changeset/fn-8677-manual-merge-hold-lanes.md @@ -0,0 +1,7 @@ +--- +"@runfusion/fusion": patch +--- + +summary: Notify operators when manual merges wait in renamed workflow review lanes. +category: fix +dev: Uses emitter-carried task update lanes without making notification listeners asynchronous. diff --git a/.changeset/task-updated-carries-lanes.md b/.changeset/task-updated-carries-lanes.md new file mode 100644 index 0000000000..48c755a8d3 --- /dev/null +++ b/.changeset/task-updated-carries-lanes.md @@ -0,0 +1,7 @@ +--- +"@runfusion/fusion": patch +--- + +summary: Keep scheduler and planning reactions working on renamed workflow lanes. +category: fix +dev: TaskStore decorates task:updated with cached resolved lanes; runtime bridges intentionally drop optional metadata. diff --git a/docs/solutions/architecture-patterns/sync-workflow-ir-readers-always-return-the-default.md b/docs/solutions/architecture-patterns/sync-workflow-ir-readers-always-return-the-default.md index 058d36cc1c..dd35c6d0db 100644 --- a/docs/solutions/architecture-patterns/sync-workflow-ir-readers-always-return-the-default.md +++ b/docs/solutions/architecture-patterns/sync-workflow-ir-readers-always-return-the-default.md @@ -149,3 +149,25 @@ next person does not start by trying to add \`await\` and conclude the codebase Each consequence needs its sync call path made async, which is a real slice per site. This document is the finding; the fixes are follow-ups. #2593 fixed only the one that was mine. + +## Cached emitter-carried answers for synchronous lifecycle listeners (FN-8658) + +`TaskStore` owns a bounded, short-TTL `TaskLaneCache`. Paths that already resolve a task workflow +warm it with `toTaskMoveLanes`; the central `TaskStore.emit` seam attaches that cached answer as the +optional second argument of `task:updated`. The seam performs only a synchronous Map lookup, so hot +update emitters add neither an IR resolution nor a database query. Local workflow-selection writes +invalidate entries and TTL bounds staleness after another PostgreSQL node changes a selection. + +`undefined` means **unknown**, never default or legacy. Subscribers must keep their literal fallback +when `meta?.lanes` is absent rather than consult the default-only sync resolver. That preserves +one-argument listener compatibility while synchronous scheduler and triage edge-trigger handlers can +handle renamed lanes correctly when metadata is present. + +Runtime/process bridge emitters (`project-manager`, `hybrid-executor`, and the in-process, +child-process, and remote-node runtimes) deliberately DROP this optional metadata. They emit from +their own EventEmitter rather than a TaskStore; forwarding would widen serialized bridge contracts, +while absent metadata is safe because listeners retain their fallback. +`task-updated-lanes-emit-surfaces.test.ts` exhaustively ratchets core emitters and +`task-updated-lanes-engine-emit-surfaces.test.ts` classifies engine emitters; extend those explicit +tables if a new `task:updated` emitter is added. `task-updated-lanes-bridge-compat.test.ts` pins the +DROP bridge compatibility contract. diff --git a/packages/core/src/__tests__/sync-workflow-ir-second-blocker.test.ts b/packages/core/src/__tests__/sync-workflow-ir-second-blocker.test.ts index a0c6dad52f..7a5b77daa2 100644 --- a/packages/core/src/__tests__/sync-workflow-ir-second-blocker.test.ts +++ b/packages/core/src/__tests__/sync-workflow-ir-second-blocker.test.ts @@ -36,28 +36,15 @@ that survives a writer on a different host. This file exists so the next person to attempt the unblock reads all three before starting, instead of shipping a selection cache and discovering the second read at integration time. -FNXC:WorkflowLifecycleColumns 2026-07-31-23:55 (the emitter route works for `task:moved` and does NOT -generalise to `task:updated` — measured, before someone does the obvious follow-on): -#3109 solved this class for `task:moved` by having the EMITTER resolve lanes once and carry them on -the payload. It is the right fix there and it retired several flags within a day. The obvious next -step is to do the same for `task:updated`, which is where the remaining sync-listener guards live -(`scheduler.ts`'s mission-failure and PR-monitoring guards, `triage.ts`'s planning-evacuation guard). +FNXC:WorkflowLifecycleColumns 2026-08-01-06:11 (FN-8658 supersedes the former non-generalisation): +The measured cost profile remains: 7 `task:moved` emit sites versus 26 `task:updated` sites across +ten files, including `audit-ops.ts`'s logEntry fast path. An IR read per update is still a regression. -The cost profile is opposite, and that is why #3109's justification does not transfer. Measured on -this tree: 7 `task:moved` emit sites versus 26 `task:updated` sites across ten files. #3109 could -argue the resolution away because a move "is already async and already post-commit, so the resolution -costs one IR read on a transition that has just done database work". `task:updated` fires on every log -append, comment, artifact write and steering message — `audit-ops.ts`'s logEntry fast path is one of -them, and it exists precisely to avoid re-reading the task. An IR read per emit there is a regression -on the hottest write path in the system. - -Partial coverage does not rescue it either, and `triage.ts` is the reason: its evacuation handler -reacts to ANY `task:updated` carrying a column — it explicitly tolerates partial payloads — so it -needs lanes on essentially every emit or it keeps its literal fallback anyway. - -So the remaining `task:updated` guards are NOT one commit behind #3109. Unblocking them wants either a -cached lane answer the emitter can attach for free, or the sync reader described above. Recorded so -the follow-on is a decision rather than a surprise. +FN-8658 therefore does not resolve on each emit. Paths that already asynchronously resolve the task +workflow warm a bounded, TTL-limited per-store cache; the central TaskStore emit seam attaches its +answer when present. The synchronous listener receives `undefined` as unknown and retains its literal +fallback. This preserves the measurement while avoiding both a hot-path read and PostgreSQL's +default-only sync resolver. */ import { describe, expect, it } from "vitest"; @@ -113,13 +100,13 @@ describe("the sync workflow-IR path is blocked twice, not once", () => { }); /* - FNXC:WorkflowLifecycleColumns 2026-07-31-23:55: - The emitter route's cost argument, pinned as a ratio rather than prose. #3109 justified resolving an - IR per `task:moved` because a move has just done database work; `task:updated` fires on every log - append and comment, so the same change there is a hot-path regression. If these counts ever converge - the trade-off changes and the note above should be re-read. + FNXC:WorkflowLifecycleColumns 2026-08-01-06:29: + The emitter route's cost argument, pinned as a ratio rather than prose. Resolving an IR per + `task:updated` is still a hot-path regression because log and comment writes emit it far more often + than moves. FN-8658 generalises the event contract through a cache-read central emit seam instead: + every emitter gets an answer without an IR read. If these counts ever converge, re-read the trade-off. */ - it("`task:updated` has far more emit sites than `task:moved`, which is why the emitter route does not generalise", () => { + it("`task:updated` has far more emit sites than `task:moved`, requiring a cached central emitter seam", () => { const countEmits = (event) => { const files = new Set(); let total = 0; diff --git a/packages/core/src/__tests__/task-lane-cache.test.ts b/packages/core/src/__tests__/task-lane-cache.test.ts new file mode 100644 index 0000000000..fbb9d09f42 --- /dev/null +++ b/packages/core/src/__tests__/task-lane-cache.test.ts @@ -0,0 +1,30 @@ +import { describe, expect, it } from "vitest"; +import { TaskLaneCache } from "../task-lane-cache.js"; + +const lanes = { hold: "queued", wip: "building", review: "reviewing" }; + +describe("TaskLaneCache", () => { + it("returns hits, misses, explicit invalidations, and fake-clock expiry", () => { + let now = 0; + const cache = new TaskLaneCache({ ttlMs: 10, now: () => now }); + expect(cache.get("missing")).toBeUndefined(); + cache.set("task", lanes); + expect(cache.get("task")).toEqual(lanes); + cache.invalidate("task"); + expect(cache.get("task")).toBeUndefined(); + cache.set("task", lanes); + now = 10; + expect(cache.get("task")).toBeUndefined(); + }); + + it("evicts least-recently-used entries at its configured bound", () => { + const cache = new TaskLaneCache({ maxSize: 2 }); + cache.set("first", lanes); + cache.set("second", lanes); + expect(cache.get("first")).toEqual(lanes); + cache.set("third", lanes); + expect(cache.get("second")).toBeUndefined(); + expect(cache.get("first")).toEqual(lanes); + expect(cache.get("third")).toEqual(lanes); + }); +}); diff --git a/packages/core/src/__tests__/task-updated-lanes-emit-surfaces.test.ts b/packages/core/src/__tests__/task-updated-lanes-emit-surfaces.test.ts new file mode 100644 index 0000000000..0532804072 --- /dev/null +++ b/packages/core/src/__tests__/task-updated-lanes-emit-surfaces.test.ts @@ -0,0 +1,92 @@ +import { describe, expect, it } from "vitest"; +import { readdirSync, readFileSync, statSync } from "node:fs"; +import { join, relative, resolve } from "node:path"; +import { TaskStore } from "../store.js"; +import type { Task } from "../types.js"; + +const REPO_ROOT = resolve(__dirname, "../../../.."); +const CORE_ROOT = join(REPO_ROOT, "packages/core/src"); + +const EMIT_SURFACES = [ + "packages/core/src/store.ts", + "packages/core/src/task-store/audit-ops.ts", + "packages/core/src/task-store/branch-group-ops.ts", + "packages/core/src/task-store/comments-ops.ts", + "packages/core/src/task-store/lifecycle-ops.ts", + "packages/core/src/task-store/merge-queue-ops.ts", + "packages/core/src/task-store/moves.ts", + "packages/core/src/task-store/project-store-ops.ts", + "packages/core/src/task-store/task-artifacts-ops.ts", + "packages/core/src/task-store/task-mutation-ops.ts", + "packages/core/src/task-store/workflow-task-create-ops.ts", +] as const; + +/* +FNXC:WorkflowEvents 2026-08-01-07:35: +These are producer paths, not a sampled list of listeners. The static scan makes a newly added +producer fail closed, while the receiver assertion below proves each existing producer invokes the +TaskStore instance seam that is exercised with warm and cold cache states in this suite. +*/ +const TASK_STORE_EMIT_RECEIVERS: Record<(typeof EMIT_SURFACES)[number], readonly string[]> = { + "packages/core/src/store.ts": ["this.emit"], + "packages/core/src/task-store/audit-ops.ts": ["store.emit", "store.emitTaskLifecycleEventSafely"], + "packages/core/src/task-store/branch-group-ops.ts": ["store.emit"], + "packages/core/src/task-store/comments-ops.ts": ["store.emit"], + "packages/core/src/task-store/lifecycle-ops.ts": ["store.emit"], + "packages/core/src/task-store/merge-queue-ops.ts": ["store.emit"], + "packages/core/src/task-store/moves.ts": ["store.emit"], + "packages/core/src/task-store/project-store-ops.ts": ["store.emit"], + "packages/core/src/task-store/task-artifacts-ops.ts": ["store.emit"], + "packages/core/src/task-store/task-mutation-ops.ts": ["store.emitTaskLifecycleEventSafely", "store.emit"], + "packages/core/src/task-store/workflow-task-create-ops.ts": ["store.emit", "store.emitTaskLifecycleEventSafely"], +}; + +function* walk(dir: string): Generator { + for (const entry of readdirSync(dir)) { + if (entry === "__tests__" || entry === "dist") continue; + const path = join(dir, entry); + if (statSync(path).isDirectory()) yield* walk(path); + else if (path.endsWith(".ts")) yield path; + } +} + +const task = { id: "FN-surface", column: "building" } as Task; + +/* +FNXC:WorkflowEvents 2026-08-01-06:57: +Every production core update producer invokes the TaskStore instance emitter, not EventEmitter's +prototype or a private relay. That structural boundary is what lets one decorated TaskStore seam +cover every producer warm and cold without adding an IR read to hot mutation paths. +*/ +describe("core task:updated emit surface", () => { + it("registers every production producer and constrains each to TaskStore's decorated emitter", () => { + const emittingModules = [...walk(CORE_ROOT)] + .filter((file) => readFileSync(file, "utf8").includes('emit("task:updated"')) + .map((file) => relative(REPO_ROOT, file).split("\\").join("/")) + .sort(); + + expect(emittingModules).toEqual([...EMIT_SURFACES].sort()); + expect(Object.keys(TASK_STORE_EMIT_RECEIVERS).sort()).toEqual([...EMIT_SURFACES].sort()); + for (const file of emittingModules) { + const source = readFileSync(join(REPO_ROOT, file), "utf8"); + const receivers = TASK_STORE_EMIT_RECEIVERS[file as keyof typeof TASK_STORE_EMIT_RECEIVERS]; + expect(receivers.some((receiver) => source.includes(receiver)), `${file} must call the TaskStore emitter`).toBe(true); + expect(source).not.toMatch(/EventEmitter\.prototype\.emit\([^\n]*task:updated/); + } + }); + + it("delivers the warm and cold result at the TaskStore seam used by every registered producer", () => { + const store = new TaskStore(process.cwd()); + const received: Array<{ lanes?: { wip?: string } } | undefined> = []; + store.on("task:updated", (_task, meta) => received.push(meta)); + + // The producers above all invoke this real TaskStore override; this verifies their shared delivery + // contract rather than allowing a module-specific helper to silently bypass cache decoration. + store.laneCache.set(task.id, { wip: "building" }); + store.emit("task:updated", task); + store.laneCache.invalidate(task.id); + store.emit("task:updated", task); + + expect(received).toEqual([{ lanes: { wip: "building" } }, undefined]); + }); +}); diff --git a/packages/core/src/__tests__/task-updated-lanes-payload.test.ts b/packages/core/src/__tests__/task-updated-lanes-payload.test.ts new file mode 100644 index 0000000000..b1c088a4be --- /dev/null +++ b/packages/core/src/__tests__/task-updated-lanes-payload.test.ts @@ -0,0 +1,42 @@ +import { describe, expect, it } from "vitest"; +import { TaskStore } from "../store.js"; +import type { Task } from "../types.js"; + +const task = { id: "FN-lanes", column: "building" } as Task; + +describe("task:updated lane payload", () => { + it("decorates cache hits while keeping one-argument listeners and misses compatible", () => { + const store = new TaskStore(process.cwd()); + const received: Array<{ lanes?: { wip?: string } } | undefined> = []; + let oneArgumentCalls = 0; + store.on("task:updated", (_task, meta) => received.push(meta)); + store.on("task:updated", () => { oneArgumentCalls += 1; }); + + store.laneCache.set(task.id, { wip: "building" }); + store.emit("task:updated", task); + store.laneCache.invalidate(task.id); + store.emit("task:updated", task); + + expect(received).toEqual([{ lanes: { wip: "building" } }, undefined]); + expect(oneArgumentCalls).toBe(2); + }); + + it("decorates safe lifecycle emissions, which invoke listeners without EventEmitter.emit", () => { + const store = new TaskStore(process.cwd()); + store.laneCache.set(task.id, { wip: "building" }); + let received: { lanes?: { wip?: string } } | undefined; + store.on("task:updated", (_task, meta) => { received = meta; }); + + store.emitTaskLifecycleEventSafely("task:updated", [task]); + expect(received).toEqual({ lanes: { wip: "building" } }); + }); + + it("preserves explicit metadata rather than replacing it from cache", () => { + const store = new TaskStore(process.cwd()); + store.laneCache.set(task.id, { wip: "cached" }); + let received: { lanes?: { wip?: string } } | undefined; + store.on("task:updated", (_task, meta) => { received = meta; }); + store.emit("task:updated", task, { lanes: { wip: "explicit" } }); + expect(received).toEqual({ lanes: { wip: "explicit" } }); + }); +}); diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 9ce8857abb..7d8bf9b60d 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -468,6 +468,7 @@ export type { WorkflowEventBus, WorkflowEventSubscriber, WorkflowEventSubscripti export { findWorkflowEventShapeViolations, isIdsOnlyWorkflowEvent, MAX_ID_VALUE_LENGTH, IMPLEMENTATION_EXITS } from "./types/workflow-events.js"; export type { WorkflowLifecycleEvent, WorkflowLifecycleEventType, WorkflowLifecycleEventBase, TaskTransitionedEvent, NodeEnteredEvent, NodeCompletedEvent, RunSuspendedEvent, RunResumedEvent, WorkflowEventShapeViolation, ImplementationExit } from "./types/workflow-events.js"; export { columnHasFlag, columnsWithFlag, declaresAnyLifecycleTrait, resolveArchiveTargetForTask, resolveCompleteColumn, resolveLifecycleColumns, resolveMergeOrchestrationColumn, resolveReboundTarget, resolveReboundTargetForTask, resolveReviewColumns, resolveTaskLifecycleColumns, resolveTerminalColumns, resolveWipTargetForTask, toTaskMoveLanes } from "./workflow-lifecycle-traits.js"; +export { TaskLaneCache, type TaskLaneCacheOptions } from "./task-lane-cache.js"; export type { LifecycleColumns, TaskMoveLanes } from "./workflow-lifecycle-traits.js"; export { resolveProjectColumnsForRoles, resolveArchivedLanes, REVIEW_ROLES, TERMINAL_ROLES, LEGACY_COLUMN_IDS_BY_ROLE, type ProjectLaneVocabularyStore, type ProjectLaneResolutionOptions } from "./project-lane-vocabulary.js"; export { resolveReviewLevelSteps, applyReviewLevelPreset } from "./review-level-preset.js"; diff --git a/packages/core/src/store.ts b/packages/core/src/store.ts index 0842fc3b40..55b236457a 100644 --- a/packages/core/src/store.ts +++ b/packages/core/src/store.ts @@ -1,5 +1,6 @@ import { EventEmitter } from "node:events"; import type { TaskMoveLanes } from "./workflow-lifecycle-traits.js"; +import { TaskLaneCache } from "./task-lane-cache.js"; import { randomUUID } from "node:crypto"; import { join } from "node:path"; import { and, eq, isNull, ne, sql } from "drizzle-orm"; @@ -161,7 +162,13 @@ export interface TaskStoreEvents { never "legacy". */ "task:moved": [data: { task: Task; from: ColumnId; to: ColumnId; source: "user" | "engine" | "scheduler"; lanes?: TaskMoveLanes }]; - "task:updated": [task: Task]; + /* + FNXC:WorkflowEvents 2026-08-01-06:11: + `task:updated` listeners are synchronous and may receive cache-warmed resolved lanes as an optional + second argument. The first argument remains the Task so existing one-argument subscribers are + unchanged; absent metadata is unknown, never a legacy-lane claim. + */ + "task:updated": [task: Task, meta?: { lanes?: TaskMoveLanes }]; "task:deleted": [task: Task, meta?: { githubIssueAction?: GithubIssueAction }]; "task:merged": [result: MergeResult]; "settings:updated": [data: { settings: Settings; previous: Settings }]; @@ -404,6 +411,8 @@ export class TaskStore extends EventEmitter { public watcher: FSWatcher | null = null; public taskCache: Map = new Map(); + /** Per-store, bounded answer cache used only to decorate synchronous task:updated events. */ + public readonly laneCache = new TaskLaneCache(); /* FNXC:IncompletePgPorts 2026-07-26-20:35: Sync getDatabaseHealth/healthCheck cannot await PostgreSQL. Cache the last @@ -528,7 +537,17 @@ export class TaskStore extends EventEmitter { this.globalSettingsDir = resolvedGlobalSettingsDir; this.globalSettingsStore = new GlobalSettingsStore(resolvedGlobalSettingsDir, this.asyncLayer ?? undefined); } + /* + FNXC:WorkflowEvents 2026-08-01-06:28: + Safe lifecycle emission invokes listeners directly to isolate listener failures, so it bypasses + EventEmitter.emit. Decorate this path too; otherwise the hot update surfaces silently miss lanes. + */ public emitTaskLifecycleEventSafely( event: "task:created" | "task:updated", args: TaskStoreEvents["task:created"] | TaskStoreEvents["task:updated"], ): boolean { + if (event === "task:updated" && args.length === 1) { + const task = args[0] as Task; + const lanes = this.laneCache.get(task.id); + if (lanes !== undefined) return emitTaskLifecycleEventSafelyImpl(this, event, [task, { lanes }]); + } return emitTaskLifecycleEventSafelyImpl(this, event, args); } @@ -1564,6 +1583,21 @@ export class TaskStore extends EventEmitter { return task; }); } + /* + FNXC:WorkflowEvents 2026-08-01-06:11: + One seam decorates every TaskStore `task:updated` emission, including hot synchronous paths, with a + cached answer only. Explicit metadata wins; runtime and process bridge EventEmitters do not call this + method and deliberately DROP lanes because absent metadata safely preserves their listener fallback. + */ + override emit(event: E, ...args: any[]): boolean { + if (event === "task:updated" && args.length === 1) { + const task = args[0] as Task; + const lanes = this.laneCache.get(task.id); + if (lanes !== undefined) return EventEmitter.prototype.emit.call(this, event, task, { lanes }); + } + return EventEmitter.prototype.emit.call(this, event, ...args); + } + async updateStep( id: string, stepIndex: number, status: import("./types.js").StepStatus, options?: { source?: "graph" }, ): Promise { return updateStepImpl(this, id, stepIndex, status, options); } diff --git a/packages/core/src/task-lane-cache.ts b/packages/core/src/task-lane-cache.ts new file mode 100644 index 0000000000..93d1583535 --- /dev/null +++ b/packages/core/src/task-lane-cache.ts @@ -0,0 +1,63 @@ +import type { TaskMoveLanes } from "./workflow-lifecycle-traits.js"; + +export interface TaskLaneCacheOptions { + ttlMs?: number; + maxSize?: number; + now?: () => number; +} + +interface TaskLaneCacheEntry { + lanes: TaskMoveLanes | undefined; + workflowId?: string; + at: number; +} + +/* +FNXC:WorkflowEvents 2026-08-01-06:11: +PostgreSQL deployments can rewrite a task workflow selection from another node, so an in-process lane +answer must expire as well as being invalidated by local selection writes. Returning an old answer +would make a synchronous listener confidently choose the wrong lane; a cache miss is deliberately +"unknown" and lets that listener retain its existing fallback. + +The clock is injectable so TTL coverage advances a deterministic fake clock instead of sleeping. The +cache never resolves an IR: update-event decoration must remain a synchronous Map lookup on hot paths. +*/ +export class TaskLaneCache { + private readonly entries = new Map(); + private readonly ttlMs: number; + private readonly maxSize: number; + private readonly now: () => number; + + constructor({ ttlMs = 30_000, maxSize = 1_000, now = Date.now }: TaskLaneCacheOptions = {}) { + this.ttlMs = ttlMs; + this.maxSize = maxSize; + this.now = now; + } + + set(taskId: string, lanes: TaskMoveLanes | undefined, options: { workflowId?: string } = {}): void { + this.entries.delete(taskId); + this.entries.set(taskId, { lanes, workflowId: options.workflowId, at: this.now() }); + while (this.entries.size > this.maxSize) this.entries.delete(this.entries.keys().next().value!); + } + + get(taskId: string): TaskMoveLanes | undefined { + const entry = this.entries.get(taskId); + if (!entry) return undefined; + if (this.now() - entry.at >= this.ttlMs) { + this.entries.delete(taskId); + return undefined; + } + // Refresh insertion order without changing the recorded cache time. + this.entries.delete(taskId); + this.entries.set(taskId, entry); + return entry.lanes; + } + + invalidate(taskId: string): void { + this.entries.delete(taskId); + } + + clear(): void { + this.entries.clear(); + } +} diff --git a/packages/core/src/task-store/archive-lifecycle-2.ts b/packages/core/src/task-store/archive-lifecycle-2.ts index 1c662c8754..d850b7c4c9 100644 --- a/packages/core/src/task-store/archive-lifecycle-2.ts +++ b/packages/core/src/task-store/archive-lifecycle-2.ts @@ -224,6 +224,7 @@ export async function deleteTaskBackendImpl(store: TaskStore, id: string, option }); // Emit lifecycle event (best-effort, outside the transaction). + store.laneCache.invalidate(task.id); store.emit("task:deleted", task, { githubIssueAction: options?.githubIssueAction ?? "auto" }); /* FNXC:TaskDeleteNotice 2026-07-26-16:10: @@ -393,7 +394,9 @@ export async function archiveTaskBackendImpl(store: TaskStore, id: string, optio lane-less left that leak reachable through this path even after the listener itself was fixed. */ const movedLanes = toTaskMoveLanes(await resolveWorkflowIrForTask(store, task.id).catch(() => undefined)); + store.laneCache.set(task.id, movedLanes); store.emit("task:moved", { task, from: fromColumn, to: "archived" as Column, source: "engine", lanes: movedLanes }); + store.laneCache.invalidate(task.id); // Best-effort near-duplicate cleanup. await store.clearNearDuplicateReferencesToFailSoft(id, { diff --git a/packages/core/src/task-store/moves.ts b/packages/core/src/task-store/moves.ts index 422a16bb84..dedb4ea0c7 100644 --- a/packages/core/src/task-store/moves.ts +++ b/packages/core/src/task-store/moves.ts @@ -1448,6 +1448,7 @@ export async function moveTaskInternalImpl(store: TaskStore, id: string, toColum "legacy" — so a listener keeps its own fallback rather than being handed a wrong answer. */ const lanes = toTaskMoveLanes(await resolveWorkflowIrForTask(store, task.id).catch(() => undefined)); + store.laneCache.set(task.id, lanes); store.emit("task:moved", { task, from: fromColumn, to: toColumn, source: moveSource, lanes }); /* FNXC:WorkflowEvents 2026-07-27-11:45 (U3 / R5, R6): diff --git a/packages/core/src/task-store/task-artifacts-ops.ts b/packages/core/src/task-store/task-artifacts-ops.ts index e81f1e8b78..e3caf50f45 100644 --- a/packages/core/src/task-store/task-artifacts-ops.ts +++ b/packages/core/src/task-store/task-artifacts-ops.ts @@ -575,6 +575,7 @@ export async function moveToDoneImpl(store: TaskStore, task: Task, dir: string): lane-less left that leak reachable through this path even after the listener itself was fixed. */ const movedLanes = toTaskMoveLanes(await resolveWorkflowIrForTask(store, task.id).catch(() => undefined)); + store.laneCache.set(task.id, movedLanes); store.emit("task:moved", { task, from: fromColumn, to: completeColumn as Column, source: "engine", lanes: movedLanes }); } diff --git a/packages/core/src/task-store/task-update.ts b/packages/core/src/task-store/task-update.ts index dbf0ddb088..540aa5445c 100644 --- a/packages/core/src/task-store/task-update.ts +++ b/packages/core/src/task-store/task-update.ts @@ -1011,6 +1011,7 @@ export async function updateTaskUnlockedImpl(store: TaskStore, id: string, updat PostgreSQL. So these two paths kept the pre-#3109 behaviour while the listeners read as resolved. */ const lanes = toTaskMoveLanes(await resolveWorkflowIrForTask(store, id).catch(() => undefined)); + store.laneCache.set(task.id, lanes); store.emit("task:moved", { task, from: "todo" as Column, to: "triage" as Column, source: "engine", lanes }); } store.emitTaskLifecycleEventSafely("task:updated", [task]); diff --git a/packages/core/src/task-store/update-task-deps.ts b/packages/core/src/task-store/update-task-deps.ts index 35c9ae4b04..8e9dd24a96 100644 --- a/packages/core/src/task-store/update-task-deps.ts +++ b/packages/core/src/task-store/update-task-deps.ts @@ -511,12 +511,14 @@ export async function updateTaskDependenciesImpl(store: TaskStore, id: string, m column the card is already in is what re-runs reset-on-entry effects downstream. */ if (movedToTriage && respecifyFromColumn !== task.column) { + const lanes = toTaskMoveLanes(await resolveWorkflowIrForTask(store, task.id).catch(() => undefined)); + store.laneCache.set(task.id, lanes); store.emit("task:moved", { task, from: respecifyFromColumn as Column, to: task.column as Column, source: "engine", - lanes: toTaskMoveLanes(await resolveWorkflowIrForTask(store, task.id).catch(() => undefined)), + lanes, }); } store.emitTaskLifecycleEventSafely("task:updated", [task]); diff --git a/packages/core/src/task-store/workflow-definitions.ts b/packages/core/src/task-store/workflow-definitions.ts index 096d5a8f6a..be0f0e7e0a 100644 --- a/packages/core/src/task-store/workflow-definitions.ts +++ b/packages/core/src/task-store/workflow-definitions.ts @@ -556,56 +556,32 @@ export async function getTaskWorkflowSelectionAsyncImpl(store: TaskStore, taskId } export async function writeTaskWorkflowSelectionImpl(store: TaskStore, taskId: string, workflowId: string, stepIds: string[]): Promise { - const updatedAt = new Date().toISOString(); - /* - FNXC:PostgresCutover 2026-07-04-00:00: - Backend-mode upsert of the task_workflow_selection row via async Drizzle (taskId is the primary key). stepIds is stored as a JSONB array. - */ - const layer = store.asyncLayer!; - /* - FNXC:SqliteDualPathCleanup 2026-07-26-15:00: - Selection upsert must include projectId — PK is (projectId, taskId) and the authoritative read pins projectId. - */ - const projectId = layer.projectId?.trim() || "__legacy_unscoped__"; - /* - FNXC:WorkflowCapacity 2026-07-28-18:05 (PR #2499 review — cross-process race): - The selection write now runs in a transaction that first takes the per-task - advisory lock, because the in-transaction capacity gate in `moves.ts` reads this - row and enforces a limit against it. Without a lock BOTH sides take, the gate - could read the selection, this write could land, and the move would commit the - task under a workflow whose pool was never the one checked. - - `selectTaskWorkflow` already wraps this call in `store.withTaskLock`, which is - an IN-PROCESS mutex — it serializes one TaskStore instance and gives nothing - across nodes. Multi-node is several nodes against one central PostgreSQL - database, so the cross-process window is a supported deployment shape. - - Acquisition order is advisory-lock-then-row-write on both sides, so the two - paths cannot deadlock against each other. - */ - await layer.transactionImmediate(async (tx) => { - await acquireTaskAdvisoryXactLock(tx, projectId, taskId); - await tx - .insert(schema.project.taskWorkflowSelection) - .values({ projectId, taskId, workflowId, stepIds, updatedAt }) - .onConflictDoUpdate({ - target: [ - schema.project.taskWorkflowSelection.projectId, - schema.project.taskWorkflowSelection.taskId, - ], - set: { workflowId, stepIds, updatedAt }, - }); - }); - return; + const updatedAt = new Date().toISOString(); + /* FNXC:PostgresCutover 2026-07-04-00:00: backend selection writes use async Drizzle. */ + const layer = store.asyncLayer!; + /* FNXC:SqliteDualPathCleanup 2026-07-26-15:00: selection identity is project-scoped. */ + const projectId = layer.projectId?.trim() || "__legacy_unscoped__"; + /* FNXC:WorkflowCapacity 2026-07-28-18:05: take the cross-node advisory lock before the selection write. */ + await layer.transactionImmediate(async (tx) => { + await acquireTaskAdvisoryXactLock(tx, projectId, taskId); + await tx + .insert(schema.project.taskWorkflowSelection) + .values({ projectId, taskId, workflowId, stepIds, updatedAt }) + .onConflictDoUpdate({ + target: [ + schema.project.taskWorkflowSelection.projectId, + schema.project.taskWorkflowSelection.taskId, + ], + set: { workflowId, stepIds, updatedAt }, + }); + }); + store.laneCache.invalidate(taskId); } export async function removeMaterializedSelectionImpl(store: TaskStore, taskId: string): Promise { - /* - FNXC:PostgresCutover 2026-07-04-00:00: - Backend-mode delete reuses purgeTaskWorkflowSelectionRowsAsyncImpl (read stepIds, delete workflow_steps children, delete the selection row) so PG stays in lockstep with the SQLite path. - */ - await purgeTaskWorkflowSelectionRowsAsyncImpl(store, taskId); - return; + /* FNXC:PostgresCutover 2026-07-04-00:00: materialized selection deletion also removes child steps. */ + await purgeTaskWorkflowSelectionRowsAsyncImpl(store, taskId); + store.laneCache.invalidate(taskId); } export function purgeTaskWorkflowSelectionRowsImpl(store: TaskStore, taskId: string): void { @@ -661,6 +637,7 @@ export async function purgeTaskWorkflowSelectionRowsAsyncImpl(store: TaskStore, } await layer.db.delete(schema.project.taskWorkflowSelection).where(eq(schema.project.taskWorkflowSelection.taskId, taskId)); store.workflowStepsCache = null; + store.laneCache.invalidate(taskId); } export async function cleanupOrphanedMaterializedStepsImpl(store: TaskStore, stepIds: string[] | undefined): Promise { diff --git a/packages/engine/src/__tests__/scheduler-task-updated-lanes.test.ts b/packages/engine/src/__tests__/scheduler-task-updated-lanes.test.ts new file mode 100644 index 0000000000..c17fb15286 --- /dev/null +++ b/packages/engine/src/__tests__/scheduler-task-updated-lanes.test.ts @@ -0,0 +1,108 @@ +import { describe, expect, it, vi } from "vitest"; +import type { Task, TaskStore } from "@fusion/core"; + +vi.mock("@fusion/core", async (importOriginal) => ({ + ...(await importOriginal()), + getCurrentRepo: () => ({ owner: "fusion", repo: "test" }), +})); + +import { Scheduler } from "../scheduler.js"; + +type Listener = (...args: unknown[]) => void; + +function createStore() { + const listeners = new Map(); + const resolveTaskWorkflowIrSync = vi.fn(() => ({ columns: [] })); + const store = { + on: vi.fn((event: string, listener: Listener) => listeners.set(event, [...(listeners.get(event) ?? []), listener])), + off: vi.fn(), + getRootDir: () => "/test/project", + getSettings: vi.fn().mockResolvedValue({ globalPause: false, enginePaused: false }), + listTasks: vi.fn().mockResolvedValue([]), + resolveTaskWorkflowIrSync, + } as unknown as TaskStore; + return { + store, + resolveTaskWorkflowIrSync, + emit(task: Task, meta?: { lanes?: { wip?: string; review?: string } }) { + for (const listener of listeners.get("task:updated") ?? []) listener(task, meta); + }, + }; +} + +function task(overrides: Partial): Task { + return { + id: "FN-update-lanes", + title: "lane test", + description: "", + column: "building", + status: null, + dependencies: [], + steps: [], + currentStep: 0, + log: [], + createdAt: "2026-01-01T00:00:00.000Z", + updatedAt: "2026-01-01T00:00:00.000Z", + ...overrides, + } as Task; +} + +/* +FNXC:WorkflowEvents 2026-08-01-06:29: +The scheduler's update handler must act in the emitter's synchronous tick. These cases pin the +renamed workflow answer, the unknown-metadata literal fallback, and the tracked-map re-entrance +barrier so a future resolver/await conversion cannot silently restore default-board behavior. +*/ +describe("scheduler task:updated lane metadata", () => { + it("records renamed-wip failures and starts renamed-review PR monitoring from payload lanes", () => { + const { store, emit } = createStore(); + const startMonitoring = vi.fn(); + const tracked = new Map(); + const scheduler = new Scheduler(store, { + prMonitor: { startMonitoring, getTrackedPrs: () => tracked, updatePrInfo: vi.fn() }, + } as never); + + emit(task({ id: "FN-failed", status: "failed", sliceId: "slice-1", column: "building" }), { lanes: { wip: "building", review: "reviewing" } }); + emit(task({ id: "FN-pr", column: "reviewing", prInfo: { number: 7, url: "https://example.invalid/7", branch: "fusion/FN-pr" } }), { lanes: { wip: "building", review: "reviewing" } }); + + expect((scheduler as unknown as { failedTaskIds: Set }).failedTaskIds).toContain("FN-failed"); + expect(startMonitoring).toHaveBeenCalledWith("FN-pr", "fusion", "test", expect.anything()); + }); + + it("keeps absent metadata on the legacy literal path without consulting the PostgreSQL sync resolver", () => { + const { store, emit, resolveTaskWorkflowIrSync } = createStore(); + const startMonitoring = vi.fn(); + const scheduler = new Scheduler(store, { + prMonitor: { startMonitoring, getTrackedPrs: () => new Map(), updatePrInfo: vi.fn() }, + } as never); + + emit(task({ id: "FN-custom", status: "failed", sliceId: "slice-1", column: "building" })); + emit(task({ id: "FN-default", status: "failed", sliceId: "slice-1", column: "in-progress" })); + emit(task({ id: "FN-custom-pr", column: "reviewing", prInfo: { number: 8, url: "https://example.invalid/8", branch: "fusion/FN-custom" } })); + + const failed = (scheduler as unknown as { failedTaskIds: Set }).failedTaskIds; + expect(failed.has("FN-custom")).toBe(false); + expect(failed.has("FN-default")).toBe(true); + expect(startMonitoring).not.toHaveBeenCalled(); + expect(resolveTaskWorkflowIrSync).not.toHaveBeenCalled(); + }); + + it("does not double-start a monitor on repeated update events", () => { + const { store, emit } = createStore(); + const startMonitoring = vi.fn(); + const tracked = new Map(); + const scheduler = new Scheduler(store, { + prMonitor: { + startMonitoring: (...args: unknown[]) => { startMonitoring(...args); tracked.set(args[0] as string, {}); }, + getTrackedPrs: () => tracked, + updatePrInfo: vi.fn(), + }, + } as never); + const prTask = task({ id: "FN-once", column: "reviewing", prInfo: { number: 9, url: "https://example.invalid/9", branch: "fusion/FN-once" } }); + + emit(prTask, { lanes: { review: "reviewing" } }); + emit(prTask, { lanes: { review: "reviewing" } }); + + expect(startMonitoring).toHaveBeenCalledTimes(1); + }); +}); diff --git a/packages/engine/src/__tests__/task-updated-lanes-bridge-compat.test.ts b/packages/engine/src/__tests__/task-updated-lanes-bridge-compat.test.ts new file mode 100644 index 0000000000..39cc4fa2d0 --- /dev/null +++ b/packages/engine/src/__tests__/task-updated-lanes-bridge-compat.test.ts @@ -0,0 +1,94 @@ +import { EventEmitter } from "node:events"; +import { describe, expect, it, vi } from "vitest"; +import type { Task, TaskStore } from "@fusion/core"; + +vi.mock("@fusion/core", async (importOriginal) => ({ + ...(await importOriginal()), + getCurrentRepo: () => ({ owner: "fusion", repo: "test" }), +})); + +import { RemoteNodeRuntime } from "../runtimes/remote-node-runtime.js"; +import { Scheduler } from "../scheduler.js"; +import { TriageProcessor } from "../triage.js"; + + +function createStore() { + const store = Object.assign(new EventEmitter(), { + getRootDir: () => "/test/project", + getSettings: vi.fn().mockResolvedValue({ globalPause: false, enginePaused: false }), + listTasks: vi.fn().mockResolvedValue([]), + logEntry: vi.fn().mockResolvedValue(undefined), + updateTask: vi.fn().mockResolvedValue(undefined), + getTask: vi.fn(), + resolveTaskWorkflowIrSync: vi.fn(() => ({ columns: [] })), + }) as unknown as TaskStore & EventEmitter; + return store; +} + +function task(overrides: Partial = {}): Task { + return { + id: "FN-bridge", + title: "bridge compatibility", + description: "", + column: "building", + status: "planning", + dependencies: [], + steps: [], + currentStep: 0, + log: [], + createdAt: "2026-01-01T00:00:00.000Z", + updatedAt: "2026-01-01T00:00:00.000Z", + ...overrides, + } as Task; +} + +/* +FNXC:WorkflowEvents 2026-08-01-06:40: +RemoteNodeRuntime is a real runtime boundary, not a synthetic EventEmitter stand-in. It deliberately +re-emits only the serialized Task argument, so a receiving store observes lanes as unknown and its +scheduler and triage listeners retain their literal fallback behavior. +*/ +describe("task:updated runtime bridge compatibility", () => { + it("drops lanes through RemoteNodeRuntime while scheduler and triage retain their absent-meta fallbacks", () => { + const store = createStore(); + const startMonitoring = vi.fn(); + const scheduler = new Scheduler(store, { + prMonitor: { startMonitoring, getTrackedPrs: () => new Map(), updatePrInfo: vi.fn() }, + } as never); + const triage = new TriageProcessor(store, "/test/project"); + triage.start(); + const session = { abort: vi.fn().mockResolvedValue(undefined), dispose: vi.fn() }; + (triage as any).activeSessions.set("FN-bridge", session); + vi.spyOn(triage as any, "recordTriageSessionTokenUsageSoon").mockImplementation(() => undefined); + + const runtime = new RemoteNodeRuntime({ + nodeConfig: { id: "node-1", name: "node", url: "https://remote.invalid", apiKey: "token" }, + projectId: "project-1", + projectName: "Project 1", + }); + const forwarded: Array<[Task, unknown]> = []; + runtime.on("task:updated", (updated, meta) => { + forwarded.push([updated, meta]); + store.emit("task:updated", updated, meta); + }); + + // Exercise the runtime's actual remote-event forwarding method. The remote payload can contain + // a renamed lane answer, but the runtime's serialized event contract must not forward it as meta. + (runtime as any).forwardRemoteEvent({ + type: "task:updated", + payload: task({ + status: "failed", + sliceId: "slice-1", + column: "building", + prInfo: { number: 7, url: "https://example.invalid/7", branch: "fusion/FN-bridge" }, + }), + }); + + expect(forwarded).toHaveLength(1); + expect(forwarded[0][1]).toBeUndefined(); + expect((scheduler as any).failedTaskIds.has("FN-bridge")).toBe(false); + expect(startMonitoring).not.toHaveBeenCalled(); + expect(session.dispose).toHaveBeenCalledOnce(); + triage.stop(); + }); +}); diff --git a/packages/engine/src/__tests__/task-updated-lanes-engine-emit-surfaces.test.ts b/packages/engine/src/__tests__/task-updated-lanes-engine-emit-surfaces.test.ts new file mode 100644 index 0000000000..2dd004d15b --- /dev/null +++ b/packages/engine/src/__tests__/task-updated-lanes-engine-emit-surfaces.test.ts @@ -0,0 +1,101 @@ +import { EventEmitter } from "node:events"; +import { describe, expect, it } from "vitest"; +import { readdirSync, readFileSync, statSync } from "node:fs"; +import { join, relative, resolve } from "node:path"; +import { TaskStore, type Task } from "@fusion/core"; +import { HybridExecutor } from "../hybrid-executor.js"; +import { ProjectManager } from "../project-manager.js"; +import { ChildProcessRuntime } from "../runtimes/child-process-runtime.js"; +import { InProcessRuntime } from "../runtimes/in-process-runtime.js"; +import { RemoteNodeRuntime } from "../runtimes/remote-node-runtime.js"; +import { TASK_UPDATED } from "../ipc/ipc-protocol.js"; + +const REPO_ROOT = resolve(__dirname, "../../../.."); +const ENGINE_ROOT = join(REPO_ROOT, "packages/engine/src"); + +const DROP_BRIDGES = { + "packages/engine/src/hybrid-executor.ts": "HybridExecutor", + "packages/engine/src/project-manager.ts": "ProjectManager", + "packages/engine/src/runtimes/child-process-runtime.ts": "ChildProcessRuntime", + "packages/engine/src/runtimes/in-process-runtime.ts": "InProcessRuntime", + "packages/engine/src/runtimes/remote-node-runtime.ts": "RemoteNodeRuntime", +} as const; + +function* walk(dir: string): Generator { + for (const entry of readdirSync(dir)) { + if (entry === "__tests__" || entry === "dist") continue; + const path = join(dir, entry); + if (statSync(path).isDirectory()) yield* walk(path); + else if (path.endsWith(".ts")) yield path; + } +} + +const task = { id: "FN-bridge-surface", column: "building" } as Task; + +/** + * FNXC:WorkflowEvents 2026-08-01-06:57: + * Runtime and manager update emitters are deliberately tested by executing their actual forwarding + * registrations. They are EventEmitter DROP boundaries rather than TaskStore seams: only the Task + * crosses the serialized/runtime boundary, and the receiving listener treats missing lanes as unknown. + */ +describe("engine task:updated emit surface", () => { + it("registers every engine emitter as an explicit DROP bridge", () => { + const emittingModules = [...walk(ENGINE_ROOT)] + .filter((file) => readFileSync(file, "utf8").includes('emit("task:updated"')) + .map((file) => relative(REPO_ROOT, file).split("\\").join("/")) + .sort(); + + expect(emittingModules).toEqual(Object.keys(DROP_BRIDGES).sort()); + for (const file of emittingModules) { + const source = readFileSync(join(REPO_ROOT, file), "utf8"); + expect(source).toMatch(new RegExp(`${DROP_BRIDGES[file as keyof typeof DROP_BRIDGES]}\\s+extends\\s+EventEmitter`)); + expect(source).toContain('this.emit("task:updated"'); + } + }); + + it("executes each DROP bridge with a well-formed lanes-free task payload", () => { + const hybridUpstream = new EventEmitter(); + const hybrid = Object.assign(new EventEmitter(), { projectManager: hybridUpstream }); + (HybridExecutor.prototype as any).setupEventForwarding.call(hybrid); + + const managerUpstream = new EventEmitter(); + const manager = Object.assign(new EventEmitter(), { logActivity: async () => undefined }); + (ProjectManager.prototype as any).setupEventForwarding.call(manager, managerUpstream, "project", "Project"); + + const inProcessUpstream = new EventEmitter(); + const inProcess = Object.assign(new EventEmitter(), { + taskStore: inProcessUpstream, + recordActivity: () => undefined, + config: { projectId: "project" }, + }); + (InProcessRuntime.prototype as any).setupEventForwarding.call(inProcess); + + const childUpstream = new EventEmitter(); + const child = Object.assign(new EventEmitter(), { ipcHost: childUpstream }); + (ChildProcessRuntime.prototype as any).setupEventForwarding.call(child); + + const remote = new RemoteNodeRuntime({ + nodeConfig: { id: "node", name: "node", url: "https://remote.invalid", apiKey: "test" }, + projectId: "project", + projectName: "Project", + }); + + const cases: Array<{ receiver: EventEmitter; upstream: EventEmitter; emit: () => void; unwrap?: (value: unknown) => unknown }> = [ + { receiver: hybrid, upstream: hybridUpstream, emit: () => hybridUpstream.emit("task:updated", task) }, + { receiver: manager, upstream: managerUpstream, emit: () => managerUpstream.emit("task:updated", task), unwrap: (value) => (value as { task: Task }).task }, + { receiver: inProcess, upstream: inProcessUpstream, emit: () => inProcessUpstream.emit("task:updated", task) }, + { receiver: child, upstream: childUpstream, emit: () => childUpstream.emit(TASK_UPDATED, { task }) }, + { receiver: remote, upstream: remote, emit: () => (remote as any).forwardRemoteEvent({ type: "task:updated", payload: task }) }, + ]; + + for (const bridge of cases) { + expect(bridge.receiver).not.toBeInstanceOf(TaskStore); + const received: unknown[][] = []; + bridge.receiver.on("task:updated", (...args: unknown[]) => received.push(args)); + bridge.emit(); + expect(received).toHaveLength(1); + expect(bridge.unwrap?.(received[0][0]) ?? received[0][0]).toBe(task); + expect(received[0][1]).toBeUndefined(); + } + }); +}); diff --git a/packages/engine/src/__tests__/triage-pause-abort.test.ts b/packages/engine/src/__tests__/triage-pause-abort.test.ts index d6d7241411..12450d7840 100644 --- a/packages/engine/src/__tests__/triage-pause-abort.test.ts +++ b/packages/engine/src/__tests__/triage-pause-abort.test.ts @@ -183,6 +183,28 @@ describe("TriageProcessor per-task pause aborts", () => { processor.stop(); }); + /* + FNXC:WorkflowEvents 2026-08-01-06:29: + A renamed execution lane is not an evacuation. The second argument supplies that synchronous + distinction; without it, unknown metadata deliberately follows the historic literal fallback. + */ + it("uses payload lanes for renamed planning evacuation and retains the absent-meta fallback", () => { + const { store, emit } = createEventedStore(); + const processor = new TriageProcessor(store, "/tmp/root"); + const retained = { abort: vi.fn().mockResolvedValue(undefined), dispose: vi.fn() }; + const evacuated = { abort: vi.fn().mockResolvedValue(undefined), dispose: vi.fn() }; + processor.start(); + (processor as any).activeSessions.set("FN-RENAMED-WIP", retained); + (processor as any).activeSessions.set("FN-UNKNOWN", evacuated); + + emit("task:updated", { id: "FN-RENAMED-WIP", column: "building", status: "planning" }, { lanes: { hold: "drafting", intake: "inbox", wip: "building" } }); + emit("task:updated", { id: "FN-UNKNOWN", column: "building", status: "planning" }); + + expect(retained.dispose).not.toHaveBeenCalled(); + expect(evacuated.dispose).toHaveBeenCalledOnce(); + processor.stop(); + }); + it("detaches the task:updated pause listener on stop", () => { const { store, emit } = createEventedStore(); const processor = new TriageProcessor(store, "/tmp/root"); diff --git a/packages/engine/src/__tests__/triage-planning-wake.test.ts b/packages/engine/src/__tests__/triage-planning-wake.test.ts index 20ee040fb8..9ab184a42d 100644 --- a/packages/engine/src/__tests__/triage-planning-wake.test.ts +++ b/packages/engine/src/__tests__/triage-planning-wake.test.ts @@ -134,6 +134,31 @@ describe("TriageProcessor planning wake (immediate poll on move)", () => { } }); + /* + FNXC:WorkflowEvents 2026-08-01-07:21: + Renamed planner lanes are authoritative only when task:updated carries cache-warmed metadata. + A cold cache or runtime bridge is unknown, so it must retain the builtin literal fallback instead + of synchronously resolving PostgreSQL's default workflow. + */ + it("uses payload lanes for renamed wake columns and keeps absent metadata on the legacy fallback", async () => { + const { store, emit } = createEventedStore(); + const processor = new TriageProcessor(store, "/tmp/root"); + processor.start(); + const poll = vi.spyOn(processor as any, "poll").mockResolvedValue(undefined); + + emit("task:updated", createTask({ id: "FN-RENAMED-WAKE", column: "drafting" }), { + lanes: { hold: "drafting", intake: "inbox" }, + }); + await settleWake(); + expect(poll).toHaveBeenCalledTimes(1); + + poll.mockClear(); + emit("task:updated", createTask({ id: "FN-UNKNOWN-RENAMED", column: "drafting" })); + await settleWake(); + expect(poll).not.toHaveBeenCalled(); + processor.stop(); + }); + it("coalesces a burst of moves into a single poll", async () => { const { store, emit } = createEventedStore(); const processor = new TriageProcessor(store, "/tmp/root"); diff --git a/packages/engine/src/notification/__tests__/notification-renamed-lifecycle-columns.test.ts b/packages/engine/src/notification/__tests__/notification-renamed-lifecycle-columns.test.ts index 0172745dea..9fd0464c72 100644 --- a/packages/engine/src/notification/__tests__/notification-renamed-lifecycle-columns.test.ts +++ b/packages/engine/src/notification/__tests__/notification-renamed-lifecycle-columns.test.ts @@ -20,7 +20,7 @@ REVERT CHECK, measured (both run): Both pass on the DEFAULT vocabulary before and after, which is the point of running both. */ import { describe, expect, it, vi } from "vitest"; -import type { NotificationProvider, Settings, Task, WorkflowIr } from "@fusion/core"; +import type { NotificationProvider, Settings, Task, TaskMoveLanes, WorkflowIr } from "@fusion/core"; import { NotificationService } from "../notification-service.js"; import { DEFAULT_VOCAB, RENAMED_VOCAB, lifecycleIr, type Vocabulary } from "../../__tests__/_workflow-vocabulary-fixture.js"; import { flushAsyncHandlers } from "../../__tests__/_flush-async-handlers.js"; @@ -30,6 +30,7 @@ vi.mock("../../logger.js", () => ({ })); type MovedListener = (data: { task: Task; from: string; to: string }) => void; +type UpdatedListener = (task: Task, meta?: { lanes?: TaskMoveLanes }) => void; /** * A store that resolves a real workflow IR, so `resolveTaskLifecycleColumns` returns the vocabulary @@ -54,16 +55,20 @@ function fixture(vocab: Vocabulary) { does carry merge orchestration, which is why this is a gap on custom boards and not a live outage. */ const ir: WorkflowIr = lifecycleIr(vocab, "notif-lifecycle", { mergeOrchestration: true }); - const listeners = new Set(); + const movedListeners = new Set(); + const updatedListeners = new Set(); const store = { getSettings: async () => ({ ntfyEnabled: true, ntfyTopic: "test" }) as Settings, getTaskWorkflowSelection: () => ({ workflowId: "notif-lifecycle", stepIds: [] }), getWorkflowDefinition: async (id: string) => (id === "notif-lifecycle" ? { ir } : undefined), - on: (event: string, listener: MovedListener) => { - if (event === "task:moved") listeners.add(listener); + on: (event: string, listener: MovedListener | UpdatedListener) => { + if (event === "task:moved") movedListeners.add(listener as MovedListener); + if (event === "task:updated") updatedListeners.add(listener as UpdatedListener); }, off: () => undefined, - emitMoved: (data: { task: Task; from: string; to: string }) => listeners.forEach((l) => l(data)), + emitMoved: (data: { task: Task; from: string; to: string }) => movedListeners.forEach((listener) => listener(data)), + emitUpdated: (updatedTask: Task, meta?: { lanes?: TaskMoveLanes }) => + updatedListeners.forEach((listener) => listener(updatedTask, meta)), }; const sendNotification = vi.fn(async () => ({ success: true, providerId: "test" })); @@ -136,6 +141,75 @@ describe("notification lifecycle guards resolve columns by ROLE, not by id", () }); } + it("classifies a manual merge hold from emitter-carried renamed review lanes synchronously", async () => { + const { store, service, sendNotification, task } = fixture(RENAMED_VOCAB); + await service.start(); + + store.emitUpdated( + task({ column: RENAMED_VOCAB.review, paused: true, pausedReason: "manual-hold", status: "in-review" }), + { lanes: { review: RENAMED_VOCAB.review } }, + ); + + // No promise turn: the synchronous listener must classify before its queued wedge work runs. + expect(sendNotification).toHaveBeenCalledWith( + "workflow-notify", + expect.objectContaining({ + taskId: "FN-9001", + metadata: expect.objectContaining({ + notificationKind: "manual_merge_hold", + notificationDedupeKey: "workflow-transition:FN-9001:manual-merge-hold", + }), + }), + ); + await service.stop(); + }); + + it("keeps absent or blank review lanes on the existing in-review fallback", async () => { + const { store, service, sendNotification, task } = fixture(RENAMED_VOCAB); + await service.start(); + + const held = (column: string) => task({ column, paused: true, pausedReason: "manual-hold", status: "in-review" }); + store.emitUpdated(held(RENAMED_VOCAB.review)); + store.emitUpdated(held(RENAMED_VOCAB.review), { lanes: { review: "" } }); + store.emitUpdated(held("in-review")); + store.emitUpdated(held("in-review"), { lanes: { review: "" } }); + await flushAsyncHandlers(); + + // Only the default-lane holds notify; absent and blank metadata are unknown rather than renamed. + expect(sendNotification).toHaveBeenCalledTimes(1); + expect(sendNotification).toHaveBeenCalledWith("workflow-notify", expect.anything()); + await service.stop(); + }); + + it("preserves marker, failed, and dedupe behavior independently of lane metadata", async () => { + const { store, service, sendNotification, task } = fixture(RENAMED_VOCAB); + await service.start(); + + const marker = { + kind: "manual-merge-hold" as const, + transitionId: "marker-hold", + column: RENAMED_VOCAB.review, + createdAt: "2026-08-01T07:44:00.000Z", + }; + store.emitUpdated(task({ id: "FN-marker", column: RENAMED_VOCAB.review, status: "in-review", workflowTransitionNotification: marker })); + store.emitUpdated(task({ id: "FN-failed", column: RENAMED_VOCAB.review, status: "failed", paused: true, pausedReason: "manual-hold" }), { lanes: { review: RENAMED_VOCAB.review } }); + const duplicate = task({ id: "FN-duplicate", column: RENAMED_VOCAB.review, status: "in-review", paused: true, pausedReason: "manual-hold" }); + store.emitUpdated(duplicate, { lanes: { review: RENAMED_VOCAB.review } }); + store.emitUpdated(duplicate, { lanes: { review: RENAMED_VOCAB.review } }); + await flushAsyncHandlers(); + + expect(sendNotification).toHaveBeenCalledTimes(2); + expect(sendNotification).toHaveBeenCalledWith( + "workflow-notify", + expect.objectContaining({ taskId: "FN-marker", metadata: expect.objectContaining({ notificationDedupeKey: "workflow-transition:FN-marker:marker-hold" }) }), + ); + expect(sendNotification).toHaveBeenCalledWith( + "workflow-notify", + expect.objectContaining({ taskId: "FN-duplicate", metadata: expect.objectContaining({ notificationDedupeKey: "workflow-transition:FN-duplicate:manual-merge-hold" }) }), + ); + await service.stop(); + }); + it("does not dispatch for a move into a lane that plays no notable role", async () => { /* Non-vacuous check on both conversions at once: without it, a service that notified on EVERY move diff --git a/packages/engine/src/notification/__tests__/task-wedge-notification.test.ts b/packages/engine/src/notification/__tests__/task-wedge-notification.test.ts index 29b28e8ca2..4c3c82fa5e 100644 --- a/packages/engine/src/notification/__tests__/task-wedge-notification.test.ts +++ b/packages/engine/src/notification/__tests__/task-wedge-notification.test.ts @@ -67,6 +67,25 @@ describe("task wedge notifications", () => { await service.stop(); }); + /* + FNXC:TaskWedgeNotifications 2026-08-01-07:44: + A recovery and re-wedge can be emitted back-to-back by synchronous task lifecycle writers. The + per-task chain must run the recovery's resolve before the second wedge's claim; otherwise the old + active episode rejects the claim and drops the second operator alert. + */ + it("serializes back-to-back recovery and re-wedge task updates", async () => { + const { store, service, sendMessageOnce, task } = fixture(); + await service.start(); + + store.emit(task()); + await vi.waitFor(() => expect(sendMessageOnce).toHaveBeenCalledTimes(1)); + store.emit(task({ status: "queued", error: undefined, column: "todo", updatedAt: "2026-07-22T12:02:00.000Z" })); + store.emit(task({ updatedAt: "2026-07-22T12:03:00.000Z" })); + + await vi.waitFor(() => expect(sendMessageOnce).toHaveBeenCalledTimes(2)); + await service.stop(); + }); + /* FNXC:WorkflowResolvedColumns 2026-07-31-18:50 (fleet): Clearing a wedge episode asks "has the card moved on?", which was four column literals. On a renamed diff --git a/packages/engine/src/notification/notification-service.ts b/packages/engine/src/notification/notification-service.ts index 5435de38b7..e47d258e1b 100644 --- a/packages/engine/src/notification/notification-service.ts +++ b/packages/engine/src/notification/notification-service.ts @@ -10,7 +10,7 @@ import type { Settings, Task, } from "@fusion/core"; -import type { LifecycleColumns, WorkflowIrResolverStore } from "@fusion/core"; +import type { LifecycleColumns, TaskMoveLanes, WorkflowIrResolverStore } from "@fusion/core"; import { DASHBOARD_USER_ID, NotificationDispatcher, resolveProjectColumnsForRoles, resolveReviewColumns, resolveTaskLifecycleColumns, resolveWorkflowIrForTask } from "@fusion/core"; import { DEFAULT_NTFY_EVENTS, buildNtfyClickUrl, formatTaskIdentifier } from "../notifier.js"; import { schedulerLog } from "../logger.js"; @@ -37,7 +37,7 @@ export interface NotificationServiceOptions { interface NotificationServiceStoreEvents { "task:created": [task: Task]; "task:moved": [data: { task: Task; from: Column; to: Column }]; - "task:updated": [task: Task]; + "task:updated": [task: Task, meta?: { lanes?: TaskMoveLanes }]; "task:merged": [result: MergeResult]; "settings:updated": [payload: { settings: Settings; previous: Settings }]; } @@ -363,7 +363,7 @@ export class NotificationService { } }; - private handleTaskUpdated = (task: Task): void => { + private handleTaskUpdated = (task: Task, meta?: { lanes?: TaskMoveLanes }): void => { /* FNXC:TaskWedgeNotifications 2026-07-22-20:00: FN-5627 transient merge failures retain an active recovery owner despite @@ -457,7 +457,7 @@ export class NotificationService { ); } - const workflowTransition = this.classifyWorkflowTransitionNotification(task); + const workflowTransition = this.classifyWorkflowTransitionNotification(task, meta?.lanes?.review); if (workflowTransition) { this.maybeNotify( task.id, @@ -1149,7 +1149,10 @@ export class NotificationService { typeof task.mergeDetails?.mergedAt === "string"; } - private classifyWorkflowTransitionNotification(task: Task): { event: NotificationEvent; metadata: Record } | null { + private classifyWorkflowTransitionNotification( + task: Task, + reviewLane?: string, + ): { event: NotificationEvent; metadata: Record } | null { /* * FNXC:WorkflowNotifications 2026-06-29-11:50: * Workflow-specific operator waits should notify from the durable task update that already represents the wait, not from a new lifecycle bus. Plan/remediation await-input, workflow CLI approval, manual merge holds, and workflow recovery requeues each use a stable dedupe key so repeated task:updated emissions stay quiet while unrelated task notifications can still fire. @@ -1179,7 +1182,7 @@ export class NotificationService { } const typedWorkflowTransition = this.workflowTransitionNotificationMarker(task); - if (task.status !== "failed" && (this.isManualMergeHold(task) || typedWorkflowTransition?.kind === "manual-merge-hold")) { + if (task.status !== "failed" && (this.isManualMergeHold(task, reviewLane) || typedWorkflowTransition?.kind === "manual-merge-hold")) { return { event: "workflow-notify", metadata: { @@ -1225,36 +1228,19 @@ export class NotificationService { } /* - FNXC:WorkflowResolvedColumns 2026-07-30-23:10 (fleet phase — FLAGGED AND LEFT COUNTED): - The last review-lane id in this file, and the one conversion that is not mechanical. This predicate is - SYNC and its only caller, `classifyWorkflowTransitionNotification`, is sync too — reached from the - `handleTaskUpdated` listener, which the store invokes as `(task: Task): void`. + FNXC:WorkflowResolvedColumns 2026-08-01-07:42: + `task:updated` remains a synchronous listener: its prologue schedules wedge and mailbox work before + notification classification, so resolving workflow IR here would defer that ordering and make the + lane conversion inert under PostgreSQL's default-only sync resolver. - Converting it therefore means making the whole chain async, which turns a synchronous listener body - into fire-and-forget and reorders notification classification against every other `task:updated` - handler. That is a behaviour change to notification ordering, not a column conversion, so it is out of - fleet scope. - - Threading a pre-resolved `LifecycleColumns` in as a parameter is the likely fix — the resolution has to - happen in `handleTaskUpdated`, which would then pay it on every task update, so it wants the same - gate-placement judgement applied to the sites above rather than a mechanical pass. - - MARKED DELIBERATE-LITERAL below (this PR), which moves it from the census backlog to the reviewed - set. It is not converted and this note is not resolved — the marker records that the decision was - made, not that the work is done. Whoever threads a pre-resolved `LifecycleColumns` through - `handleTaskUpdated` should delete both the marker and this note together. + FN-8658 instead carries the emitter's cache-warmed `TaskMoveLanes` answer on the event. The review + lane is passed explicitly through this synchronous chain, while `enqueueWedgeHandling` already + serializes resolve-then-claim wedge episodes per task. Missing or blank metadata means unknown, not + legacy, and preserves the established `in-review` fallback exactly. */ - private isManualMergeHold(task: Task): boolean { - /* DELIBERATE-LITERAL — see the note above: this method and its only caller - (`classifyWorkflowTransitionNotification`) are SYNC, reached from the `handleTaskUpdated` - listener the store invokes as `(task: Task): void`. Resolving a lane here makes that whole - chain async, turning a synchronous listener body into fire-and-forget and reordering - notification classification against every other `task:updated` handler — a behaviour change to - notification ordering rather than a column conversion. */ - if (task.column !== "in-review") { - return false; - } - return task.pausedReason === "manual-hold"; + private isManualMergeHold(task: Task, reviewLane?: string): boolean { + const resolvedReviewLane = reviewLane?.trim() || "in-review"; + return task.column === resolvedReviewLane && task.pausedReason === "manual-hold"; } private workflowTransitionNotificationMarker(task: Task): Task["workflowTransitionNotification"] | undefined { diff --git a/packages/engine/src/scheduler.ts b/packages/engine/src/scheduler.ts index a75f3c3274..93e850913e 100644 --- a/packages/engine/src/scheduler.ts +++ b/packages/engine/src/scheduler.ts @@ -1205,7 +1205,7 @@ export class Scheduler { * PR Monitoring: Start monitoring when PR is linked to an in-review task. * Also detects task-level unpause transitions and triggers immediate scheduling. */ - this.store.on("task:updated", (task) => { + this.store.on("task:updated", (task, meta) => { const nextFingerprint = computeAutoClaimFingerprint(task); const previousFingerprint = this.lastAutoClaimFingerprint.get(task.id); if (!previousFingerprint || previousFingerprint !== nextFingerprint) { @@ -1213,51 +1213,17 @@ export class Scheduler { this.options.snapshotManager?.invalidate("task:updated"); } /* - FNXC:WorkflowResolvedColumns 2026-07-31-23:58 (FLAGGED AND LEFT COUNTED — do NOT convert with - `resolveTaskParkedColumnsSync`): - This literal and the `in-review` one further down are the two the sync-lane pass did not take, - and nothing in this file said why. Converting them the way the other ten were converted would - make them INERT, not fixed: `getTaskWorkflowSelectionImpl` returns `undefined` unconditionally - under PostgreSQL, so `resolveTaskWorkflowIrSync` always answers with the DEFAULT builtin IR and - every lane it yields is the legacy id (proved in `postgres/sync-workflow-ir-is-always-default.pg.test.ts`; - `check-inert-sync-lane-conversions` baselines the twenty guards already in that state here). - - They stay literal and COUNTED, which is the honest state: an unconverted literal is at least - visible to the census, while an inert conversion leaves the backlog and takes the evidence with - it. Both live in a synchronous `task:updated` listener, so the async resolver is unavailable - without reordering this handler against every other subscriber. - - FNXC:WorkflowResolvedColumns 2026-07-31-23:59 (THE REASON CHANGED — #3128 made async resolution - available here, so "it would be inert" is no longer why these two stay): - #3128 converted the rest of this listener by deferring the resolve into `void (async () => ...)` - blocks, which reach the ASYNC resolver and are genuinely correct. So the sync-resolver argument - above no longer explains these two. The real reason is the one #3128's own note states, three - branches down: - - "The `planningTaskIds.delete` stays SYNCHRONOUS — it is the edge-trigger bookkeeping, and - deferring it would let a second update re-enter this branch." - - Both remaining literals are that case: - - `failedTaskIds.add` below is edge-trigger bookkeeping raced against `moveTask` clearing the - failure metadata — the comment on it says so. Deferring the add can miss that window. - - the PR-monitoring guard further down gates `prMonitor.getTrackedPrs()` / - `startMonitoring()`, where `tracked.has(task.id)` IS the re-entrance guard. Move the lane - answer behind an await and two updates for the same task can both pass that check before - either starts, double-starting a monitor. - - LEFT COUNTED, both of them: an unconverted literal is visible to the census, and marking these - exempt would assert the code is fine when it is blocked. - - So these are not waiting on a resolver. They are waiting on somewhere to put the answer that is - not behind an await — the emitter-carried `lanes` that #3109 added to `task:moved` would do it, - and extending that to `task:updated` is measured as expensive rather than impossible - (`sync-workflow-ir-second-blocker.test.ts`: 26 emit sites against 7, on the hottest write path). + FNXC:WorkflowEvents 2026-08-01-07:14: + TaskStore decorates `task:updated` with its cache-warmed lanes, letting the failure and PR + edge-trigger guards remain synchronous on renamed boards. A missing payload is unknown, not a + legacy claim, so its fallback stays literal instead of consulting PostgreSQL's default-only sync + resolver; that preserves existing bridge and cold-cache behaviour without reordering listeners. */ + const eventLanes = meta?.lanes; + // DELIBERATE-LITERAL — absent event metadata is unknown, so the legacy wip fallback stays explicit. // Track mission failure signals before moveTask clears failure metadata. if (task.sliceId && task.status === "failed") { - /* DELIBERATE-LITERAL — see the note above: converting this needs the async resolver on the - hottest write path (26 emit sites against 7, measured), not a signature change here. */ - if (task.column === "in-progress") this.failedTaskIds.add(task.id); + if (eventLanes ? task.column === eventLanes.wip : task.column === "in-progress") this.failedTaskIds.add(task.id); /* FNXC:MissionReconciliation 2026-08-01-00:00: In-place failure parks do not emit task:moved, but they release the @@ -1340,14 +1306,8 @@ export class Scheduler { } if (!this.options.prMonitor) return; - /* FNXC:WorkflowResolvedColumns 2026-07-31-23:59: the second of the two honest literals. NOT - because a conversion would be inert — #3128 made the async resolver reachable in this - listener — but because `tracked.has(task.id)` below is a re-entrance guard, and moving this - answer behind an await lets two updates for the same task both pass it and double-start a - monitor. LEFT COUNTED. See the fuller note on the mission-failure guard above. */ - /* DELIBERATE-LITERAL — see the note directly above: an await here lets two updates for the - same task both pass the `tracked.has` re-entrance guard and double-start a monitor. */ - if (task.column !== "in-review") return; + // DELIBERATE-LITERAL — runtime bridges drop lanes; never replace this unknown fallback with the sync resolver. + if (eventLanes ? task.column !== eventLanes.review : task.column !== "in-review") return; if (!task.prInfo) return; // Check if we're already monitoring this task diff --git a/packages/engine/src/triage.ts b/packages/engine/src/triage.ts index 8d69097c65..e9fe8c0fea 100644 --- a/packages/engine/src/triage.ts +++ b/packages/engine/src/triage.ts @@ -10,6 +10,7 @@ import type { AgentPermissionPolicy, PermanentAgentGatingContext, WorkflowIr, + TaskMoveLanes, } from "@fusion/core"; import { DUPLICATE_OF_METADATA_KEY, @@ -70,6 +71,16 @@ type TaskListFormatter = ( const TRIAGE_STUCK_RESUME_LOG_ACTION = "Triage stuck re-queue will resume existing planning draft"; const TRIAGE_STUCK_RESUME_FEEDBACK = "The previous triage session was killed by the stuck-task detector after writing a non-empty planning draft. Resume from the existing draft below: preserve useful structure and decisions, fill gaps, and continue toward review instead of restarting planning from scratch."; +/* +FNXC:WorkflowEvents 2026-08-01-07:21: +When a task:updated bridge omits lanes, planner membership is unknown and synchronous wake and + evacuation handlers retain their historic builtin-board fallback. Keep those compatibility sets +separate from the PostgreSQL sync resolver, which would falsely claim default lanes for renamed +workflows; metadata is the only authoritative renamed-lane answer in this event tick. +*/ +const LEGACY_PLANNER_WAKE_COLUMNS = new Set(["todo", "triage"]); +const LEGACY_PLANNER_COLUMNS = new Set([...LEGACY_PLANNER_WAKE_COLUMNS, "in-progress"]); + /* FNXC:PlanReviewReplan 2026-07-13-00:00: The triage pre-execution Plan Review gate (runPlanReviewBeforeExecution) routes a REVISE @@ -122,7 +133,7 @@ import type { AgentSession, } from "@earendil-works/pi-coding-agent"; import { ModelFallbackExhaustedError, describeModel, formatModelMarkerDetails, promptWithFallback } from "./pi.js"; -import { hasAdvancedPastPlanning, isTaskStillInPlanningStage, resolvePlannerLanes, resolvePlannerLanesForTaskAsync } from "./replan-target.js"; +import { hasAdvancedPastPlanning, isTaskStillInPlanningStage, resolvePlannerLanesForTaskAsync } from "./replan-target.js"; import { createResolvedAgentSession, extractRuntimeHint, @@ -448,9 +459,9 @@ export class TriageProcessor { private taskDeletedHandler?: (task: Task) => void; private taskPausedHandler?: (task: Task) => void; /** FNXC:CodingIdeasWorkflow 2026-07-25-11:20: store-event wake for planning-eligible columns. */ - private taskColumnWakeHandler?: (task: Task) => void; + private taskColumnWakeHandler?: (task: Task, meta?: { lanes?: TaskMoveLanes }) => void; /** FNXC:PlanningEvacuation 2026-07-25-23:00: stops planning when a card leaves the planner lanes. */ - private taskEvacuatedFromPlanningHandler?: (task: Task) => void; + private taskEvacuatedFromPlanningHandler?: (task: Task, meta?: { lanes?: TaskMoveLanes }) => void; private _approvalRequestStore?: ApprovalRequestStore; /** @@ -703,10 +714,12 @@ export class TriageProcessor { The handler is deliberately dumb: it filters on column only and delegates every real decision to the poll, so it cannot bypass a pause, dependency, seed-prompt, or concurrency gate. */ - this.taskColumnWakeHandler = (task: Task) => { + this.taskColumnWakeHandler = (task: Task, meta?: { lanes?: TaskMoveLanes }) => { if (!task?.id) return; - const wakeLanes = resolvePlannerLanes(this.store, task.id); - if (task.column !== wakeLanes.hold && task.column !== wakeLanes.intake) return; + const isPlannerWakeColumn = meta?.lanes + ? task.column === meta.lanes.hold || task.column === meta.lanes.intake + : LEGACY_PLANNER_WAKE_COLUMNS.has(task.column); + if (!isPlannerWakeColumn) return; if (task.paused === true || task.userPaused === true) return; // Already being planned (or mid-plan) — the running poll/session owns it. if (this.processing.has(task.id) || this.hasLivePlanningWork(task.id)) return; @@ -735,7 +748,7 @@ export class TriageProcessor { that legitimately advances into execution is not an evacuation — its session is already unwinding on its own. */ - this.taskEvacuatedFromPlanningHandler = (task: Task) => { + this.taskEvacuatedFromPlanningHandler = (task: Task, meta?: { lanes?: TaskMoveLanes }) => { if (!task?.id) return; /* Only an explicit, known destination column is evidence of evacuation. `task:updated` also @@ -745,76 +758,16 @@ export class TriageProcessor { */ if (typeof task.column !== "string") return; /* - FNXC:WorkflowResolvedColumns 2026-07-31-23:58 (RECONCILING TWO NOTES THAT CONTRADICTED EACH OTHER - — the earlier one was mine): - - My flag here said the third arm must not be converted with the same helper because that adds a - third INERT comparison. #3114 then converted it. Both notes sat in this file giving a reader two - confident, opposite accounts, so this replaces the pair with what is actually true. - - #3114 IS RIGHT ABOUT THE SHAPE AND THE BUG. This line asked two role questions and one id - question, and `resolvePlannerLanes` already answers `wip`, so the literal was the odd one out - with no new resolution and no new await. Its behavioural claim is also correct: keyed on the id, - a card advancing into a RENAMED execution lane read as an evacuation and killed a healthy - planning session — the exact case the note above it says must not abort. - - WHAT IT DOES NOT DO IS FIX THAT UNDER POSTGRESQL, and the evidence is mechanical rather than - argued: `resolvePlannerLanes` resolves through `resolveTaskWorkflowIrSync`, which cannot answer - for a CUSTOM workflow — the sync selection reader returns `undefined` unconditionally, AND the - custom-workflow IR read goes through `store.db`, whose implementation is an unconditional throw - (`sync-workflow-ir-second-blocker.test.ts`). All three arms therefore evaluate to `todo` / - `triage` / `in-progress` on every board the product ships. `check-inert-sync-lane-conversions` - records this file at SEVEN inert guards, which is where the truth now lives. - - SO READ THE ZERO CAREFULLY. This file's lifecycle-column-census count is now 0, and the census's - own `--triage` output warns that for a sync-resolved file "a count of 0 is the WORST case, not - the best — the file reads as fully converted". That is this file. The guard is uniform and - honest in shape, and still wrong on a renamed board. - - UNBLOCKING is not "convert the remaining arm" — there is none. It needs the answer to arrive - without the sync resolver: the emitter-carried lanes #3109 added for `task:moved`, extended to - `task:updated` (measured as NOT a cheap follow-on — 26 emit sites against 7, on the hottest - write path; see `sync-workflow-ir-second-blocker.test.ts`), or a sync reader that answers for - custom workflows. - - The guard's answer is also consumed SYNCHRONOUSLY — `pauseAborted.add`, `session.dispose()` and - `activeSessions.delete` all mutate in-memory state in this tick — so whatever supplies it must - not require an await. + FNXC:WorkflowEvents 2026-08-01-06:57: + `task:updated` now carries a cache-warmed lane answer for this synchronous evacuation guard. + Metadata can be absent at cache misses and runtime bridges, so that case remains unknown and + intentionally preserves the legacy planner-column fallback rather than consulting PostgreSQL's + default-only sync workflow resolver. */ - const disposeLanes = resolvePlannerLanes(this.store, task.id); - /* - FNXC:WorkflowResolvedColumns 2026-07-31-21:30 (fleet — the third lane on this line): - `disposeLanes.wip`, not the literal — the same resolver already answers the other two. - - This line asked two role questions and one id question. `resolvePlannerLanes` is already called - immediately above and its result carries `wip`, so no new resolution and no new await are - introduced: the literal simply stops being the odd one out. On a board whose execution lane is - renamed, the previous form treated a card advancing into execution as an EVACUATION and killed - a healthy planning session — the one case the note above says must not abort. - - `wip` is optional by design (PR #2628: a missing role stays undefined so callers refuse rather - than invent a column). Undefined here means the board declares no execution lane, so there is - no advance-into-execution to exclude and the comparison is correctly false. - - FNXC:WorkflowResolvedColumns 2026-07-31-23:58 (the conversion is REVERTED; the analysis is kept): - The bug described above is REAL and this is the clearest statement of it in the file, which is - why the paragraphs stay. The change did not fix it. - - Measured: `disposeLanes.wip` resolves through `resolveTaskWorkflowIrSync`, which answers with the - DEFAULT board under PostgreSQL, so it evaluates to `in-progress` — the same value as the literal - it replaced. A card advancing into a renamed execution lane still matches nothing, still reads as - an evacuation, and still kills a healthy planning session. Identical behaviour, on every board. - - What the change DID do was add an eighth entry to `check-inert-sync-lane-conversions` for this - file (7 -> 8) and leave `main` red on that gate. Its failure text is explicit that the fix is not - to re-record: "Do NOT re-record the baseline to clear this — that is the same false green one - layer up." So the arm goes back to the literal, which is honest about being one and keeps this - file's census entry pointing at work that is still outstanding. - - DELIBERATE-LITERAL — THE SPECIFICATION IS ABOVE. Whoever supplies a lane answer that is not sync-resolved should make - this line read `disposeLanes.wip` and delete this note. LEFT COUNTED until then. - */ - if (task.column === disposeLanes.hold || task.column === disposeLanes.intake || task.column === "in-progress") return; + const disposeLanes = meta?.lanes; + if (disposeLanes + ? task.column === disposeLanes.hold || task.column === disposeLanes.intake || task.column === disposeLanes.wip + : LEGACY_PLANNER_COLUMNS.has(task.column)) return; if (this.activeSubagentSessions.has(task.id)) { this.disposeSubagentsForTask(task.id, `task moved to ${task.column}`); } diff --git a/scripts/lib/inert-sync-lane-baseline.json b/scripts/lib/inert-sync-lane-baseline.json index de7ab58438..e755b05c52 100644 --- a/scripts/lib/inert-sync-lane-baseline.json +++ b/scripts/lib/inert-sync-lane-baseline.json @@ -1,7 +1,6 @@ { - "total": 6, + "total": 2, "byFile": { - "packages/engine/src/executor.ts": 2, - "packages/engine/src/triage.ts": 4 + "packages/engine/src/executor.ts": 2 } }