root/fusion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(engine): serialise wedge handling per task, then convert the lane guards it was blocking (5 → 1) (#3087)

Browse Source 
The largest unclaimed census cluster, and the one two earlier fleet
passes explicitly declined.

## The standing blocker, taken on

Both passes converted these four ids and reverted, each time after the
same test went red:

```
task-wedge-notification.test.ts > sends one actionable push and mailbox message per active terminal episode
  expected 2 calls, got 1
```

Their diagnosis was right and I have kept it: this branch **resolves** a
wedge episode, `handleTaskUpdated` starts it fire-and-forget from a
synchronous `(task) => void` listener, and **any** await introduced
before the resolve lets a re-wedge arriving close behind reach `claim`
while the previous episode is still active — `claimed: false`, second
operator notification silently dropped. Column resolution needs an
await, so the conversion could not be made safe from inside the branch.

Both notes named the fix and left it for "whoever owns the wedge episode
contract": *serialise wedge handling per task*. This PR does that, then
takes the conversion.

## 1. Serialisation

`enqueueWedgeHandling` chains handling per task id, so
resolve-then-claim keeps its order however many awaits either branch
acquires. Details that matter:

- **Keyed by task, not global** — different tasks stay concurrent, so
this is not a throughput regression on a busy board.
- **The map entry is dropped when its chain drains**, and only if no
later link was appended while it ran, so it does not grow with the task
table.
- **Links never reject.** `maybeNotifyTaskWedge` already owns its error
handling; a rejected link would poison every later notification for that
task.

## 2. The conversion it was blocking

The four ids are an enumeration of *"every lane except review"* — the
lanes whose occupancy proves a wedged card's lifecycle has visibly
resumed. On a renamed board none of them matched, so a recovered card's
episode never resolved. Two consequences, and the second is worse than
the first:

1. the operator keeps an open "needs operator action" alert for work
that has moved on;
2. an active episode **suppresses re-claim**, so the *next* genuine
wedge on that task is never delivered.

Membership over the four roles, legacy-seeded, so an unconverted board
resolves exactly the four ids it used to compare.

## Measured

**The acceptance test the earlier notes named is the gate on both
halves.** With the conversion and *without* the serialisation, "sends
one actionable push and mailbox message per active terminal episode"
fails exactly as they reported. With the serialisation, green. I
reproduced their finding rather than taking it on trust — it is the
evidence that the serialisation is load-bearing and not incidental
refactoring.

| | result |
|---|---|
| `task-wedge-notification.test.ts` | **15/15** (2 new) |
| notification suites | **11 files / 234 tests pass** |
| `tsc --noEmit -p packages/engine` | clean |
| census `--strict`, `check-lane-wiring`,
`check-inert-sync-lane-conversions`, `check-fnxc-future-dates` | clean |

**MUTATION**: restoring the four literals fails the renamed-recovery
case and leaves its paired negative green.

**A vacuity I caught and fixed, worth stating plainly.** My first
version of the renamed case recovered the card with `status: "queued"`.
`hasProgressed` is an OR whose other arm is *"status is a non-failed
string"* — so that arm answered true and the column comparison never
ran. The mutation did not fail it. The case now clears `status` and
`error` together, which makes column membership the only thing that can
resolve the episode, and the paired negative uses the identical shape so
only the lane differs.

## Census

| | before | after |
|---|---|---|
| `notification-service.ts` | 5 | **1** |
| repo backlog | 71 | **67** |

## The remaining 1, flagged not guessed

`isManualMergeHold` (`task.column !== "in-review"`) is sync, and so is
its only caller `classifyWorkflowTransitionNotification`, reached from
the same `handleTaskUpdated` listener. Converting it means making that
whole chain async — a change to notification *classification ordering*
against every other `task:updated` handler, which is a different
contract from the episode one this PR owns. The serialisation added here
does not cover it: it wraps wedge handling, not transition
classification. Threading a pre-resolved `LifecycleColumns` in as a
parameter is the likely fix, and it wants the same gate-placement
judgement applied deliberately rather than swept in behind this.

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
gsxdsm
2026-07-31 04:10:03 -07:00
committed by GitHub
parent 701677a2e5
commit a7b2a757fa
3 changed files with 147 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

78
packages/engine/src/notification/__tests__/task-wedge-notification.test.ts
View File

@@ -4,7 +4,24 @@ import { NotificationService } from "../notification-service.js";
import { describeSelfHealingNoActionWedge, describeTaskWedge } from "../task-wedge-notification.js";
type Listener = (task: Task) => void;
function fixture() {
/*
FNXC:WorkflowResolvedColumns 2026-07-31-21:35:
A board whose lanes carry no legacy id: hold `drafting`, wip `building`, review `checking`,
complete `shipped`. Supplied through `listWorkflowDefinitions`, the only store read
`resolveProjectColumnsForRoles` makes and one that is answerable under PostgreSQL.
*/
const RENAMED_IR = {
version: "v2", id: "wf-renamed", name: "renamed", nodes: [], edges: [],
columns: [
{ id: "drafting", name: "Drafting", traits: [{ trait: "hold", config: { release: "capacity" } }] },
{ id: "building", name: "Building", traits: [{ trait: "wip", config: { limitSetting: "maxConcurrent" } }] },
{ id: "checking", name: "Checking", traits: [{ trait: "merge" }] },
{ id: "shipped", name: "Shipped", traits: [{ trait: "complete" }] },
],
};
function fixture(workflowIr?: unknown) {
const listeners = new Set<Listener>();
let wedge: Task["wedgeNotification"];
const store = {
@@ -21,6 +38,8 @@ function fixture() {
wedge = { reasonKey, episodeId: `${taskId}-${reasonKey}-${Date.now()}`, status: "active", transitionedAt: new Date().toISOString() };
return { claimed: true, episodeId: wedge.episodeId };
},
/* Absent → the helper keeps the legacy ids, which is every pre-existing case in this file. */
...(workflowIr ? { listWorkflowDefinitions: async () => [{ ir: workflowIr }] } : {}),
};
const sendMessageOnce = vi.fn(async (_input: unknown, _key: string) => ({ message: {} as any, inserted: true }));
const service = new NotificationService(store as any, { messageStore: { on: () => undefined, sendMessageOnce } as any, failedNotificationGraceMs: 60_000 });
@@ -131,4 +150,61 @@ describe("task wedge notifications", () => {
expect(sendMessageOnce.mock.calls.map((call) => call[1])).not.toContain(expect.stringContaining("details"));
await service.stop();
});
/*
FNXC:WorkflowResolvedColumns 2026-07-31-21:35:
A RECOVERED CARD ON A RENAMED BOARD MUST CLOSE ITS WEDGE EPISODE.
The resolve branch tested "has this card moved on" by comparing against `todo`/`in-progress`/
`done`/`archived`. On a board using none of those ids nothing matched, so the episode stayed
`active` after the card visibly recovered — the operator kept an open "needs operator action"
alert for work that had moved on, and (because an active episode suppresses re-claim) the NEXT
genuine wedge on that task was never delivered either.
The observable is the second delivery, not the episode record: an episode that resolves but
delivers nothing new would be a silent regression of the same alert.
*/
it("resolves an episode when the card recovers into a RENAMED lane, and re-delivers on re-wedge", async () => {
const { store, service, sendMessageOnce, task } = fixture(RENAMED_IR);
await service.start();
const wedged = (updatedAt: string) => task({ column: "checking" as never, updatedAt });
store.emit(wedged("2026-07-22T12:00:00.000Z"));
await vi.waitFor(() => expect(sendMessageOnce).toHaveBeenCalledTimes(1));
/*
THE RECOVERY CARRIES NO STATUS, and that is what makes this test about the column at all.
`hasProgressed` is an OR whose other arm is "status is a non-failed string" — my first version
recovered with `status: "queued"`, that arm answered true, and the case passed against the
literals. Measured: the mutation did not fail it. With `status` and `error` both cleared, the
column membership is the ONLY thing that can resolve this episode.
*/
store.emit(task({ status: undefined, error: undefined, column: "building" as never, updatedAt: "2026-07-22T12:02:00.000Z" }));
store.emit(wedged("2026-07-22T12:03:00.000Z"));
await vi.waitFor(() => expect(sendMessageOnce).toHaveBeenCalledTimes(2));
await service.stop();
});
/*
The paired negative. The conversion widens membership over four roles, so it must not treat the
REVIEW lane as progress — a wedged card sitting in review has not moved on, and resolving there
would clear every episode on the next incidental update and re-alert forever.
*/
it("does NOT resolve on a status-less update while the card sits in the renamed REVIEW lane", async () => {
const { store, service, sendMessageOnce, task } = fixture(RENAMED_IR);
await service.start();
store.emit(task({ column: "checking" as never, updatedAt: "2026-07-22T12:00:00.000Z" }));
await vi.waitFor(() => expect(sendMessageOnce).toHaveBeenCalledTimes(1));
/* Same shape as the positive — no status, no error — so only the lane differs. Review is not
progress: resolving here would clear the episode on any incidental update and re-alert. */
store.emit(task({ status: undefined, error: undefined, column: "checking" as never, updatedAt: "2026-07-22T12:01:00.000Z" }));
store.emit(task({ column: "checking" as never, updatedAt: "2026-07-22T12:02:00.000Z" }));
await new Promise((resolve) => setTimeout(resolve, 20));
expect(sendMessageOnce).toHaveBeenCalledTimes(1);
await service.stop();
});
});

104
packages/engine/src/notification/notification-service.ts
View File

@@ -11,7 +11,7 @@ import type {
Task,
} from "@fusion/core";
import type { LifecycleColumns, WorkflowIrResolverStore } from "@fusion/core";
import { DASHBOARD_USER_ID, NotificationDispatcher, resolveReviewColumns, resolveTaskLifecycleColumns, resolveWorkflowIrForTask } 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";
import { classifyTransientMergeError } from "../transient-merge-error-classifier.js";
@@ -60,6 +60,14 @@ interface NotificationServiceStore {
getTaskWorkflowSelection?: WorkflowIrResolverStore["getTaskWorkflowSelection"];
getTaskWorkflowSelectionAsync?: WorkflowIrResolverStore["getTaskWorkflowSelectionAsync"];
getWorkflowDefinition?: WorkflowIrResolverStore["getWorkflowDefinition"];
/*
FNXC:WorkflowResolvedColumns 2026-07-31-21:20:
The PROJECT-level read `resolveProjectColumnsForRoles` needs. Optional for the same reason as the
three above — an absent method degrades to the legacy ids rather than forcing a workflow surface
into every notification fake — and, unlike the sync selection readers, this one is answerable under
PostgreSQL, which is what makes the lane answers here real rather than decorative.
*/
listWorkflowDefinitions?: () => Promise<ReadonlyArray<{ ir?: unknown }>>;
on<K extends keyof NotificationServiceStoreEvents>(
event: K,
listener: (...args: NotificationServiceStoreEvents[K]) => void,
@@ -118,6 +126,41 @@ export class NotificationService {
private failureNotificationMode: "sticky-only" | "all" | "terminal-only" = "sticky-only";
/** Compatibility fallback for lightweight test stores without the durable TaskStore CAS. */
private readonly activeWedgeReasons = new Map<string, string>();
/*
FNXC:TaskWedgeNotifications 2026-07-31-21:10:
PER-TASK SERIALISATION OF WEDGE HANDLING — the blocker two earlier fleet passes recorded and
declined to take on.
`handleTaskUpdated` is a synchronous `(task) => void` listener that starts `maybeNotifyTaskWedge`
fire-and-forget, and one branch of that method RESOLVES an episode while another CLAIMS one. With
no ordering between them, any await added before the resolve lets a re-wedge arriving close behind
reach `claim` while the previous episode is still `active`; the claim returns `claimed: false` and
the second operator notification is silently dropped. That is measured, not theoretical: it is what
`task-wedge-notification.test.ts > "sends one actionable push and mailbox message per active
terminal episode"` catches, and it is why the column conversion in this file was reverted twice.
A per-task promise chain fixes the ordering itself rather than the symptom. Handling for one task
runs to completion before the next handling for that task begins, so resolve-then-claim keeps its
order no matter how many awaits either branch acquires. Different tasks stay concurrent — the chain
is keyed by task id, not global.
The entry is deleted when the chain drains, so this map does not grow with the task table. Chain
links never reject: `maybeNotifyTaskWedge` already owns its own error handling, and a rejected link
would poison every later notification for that task.
*/
private readonly wedgeHandlingChains = new Map<string, Promise<void>>();
/** Queues wedge handling for one task behind any handling already in flight for it. */
private enqueueWedgeHandling(taskId: string, run: () => Promise<void>): Promise<void> {
const previous = this.wedgeHandlingChains.get(taskId) ?? Promise.resolve();
const next = previous.then(run, run);
this.wedgeHandlingChains.set(taskId, next);
/* Drop the entry only if no later link was appended while this one ran. */
void next.finally(() => {
if (this.wedgeHandlingChains.get(taskId) === next) this.wedgeHandlingChains.delete(taskId);
});
return next;
}
constructor(
private readonly store: NotificationServiceStore,
@@ -337,7 +380,7 @@ export class NotificationService {
only operator notification; dispatch-time suppression below covers races.
*/
if (wedge) this.cancelPendingFailureNotification(task.id, "classified-terminal-wedge");
if (!transientFailure) void this.maybeNotifyTaskWedge(task, wedge);
if (!transientFailure) void this.enqueueWedgeHandling(task.id, () => this.maybeNotifyTaskWedge(task, wedge));
void this.maybeSuppressTransientFailedNotification(task, `status=${task.status ?? "undefined"}`);
/*
@@ -489,7 +532,7 @@ export class NotificationService {
*/
/** Delivers a self-healing no-action escalation through the durable wedge episode seam. */
async notifyTaskWedge(task: Task, descriptor: TaskWedgeDescriptor): Promise<void> {
await this.maybeNotifyTaskWedge(task, descriptor);
await this.enqueueWedgeHandling(task.id, () => this.maybeNotifyTaskWedge(task, descriptor));
}
private async maybeNotifyTaskWedge(task: Task, suppliedDescriptor?: TaskWedgeDescriptor | null): Promise<void> {
@@ -509,43 +552,32 @@ export class NotificationService {
// stays in review, so arbitrary in-review/status writes are not resolution
// evidence. Only an active owner state or real lifecycle advance can close it.
/*
FNXC:WorkflowResolvedColumns 2026-07-30-23:35 (fleet phase — REVERTED AFTER MEASUREMENT, flagged and left counted):
These four ids are an enumeration of "every lane except review". I converted them to the four ROLES
they name and it PASSED typecheck and the notification suites, then failed
`task-wedge-notification.test.ts > sends one actionable push and mailbox message per active terminal
episode` — green on main, red with the conversion, `expected 2 calls, got 1`.
FNXC:WorkflowResolvedColumns 2026-07-31-21:20 (the blocker is gone, so the conversion lands):
These four ids are an enumeration of "every lane except review" — the lanes whose occupancy
proves a wedged card's lifecycle has visibly resumed. On a renamed board none of them matched,
so a recovered card's episode was never resolved and the operator kept an open wedge alert for
work that had moved on.
The cause is not the test. `handleTaskUpdated` is a synchronous `(task) => void` listener that starts
this work fire-and-forget, and THIS is the branch that RESOLVES a wedge episode. Adding an await
before the resolve means a re-wedge arriving close behind still sees the previous episode `active`,
its claim returns `claimed: false`, and the second operator notification is DROPPED. The awaits added
elsewhere in this file are downstream of an existing await or inside a timer callback; this one sits
on the only path that closes an episode, so it changes delivery rather than just timing.
TWO EARLIER PASSES CONVERTED THIS AND REVERTED IT, both times after
`task-wedge-notification.test.ts > "sends one actionable push and mailbox message per active
terminal episode"` went red with `expected 2 calls, got 1`. Their diagnosis was right and worth
restating: this branch RESOLVES an episode, `handleTaskUpdated` starts it fire-and-forget from a
synchronous listener, and ANY await introduced before the resolve lets a re-wedge arriving close
behind reach `claim` while the previous episode is still active — `claimed: false`, second
notification dropped. Column resolution needs an await, so the conversion could not be made
safe from inside this branch.
Fixing it properly means serialising wedge handling per task (a queue or a per-task lock) so
resolution cannot interleave with the next claim. That is a delivery-semantics change to operator
notifications, not a column conversion, so it is out of fleet scope and left for whoever owns the
wedge episode contract.
It is safe now because `enqueueWedgeHandling` serialises wedge handling PER TASK, so
resolve-then-claim keeps its order however many awaits either branch acquires. That is the
wedge-episode-contract change the earlier notes said this was waiting on; it lands in the same
commit, and the named acceptance test is the gate on both halves.
Left COUNTED with no exemption marker — four of this file's five remaining entries are here, and the
census should keep saying so.
FNXC:WorkflowResolvedColumns 2026-07-31-02:40 (ATTEMPTED, MEASURED, REVERTED — do not retry as written):
I converted these four ids to a resolved `progressedLanes` set and it broke an existing gate test
(`task-wedge-notification.test.ts` -> "sends one actionable push and mailbox message per active
terminal episode": 1 message delivered, 2 expected).
The cause is the paragraph directly above, and it is stronger than it reads: the hazard is not
specific to the resolve/claim ordering, it is ANY await added before the resolve. Column resolution
needs one, so a resolved answer here costs a dropped operator notification whenever a re-wedge
arrives close behind a recovery. The `task:updated` listeners fire synchronously, so the second
emit reaches `claim` while the first episode is still open.
This is therefore blocked on serialising wedge handling per task, NOT on the conversion being hard.
Convert these four only in a change that already owns the wedge-episode contract, and re-run that
test as the acceptance check — it fails loudly, which is why this is recorded rather than exempted.
MEMBERSHIP over the four roles, not first-match: "has this card moved on" can be true of more
than one lane per role on a renamed board, and a first-match answer would silently ignore the
others. Legacy-seeded, so an unconverted board resolves exactly the four ids it used to compare.
*/
const hasProgressed = task.column === "todo" || task.column === "in-progress" || task.column === "done" || task.column === "archived"
const progressedLanes = await resolveProjectColumnsForRoles(this.store, ["hold", "countsTowardWip", "complete", "archived"]);
const hasProgressed = progressedLanes.has(task.column)
|| (!isActiveSelfHealingNoAction && typeof task.status === "string" && task.status !== "failed")
|| (isActiveSelfHealingNoAction && ["queued", "planning", "in-progress", "merging", "merging-pr", "merged", "done"].includes(task.status ?? ""));
if (hasProgressed) {

4
scripts/lib/lifecycle-column-census-baseline.json
View File

@@ -1,8 +1,7 @@
{
"generatedFrom": "node scripts/lifecycle-column-census.mjs --strict --update-baseline",
"byFile": {
"packages/engine/src/self-healing.ts": 26,
"packages/engine/src/notification/notification-service.ts": 5,
"packages/engine/src/self-healing.ts": 22,
"packages/engine/src/executor.ts": 4,
"packages/core/src/task-store/project-store-ops.ts": 2,
"packages/dashboard/app/utils/taskRevert.ts": 2,
@@ -19,6 +18,7 @@
"packages/core/src/task-store/task-id-integrity.ts": 1,
"packages/dashboard/app/components/ResearchTaskActionModal.tsx": 1,
"packages/dashboard/app/components/TaskCard.tsx": 1,
"packages/engine/src/notification/notification-service.ts": 1,
"packages/engine/src/triage.ts": 1
},
"deliberateByFile": {