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

Phase A: workflow-owned lifecycle foundation (U1, U2, U3) (#2467)

Browse Source 
Phase A (Foundation) of
`docs/plans/2026-07-26-001-refactor-workflow-owned-lifecycle-plan.md`.
Three units, one commit each. No operator-visible behavior change.

## U1 — Lifecycle-column resolution seam

`resolveLifecycleColumns(ir)` returns `{ intake, hold, wip, review,
complete, archived }` — the first column carrying each trait,
`undefined` for a role no column carries.
`resolveTaskLifecycleColumns(store, taskId, cache?)` is the store-aware
form; the cache is caller-owned so a sweep reads one IR per workflow
rather than one per card.

A v1/column-less IR resolves to `undefined` for the **whole struct**
rather than a struct of undefined roles. A caller must be able to
distinguish "this workflow declares no hold column" (a real shape to
honor) from "no column vocabulary at all" (skip and log) — only the
second licenses conservative fallback.

Nothing consumes the seam yet; Phases B–D convert the ~207 hardcoded
column literals onto it.

## U2 — Delete the pre-cutover parity machinery (delete-only)

**`workflow-columns-settings.ts`** — `isWorkflowColumnsEnabled` had the
body `return true`. Six live call sites branched on it, so every
flag-OFF arm was dead code that read as a supported configuration.
Deleted; surviving side inlined at self-healing's transitionPending
sweep, the scheduler's per-column capacity diagnostic, merge-trait's
policy resolver, the board-workflows payload, two task-workflow routes,
and the CLI TUI's column enrichment.

**`workflow-parity.ts`** — asserted the default workflow's adjacency
*equals* the legacy `VALID_TRANSITIONS`. U11 deliberately breaks that
equality by merging Todo into Planning, so this is not a stale assertion
to update; it is a contract against the target state. Its emitter
(`workflow-parity-observer.ts`) is already a tombstone, so
`getWorkflowParitySummary` and `computeWorkflowColumnsGraduationReport`
aggregated run-audit rows nothing writes and had no caller outside
`TaskStore`. Both store methods go with it.

`flagEnabled` stays on the board-workflows **wire** as a constant `true`
— shipped dashboard clients still branch on it, and changing the
response shape is not a deletion. U10 retires the field once no client
reads it.

The `legacy-tombstones` ratchet is extended to both files plus seven
symbols, each with the reason it is gone.

### ⚠️ Finding: the third listed deletion was NOT dead

The plan also lists "the flag-off inline move path" in
`task-store/moves.ts`. It is **not** deleted, per U2's execution note
("any behavior change found while removing a branch means the branch was
not dead").

That path is gated on `isWorkflowColumnsCompatibilityFlagEnabled`
(`store.ts:38`) — a **different** function from the always-true public
helper. It reads the raw `experimentalFeatures.workflowColumns` setting,
which nothing in production sets (`settings-schema.ts:396` — "no default
flags are emitted"; zero non-test writers; the operator's own
`~/.fusion/settings.json` has no such key). So `useWorkflow` is false
for effectively every real project: the flag-OFF inline side effects are
the **live** default move path and the flag-ON `default-workflow-hooks`
path is the dead one. The code says so itself at `moves.ts:638`.

Deleting that branch would swap every project onto an untravelled code
path — a behavior change, not a deletion.

**Carry this into Phases B and C, stated plainly so the plan's error is
not repeated:**

> **The inline move path in `moves.ts` is LIVE.
`default-workflow-hooks.ts` (the trait-hook path) is DEAD.** KTD-6
asserted the inverse. Until the convergence unit lands, **nothing may
assume trait hooks run** — a guard, sweep, or subscriber written against
`applyDefaultWorkflowMoveEffects` would never fire in production and
would still pass its tests.

Convergence is **not** attempted here. It is its own unit (Phase A2)
with a proper equivalence proof, per operator decision.

### U3's emit point is on the LIVE path — the seam is not born dead

Worth stating explicitly because it is the failure mode that would make
every later subscriber silently never fire: the `TaskTransitioned` emit
is **not** inside the `if (useWorkflow)` branch. That block closes at
`moves.ts:1212`; the emit sits at `:1214`, beside the existing
`store.emit("task:moved", …)`, on the unconditional post-commit path. It
therefore fires on **both** the live inline path and the dead hooks
path, and the convergence unit inherits the obligation to keep it firing
on whichever path survives — same events, same order, same payloads.

The graph-side emitters (`NodeEntered`, `RunSuspended`) carry the same
risk from a different direction: the bus refuses an invalid payload
*silently* by design, so an emitter regression would stop the event with
no test failure. They are asserted end-to-end through the real bus —
"did a subscriber actually receive it", not "was emit called" — because
a spy passes on a refused payload. The `moveTaskInternalImpl` emit does
**not** yet have that end-to-end assertion against a real store move;
that proof belongs to the convergence unit, which has to build the
both-paths fixture anyway.

## U3 — Post-commit event seam with a transactional outbox

**The bus is not a queue, not a transaction participant, and not a
delivery guarantee.** Durable follow-on work uses the transactional
outbox — a `workflow_work_items` row written *inside* the transition
transaction (the shape `createCompletionHandoffWorkflowWork` already
uses). "Emit after commit, let a subscriber enqueue the work" has a
crash window where a process dies between commit and subscriber, leaving
no event *and* no work-item row, so required work is skipped permanently
with nothing to recover from. Post-commit subscribers therefore carry
only losable reactions.

Emission is consequently lossy and isolated by design: a throwing or
rejecting subscriber is caught and logged, cannot roll back the
transition, and cannot stop the others. Deliveries append to one serial
chain, so two transitions on a task deliver in commit order.

The ids/outcomes-only rule is **mechanised, not documented** —
run-audit's equivalent lives only in prose and has been violated
repeatedly. A payload carrying an object body or a prose string is
refused at the emit boundary and never reaches a subscriber or log sink.
It degrades rather than throws: the emitter is post-commit, so a shape
bug must not become a lifecycle failure.

Emit points: `TaskTransitioned` from the single post-commit point in
`moveTaskInternalImpl`; `NodeEntered` and `RunSuspended` from the graph
column boundary, the latter *after* the durable continuation is
persisted so an observed suspension implies a resumable run.

`registerWorkflowEventSubscribers` (engine) is empty on purpose —
U7/U8/U10 move real reactions onto it, each with the characterization
test proving the reaction was non-authoritative first.

## Verification

- `pnpm test:gate` — green (2/10, 16/299, 1/71).
- `pnpm lint`, `pnpm build`, `tsc --noEmit` on core and engine — green.
- U1: 20 tests in `workflow-lifecycle-traits.test.ts`, including the
fully-renamed-workflow case (fails if the resolver falls back to a
literal) and a shared-cache read-count assertion.
- U2: `legacy-tombstones.test.ts` green with the extended ratchet;
`board-workflows`, `merge-trait`, `workflow-graph-executor-parity`, and
move-hook suites green with no expectation edits.
- U3: 20 bus-invariant unit tests (isolation, ordering, the allowed-key
and required-key halves of the ids-only rule, lossiness) plus 3
end-to-end emitter-delivery tests; 5 outbox tests against a **real
PostgreSQL** work-item table (crash survival, rollback, at-least-once
redelivery on lease expiry, idempotent handler → one effect,
dropped-subscriber vs. durable work). A hand-written fake of the lease
predicate would only prove the fake redelivers.

**Not verified:** the `moveTaskInternalImpl` emit is confirmed on the
unconditional post-commit path by structure and by the surrounding
tests, but is *not* yet asserted end-to-end against a real store move on
both flag settings — that is Phase A2's fixture. The engine subscriber
registry ships empty by design, so no production subscriber exercises
the bus end-to-end yet. `settings-defaults.test.ts` has one pre-existing
failure on `main` (a logger-prefix mismatch in the
`mergeIntegrationWorktree=cwd-main` warning) — confirmed present on a
clean tree, unrelated to this branch.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

* **New Features**
* Workflow lifecycle columns are now derived from workflow definitions,
supporting renamed and custom workflows.
* Added post-commit lifecycle events for task transitions, node entry,
and run suspend/resume with validated payloads.
* Follow-on processing for lifecycle emissions is now more robust
(rollback-safe, at-least-once delivery, idempotent handling).
* **Bug Fixes**
* Workflow board responses, task enrichment, and promotion no longer
depend on workflow-columns feature-flag gating.
  * Subscriber failures no longer impact committed workflow transitions.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->
This commit is contained in:
gsxdsm
2026-07-27 13:30:13 -07:00
committed by GitHub
parent d873ee8160
commit 4158cf1ab7
29 changed files with 1886 additions and 961 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
.changeset/workflow-owned-lifecycle-phase-a.md Normal file
View File

@@ -0,0 +1,7 @@
---
"@runfusion/fusion": patch
---
summary: Internal groundwork for workflow-owned lifecycle; no operator-visible behavior change.
category: internal
dev: Phase A of the workflow-owned-lifecycle program. U1 adds `resolveLifecycleColumns(ir)` / `resolveTaskLifecycleColumns(store, taskId, cache?)` in `@fusion/core` — the trait-driven seam later phases convert ~207 hardcoded column literals onto. U2 deletes `workflow-columns-settings.ts` (`isWorkflowColumnsEnabled` returned a literal `true`; its six flag-OFF branches were unreachable) and `workflow-parity.ts` with its two dead TaskStore methods (`getWorkflowParitySummary`, `computeWorkflowColumnsGraduationReport`); both files plus seven symbols are added to the `legacy-tombstones` ratchet. The board-workflows response keeps `flagEnabled: true` on the wire for shipped clients. U3 adds the post-commit lifecycle event bus (`getWorkflowEventBus`, `emitWorkflowLifecycleEvent`, ids/outcomes-only payloads enforced at the emit boundary) emitting `TaskTransitioned` from the single post-commit point in `moveTaskInternalImpl` and `NodeEntered`/`RunSuspended` from the graph column boundary, plus `registerWorkflowEventSubscribers` in `@fusion/engine` (empty by design). Durable follow-on work stays in the transactional outbox — a `workflow_work_items` row written inside the transition transaction — with at-least-once delivery proven against real PostgreSQL; subscribers carry only losable reactions.

35
packages/cli/src/commands/dashboard.ts
View File

@@ -18,7 +18,6 @@ import {
GlobalSettingsStore,
resolveGlobalDir,
DEFAULT_AGENT_HEARTBEAT_INTERVAL_MS,
isWorkflowColumnsEnabled,
isWorkspaceTask,
resolveColumnFlags,
BUILTIN_CODING_WORKFLOW_IR,
@@ -1068,20 +1067,24 @@ export async function runDashboard(port: number, opts: { paused?: boolean; dev?:
//
// The CLI TUI degrades gracefully (R18): cards in workflow columns it can't
// express must map by trait flags into its buckets or a read-only "other"
// bucket, never silently disappear. The TUI is flag-blind, so when
// `workflowColumns` is ON we enrich each slim task with its resolved column's
// display name + merged trait flags. Self-contained: derives everything from
// already-exposed store methods (workflow selection + definition) + the core
// `resolveColumnFlags` export — no dependency on concurrent U9 server work.
// Flag-OFF: returns undefineds and the TUI renders exactly as before.
// bucket, never silently disappear. So we enrich each slim task with its
// resolved column's display name + merged trait flags. Self-contained: derives
// everything from already-exposed store methods (workflow selection +
// definition) + the core `resolveColumnFlags` export. An unresolvable workflow
// returns undefineds and the TUI falls back to legacy column-id bucketing.
type ResolvedColumnInfo = { columnName?: string; columnFlags?: TraitFlags };
async function resolveTaskColumnInfo(
projectStore: TaskStore,
flagOn: boolean,
workflowIrCache: Map<string | undefined, WorkflowIrColumn[] | null>,
task: { id: string; column: string },
): Promise<ResolvedColumnInfo> {
if (!flagOn) return {};
/*
FNXC:WorkflowColumns 2026-07-27-09:56 (U2 / R9):
The `flagOn` parameter and its `if (!flagOn) return {}` early exit are gone.
Its only caller derived it from `isWorkflowColumnsEnabled`, a literal `true`,
so column enrichment was already unconditional — as was the settings read that
fed it, now also removed.
*/
try {
/*
FNXC:WorkflowSelection 2026-07-14-17:06:
@@ -2952,17 +2955,15 @@ export async function runDashboard(port: number, opts: { paused?: boolean; dev?:
listTasks: async (projectPath: string) => {
const projectStore = await getProjectStore(projectPath);
const tasks = await projectStore.listTasks({ slim: true, includeArchived: false });
// U11 (R18): when the workflow-columns flag is ON, enrich each task
// with its resolved column display name + trait flags so the
// flag-blind TUI can map non-legacy columns into its buckets (or the
// read-only "other" bucket) instead of silently dropping them. The
// IR cache keeps this O(workflows) rather than O(tasks) DB reads.
const settings = await projectStore.getSettings();
const flagOn = isWorkflowColumnsEnabled(settings);
// U11 (R18): enrich each task with its resolved column display name
// + trait flags so the column-blind TUI can map non-legacy columns
// into its buckets (or the read-only "other" bucket) instead of
// silently dropping them. The IR cache keeps this O(workflows)
// rather than O(tasks) DB reads.
const workflowIrCache = new Map<string | undefined, WorkflowIrColumn[] | null>();
return Promise.all(
tasks.map(async (t) => {
const info = await resolveTaskColumnInfo(projectStore, flagOn, workflowIrCache, t);
const info = await resolveTaskColumnInfo(projectStore, workflowIrCache, t);
return {
id: t.id,
title: t.title,

331
packages/core/src/__tests__/postgres/workflow-events-outbox.pg.test.ts Normal file
View File

@@ -0,0 +1,331 @@
/*
FNXC:WorkflowEvents 2026-07-27-13:10 (U3 / R5 — workflow-owned lifecycle):
The TRANSACTIONAL OUTBOX half of the post-commit event seam, proven against a
REAL PostgreSQL `workflow_work_items` table.
Why real PG and not a fake: both properties under test are properties of the
STORE, not of any code this repo would mock. A hand-written fake of the lease
predicate would only prove the fake redelivers, which is exactly the
green-suite-proves-nothing failure the plan warns about.
The two claims:
1. CRASH SURVIVAL. Durable follow-on work must be written INSIDE the transition
transaction, not by a post-commit subscriber. The alternative — "emit after
commit, let a subscriber enqueue the work" — has a window where a process
dies between the commit and the subscriber, leaving no event AND no
work-item row: the work is skipped permanently with nothing to recover
from. Simulated here by committing the transaction and then never running
any post-commit code at all, which is strictly worse than a real crash.
2. AT-LEAST-ONCE DELIVERY. `listDueWorkflowWorkItems` returns items whose lease
has EXPIRED (`leaseExpiresAt IS NULL OR <= now`), so a worker that performs
a side effect and dies before recording completion will have its item
re-claimed and re-run. Calling this substrate "at-most-once" would invite
subscribers to skip the deduplication the delivery semantics actually
require — so the redelivery is asserted, and an idempotent handler is shown
producing ONE effect across TWO deliveries.
Skipped when PostgreSQL is unreachable (FUSION_PG_TEST_SKIP=1) so the merge gate
stays green without a running server — the same posture as the sibling PG suites.
*/
import { describe, it, expect, afterEach } from "vitest";
import postgres from "postgres";
import { execSync } from "node:child_process";
import { createAsyncDataLayer, type AsyncDataLayer } from "../../postgres/data-layer.js";
import { createConnectionSetFromUrl } from "../../postgres/connection.js";
import type { ResolvedBackend } from "../../postgres/backend-resolver.js";
import { applySchemaBaseline } from "../../postgres/schema-applier.js";
import { insertTaskRow, readTaskRow } from "../../task-store/async-persistence.js";
import * as schema from "../../postgres/schema/index.js";
import { and, eq } from "drizzle-orm";
import type { DbTransaction } from "../../postgres/data-layer.js";
import {
upsertWorkflowWorkItem,
listDueWorkflowWorkItems,
transitionWorkflowWorkItem,
getWorkflowWorkItem,
} from "../../task-store/async-workflow-workitems.js";
import { createWorkflowEventBus } from "../../workflow-events.js";
import type { WorkflowLifecycleEvent } from "../../types/workflow-events.js";
const PG_TEST_URL_BASE = process.env.FUSION_PG_TEST_URL_BASE ?? "postgresql://localhost:5432";
const PG_AVAILABLE = process.env.FUSION_PG_TEST_SKIP !== "1" && Boolean(PG_TEST_URL_BASE);
const pgDescribe = PG_AVAILABLE ? describe : describe.skip;
const TEST_PROJECT_ID = "proj_test_u3_outbox";
function uniqueDbName(): string {
return `fusion_u3_outbox_${process.pid}_${Math.random().toString(36).slice(2, 8)}`;
}
function adminExec(statement: string): void {
execSync(
`psql "${PG_TEST_URL_BASE}/postgres" -v ON_ERROR_STOP=1 -c "${statement.replace(/"/g, '\\"')}"`,
{ stdio: "pipe", env: process.env },
);
}
interface TestCtx {
dbName: string;
layer: AsyncDataLayer;
adminSql: ReturnType<typeof postgres>;
}
async function setupCtx(): Promise<TestCtx> {
const dbName = uniqueDbName();
try {
adminExec(`DROP DATABASE IF EXISTS "${dbName}"`);
} catch {
// may not exist
}
adminExec(`CREATE DATABASE "${dbName}"`);
const testUrl = `${PG_TEST_URL_BASE}/${dbName}`;
const backend: ResolvedBackend = {
mode: "external",
runtimeUrl: testUrl,
migrationUrl: testUrl,
migrationUrlOverridden: false,
};
const schemaConnections = await createConnectionSetFromUrl(backend, { poolMax: 1, connectTimeoutSeconds: 5 });
await applySchemaBaseline(schemaConnections.migration);
await schemaConnections.close();
// Bind the layer to a project, as production does — an unbound harness runs
// with RLS bypassed and writes rows the bound reader cannot see.
const connections = await createConnectionSetFromUrl(backend, {
poolMax: 5,
connectTimeoutSeconds: 5,
projectId: TEST_PROJECT_ID,
});
const layer = createAsyncDataLayer(connections, { projectId: TEST_PROJECT_ID });
const adminSql = postgres(testUrl, {
max: 2,
prepare: false,
onnotice: () => {},
connection: { "fusion.project_id": TEST_PROJECT_ID },
});
return { dbName, layer, adminSql };
}
async function teardownCtx(ctx: TestCtx | null): Promise<void> {
if (!ctx) return;
try { await ctx.layer.close(); } catch { /* closing best-effort */ }
try { await ctx.adminSql.end({ timeout: 5 }); } catch { /* closing best-effort */ }
try { adminExec(`DROP DATABASE IF EXISTS "${ctx.dbName}"`); } catch { /* dropped best-effort */ }
}
async function seedTask(ctx: TestCtx, id: string, column = "in-progress"): Promise<void> {
await insertTaskRow(
ctx.layer,
{
id,
title: `outbox ${id}`,
description: "",
column,
priority: "medium",
steps: [],
createdAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
columnMovedAt: new Date().toISOString(),
} as unknown as Record<string, unknown>,
TEST_PROJECT_ID,
);
}
/** The column half of a transition, written on the SAME transaction handle as
* the outbox row — the atomicity this suite exists to demonstrate. */
async function moveColumnInTransaction(tx: DbTransaction, taskId: string, column: string): Promise<void> {
await tx
.update(schema.project.tasks)
.set({ column, columnMovedAt: new Date().toISOString() })
.where(and(eq(schema.project.tasks.id, taskId), eq(schema.project.tasks.projectId, TEST_PROJECT_ID)));
}
pgDescribe("transactional outbox — durable follow-on work (U3 / R5)", () => {
let ctx: TestCtx | null = null;
afterEach(async () => {
await teardownCtx(ctx);
ctx = null;
});
it("a work item written INSIDE the transition transaction survives a crash before the event is emitted", async () => {
ctx = await setupCtx();
await seedTask(ctx, "FN-OUT-1");
// The transition transaction: the COLUMN CHANGE and the durable follow-on
// work commit together — the shape `createCompletionHandoffWorkflowWork`
// already uses at the handoff seam in moves.ts.
await ctx.layer.transactionImmediate(async (tx) => {
await moveColumnInTransaction(tx, "FN-OUT-1", "in-review");
await upsertWorkflowWorkItem(
ctx!.layer,
{ runId: "run-1", taskId: "FN-OUT-1", nodeId: "merge-gate", kind: "merge", state: "runnable" },
tx,
);
});
// ...and then the process "dies": no post-commit code runs at all — not the
// emit, not a subscriber, nothing. This is strictly worse than a real crash
// between commit and emit.
//
// Both halves are durable: the transition landed AND the work that must
// follow it is queued and claimable by the next process to come up.
expect((await readTaskRow(ctx.layer, "FN-OUT-1"))?.column).toBe("in-review");
const due = await listDueWorkflowWorkItems(ctx.layer.db, { kinds: ["merge"] });
expect(due.map((item) => item.taskId)).toEqual(["FN-OUT-1"]);
expect(due[0].state).toBe("runnable");
});
it("a ROLLED-BACK transition leaves no column change, no orphan work, and emits NOTHING", async () => {
ctx = await setupCtx();
await seedTask(ctx, "FN-OUT-2");
// The emit point sits AFTER the transaction block in moves.ts, so a throw
// inside the block unwinds past it — this test models that placement.
const bus = createWorkflowEventBus();
const reaction = countingSubscriber();
bus.subscribe(reaction, { name: "notify" });
await expect(
ctx.layer.transactionImmediate(async (tx) => {
await moveColumnInTransaction(tx, "FN-OUT-2", "in-review");
await upsertWorkflowWorkItem(
ctx!.layer,
{ runId: "run-2", taskId: "FN-OUT-2", nodeId: "merge-gate", kind: "merge", state: "runnable" },
tx,
);
// A guard rejects after both writes — the whole transition unwinds.
throw new Error("transition guard rejected");
}).then(() => {
// Unreachable: the emit would only run on a committed transition.
bus.emit({
type: "TaskTransitioned", taskId: "FN-OUT-2", at: new Date().toISOString(),
from: "in-progress", to: "in-review",
} as WorkflowLifecycleEvent);
}),
).rejects.toThrow("transition guard rejected");
await bus.drain();
expect((await readTaskRow(ctx.layer, "FN-OUT-2"))?.column).toBe("in-progress");
expect(await listDueWorkflowWorkItems(ctx.layer.db, { kinds: ["merge"] })).toEqual([]);
expect(reaction.calls).toBe(0);
});
});
pgDescribe("transactional outbox — delivery is AT-LEAST-ONCE (U3 / R5)", () => {
let ctx: TestCtx | null = null;
afterEach(async () => {
await teardownCtx(ctx);
ctx = null;
});
it("an item whose lease EXPIRED mid-flight is re-claimable — the store redelivers", async () => {
ctx = await setupCtx();
await seedTask(ctx, "FN-OUT-3");
const created = await upsertWorkflowWorkItem(ctx.layer, {
runId: "run-3", taskId: "FN-OUT-3", nodeId: "merge-gate", kind: "merge", state: "runnable",
});
// Worker A claims it with a lease, then dies without recording completion.
const future = new Date(Date.now() + 60_000).toISOString();
await transitionWorkflowWorkItem(ctx.layer, created.id, "running", {
leaseOwner: "worker-a",
leaseExpiresAt: future,
});
// While the lease is live the item is NOT due — no double-dispatch.
const whileLeased = await listDueWorkflowWorkItems(ctx.layer.db, { kinds: ["merge"], now: new Date().toISOString() });
expect(whileLeased).toEqual([]);
// Once the lease expires the SAME item comes back — the side effect worker A
// may already have performed is about to happen a second time.
const afterExpiry = await listDueWorkflowWorkItems(ctx.layer.db, {
kinds: ["merge"],
now: new Date(Date.now() + 120_000).toISOString(),
});
expect(afterExpiry.map((item) => item.id)).toEqual([created.id]);
});
it("an IDEMPOTENT handler run twice over one redelivered item produces exactly one effect", async () => {
ctx = await setupCtx();
await seedTask(ctx, "FN-OUT-4");
const created = await upsertWorkflowWorkItem(ctx.layer, {
runId: "run-4", taskId: "FN-OUT-4", nodeId: "merge-gate", kind: "merge", state: "runnable",
});
/*
The handler contract every durable subscriber must meet: keyed on a
payload-identifying invariant, not on "have I been called". Here that key is
(runId, taskId, nodeId) — the same tuple the work-item table is unique on —
so a redelivery recognises the effect it already produced.
*/
const effects = new Set<string>();
const handle = (item: { runId: string; taskId: string; nodeId: string }): void => {
effects.add(`${item.runId}:${item.taskId}:${item.nodeId}`);
};
handle(created);
const redelivered = await getWorkflowWorkItem(ctx.layer.db, created.id);
expect(redelivered).toBeTruthy();
handle(redelivered!);
expect(effects.size).toBe(1);
});
});
pgDescribe("post-commit event vs. outbox — the division of labour (U3 / R5, KTD-3)", () => {
let ctx: TestCtx | null = null;
afterEach(async () => {
await teardownCtx(ctx);
ctx = null;
});
it("dropping every subscriber loses the REACTION but not the durable work", async () => {
ctx = await setupCtx();
await seedTask(ctx, "FN-OUT-5");
const bus = createWorkflowEventBus();
const reaction = countingSubscriber();
const off = bus.subscribe(reaction, { name: "notify" });
await ctx.layer.transactionImmediate(async (tx) => {
await upsertWorkflowWorkItem(
ctx!.layer,
{ runId: "run-5", taskId: "FN-OUT-5", nodeId: "merge-gate", kind: "merge", state: "runnable" },
tx,
);
});
// Every subscriber goes away before the post-commit emit — the crash window.
off();
bus.emit({
type: "TaskTransitioned",
taskId: "FN-OUT-5",
at: new Date().toISOString(),
from: "in-progress",
to: "in-review",
} as WorkflowLifecycleEvent);
await bus.drain();
expect(reaction.calls).toBe(0);
// The unit of work is still there and still claimable. A dropped event costs
// a notification, never a state change or a unit of work.
const due = await listDueWorkflowWorkItems(ctx.layer.db, { kinds: ["merge"] });
expect(due.map((item) => item.taskId)).toEqual(["FN-OUT-5"]);
});
});
/** Minimal call counter — this suite avoids vitest mock objects so the payload
* assertions above stay plain-value comparisons. */
function countingSubscriber(): { (event: WorkflowLifecycleEvent): void; calls: number } {
const fn = ((_event: WorkflowLifecycleEvent) => { fn.calls += 1; }) as {
(event: WorkflowLifecycleEvent): void;
calls: number;
};
fn.calls = 0;
return fn;
}

2
packages/core/src/__tests__/settings-defaults.test.ts
View File

@@ -2,7 +2,6 @@ import { afterEach, describe, expect, it, vi } from "vitest";
import { CONSECUTIVE_TOOL_FAILURE_RETRY_THRESHOLD, DEFAULT_CONSECUTIVE_TOOL_FAILURE_RETRY_BACKOFF_MS, DEFAULT_MAX_CONSECUTIVE_TOOL_FAILURE_RETRIES, DEFAULT_MAX_AUTO_MERGE_RETRIES, resolveConsecutiveToolFailureRetryBackoffMs, resolveConsecutiveToolFailureThreshold, resolveExecutorEscalationTarget, resolveMaxAutoMergeRetries, resolveMaxConsecutiveToolFailureRetries } from "../in-review-stall.js";
import { isExperimentalFeatureEnabled } from "../experimental-features.js";
import { DEFAULT_GLOBAL_SETTINGS, DEFAULT_PROJECT_SETTINGS, GLOBAL_SETTINGS_KEYS, PROJECT_SETTINGS_KEYS, isGlobalOnlySettingsKey } from "../settings-schema.js";
import { isWorkflowColumnsEnabled } from "../workflow-columns-settings.js";
import {
__resetLegacyCwdMainWarningForTests,
normalizeMergeIntegrationWorktreeMode,
@@ -70,7 +69,6 @@ describe("settings defaults invariants", () => {
expect(isExperimentalFeatureEnabled(undefined, "workflowGraphExecutor")).toBe(false);
expect(isExperimentalFeatureEnabled(undefined, "workflowInterpreterDualObserve")).toBe(false);
expect(isExperimentalFeatureEnabled({ experimentalFeatures: { workflowInterpreterDualObserve: true } }, "workflowInterpreterDualObserve")).toBe(false);
expect(isWorkflowColumnsEnabled({ experimentalFeatures: { workflowColumns: false } })).toBe(true);
});
it("defaults maxAutoMergeRetries to the historical project-scoped cap", () => {

298
packages/core/src/__tests__/workflow-events.test.ts Normal file
View File

@@ -0,0 +1,298 @@
/*
FNXC:WorkflowEvents 2026-07-27-12:40 (U3 / R5, R6 — workflow-owned lifecycle):
The invariants everything downstream assumes about the post-commit bus. Written
test-first per the unit's execution note, because these are properties later
units BUILD ON rather than merely benefit from — a subscriber author who assumes
isolation and gets none writes a plugin that can fail a lifecycle transition.
Four properties are load-bearing:
ISOLATION — a throwing subscriber cannot stop the others or reach the caller.
ORDERING — two seams on one task deliver in the order they committed.
IDS-ONLY — a payload carrying prose or an object body is REFUSED at emit, so
it never reaches a plugin subscriber or a log sink.
LOSSINESS — dropping every subscriber changes nothing, which is what makes
"reactions are non-authoritative" checkable rather than aspirational.
The outbox half of R5 (durable work survives a crash between commit and emit, and
its at-least-once redelivery) is proven against a REAL PostgreSQL work-item table
in `workflow-events-outbox.pg.test.ts` — a hand-written fake of the lease
predicate would only prove the fake redelivers.
*/
import { describe, expect, it, vi } from "vitest";
import { createWorkflowEventBus } from "../workflow-events.js";
import {
findWorkflowEventShapeViolations,
isIdsOnlyWorkflowEvent,
MAX_ID_VALUE_LENGTH,
type WorkflowLifecycleEvent,
} from "../types/workflow-events.js";
function transitioned(overrides: Partial<WorkflowLifecycleEvent> = {}): WorkflowLifecycleEvent {
return {
type: "TaskTransitioned",
taskId: "FN-1",
at: "2026-07-27T00:00:00.000Z",
from: "todo",
to: "in-progress",
...overrides,
} as WorkflowLifecycleEvent;
}
describe("workflow event bus — isolation (R5)", () => {
it("a throwing subscriber does not reach the emitter and does not stop the others", async () => {
const bus = createWorkflowEventBus();
const before = vi.fn();
const after = vi.fn();
bus.subscribe(before, { name: "before" });
bus.subscribe(() => { throw new Error("subscriber exploded"); }, { name: "boom" });
bus.subscribe(after, { name: "after" });
// The emitter is on a post-commit path: the transition has already committed,
// so a throw here would be a lifecycle fault caused by a reaction.
expect(() => bus.emit(transitioned())).not.toThrow();
await bus.drain();
expect(before).toHaveBeenCalledTimes(1);
expect(after).toHaveBeenCalledTimes(1);
});
it("isolates a REJECTING async subscriber the same way as a throwing sync one", async () => {
const bus = createWorkflowEventBus();
const after = vi.fn();
bus.subscribe(async () => { throw new Error("async boom"); }, { name: "async-boom" });
bus.subscribe(after, { name: "after" });
bus.emit(transitioned());
await expect(bus.drain()).resolves.toBeUndefined();
expect(after).toHaveBeenCalledTimes(1);
});
it("a subscriber unsubscribing mid-delivery does not corrupt the event in flight", async () => {
const bus = createWorkflowEventBus();
const seen: string[] = [];
let offSecond: (() => void) | undefined;
bus.subscribe(() => { seen.push("first"); offSecond?.(); }, { name: "first" });
offSecond = bus.subscribe(() => { seen.push("second"); }, { name: "second" });
bus.emit(transitioned());
await bus.drain();
// The snapshot taken at delivery still includes `second` for THIS event…
expect(seen).toEqual(["first", "second"]);
seen.length = 0;
bus.emit(transitioned());
await bus.drain();
// …and excludes it for the next one.
expect(seen).toEqual(["first"]);
});
});
describe("workflow event bus — ordering (R5)", () => {
it("delivers two seams on one task in commit order even when subscribers are async", async () => {
const bus = createWorkflowEventBus();
const order: string[] = [];
bus.subscribe(async (event) => {
// A slow first delivery must not let the second overtake it — this is the
// property that lets a subscriber maintain derived state without its own
// sequencing.
if ((event as { to?: string }).to === "in-progress") {
await new Promise((resolve) => setTimeout(resolve, 20));
}
order.push(`${(event as { from?: string }).from}->${(event as { to?: string }).to}`);
}, { name: "recorder" });
bus.emit(transitioned({ from: "todo", to: "in-progress" } as Partial<WorkflowLifecycleEvent>));
bus.emit(transitioned({ from: "in-progress", to: "in-review" } as Partial<WorkflowLifecycleEvent>));
await bus.drain();
expect(order).toEqual(["todo->in-progress", "in-progress->in-review"]);
});
});
describe("workflow event payloads — ids/outcomes only (R5)", () => {
it("accepts the real event shapes", () => {
expect(isIdsOnlyWorkflowEvent(transitioned({ nodeId: "execute", moveSource: "engine" } as Partial<WorkflowLifecycleEvent>))).toBe(true);
expect(isIdsOnlyWorkflowEvent({
type: "RunSuspended", taskId: "FN-2", at: "2026-07-27T00:00:00.000Z",
nodeId: "execute", reason: "capacity", fromColumn: "todo", toColumn: "in-progress",
})).toBe(true);
});
/* The VALUE-shape half. Asserted on DECLARED keys on purpose: an undeclared
key is refused by the allow-list first, which would mask a value-rule
regression behind an `unknown-key` verdict. */
it("rejects an object BODY on a declared key — the spread-a-row mistake", () => {
const violations = findWorkflowEventShapeViolations({
...transitioned(),
to: { id: "in-review", name: "In review" },
});
expect(violations).toEqual([{ path: "to", reason: "object-body" }]);
});
it("rejects PROSE on a declared key — the attach-the-message mistake", () => {
const long = "x".repeat(MAX_ID_VALUE_LENGTH + 1);
expect(findWorkflowEventShapeViolations({ ...transitioned(), moveSource: long }))
.toEqual([{ path: "moveSource", reason: "prose-string" }]);
// A multi-line value is prose regardless of length — stack traces are short lines.
expect(findWorkflowEventShapeViolations({ ...transitioned(), moveSource: "line one\nline two" }))
.toEqual([{ path: "moveSource", reason: "prose-string" }]);
});
/*
FNXC:WorkflowEvents 2026-07-27-15:50 (U3, PR #2467 review):
The half that value-shape checking alone MISSES. `error: "auth failed"`,
`prompt: "summarize"`, and `modelId` are all short single-line scalars — they
pass every value rule and would still reach a plugin subscriber. The per-type
key allow-list is what stops them, so it gets its own cases.
*/
it("rejects an unknown KEY even when its value is a perfectly good scalar", () => {
expect(findWorkflowEventShapeViolations({ ...transitioned(), error: "auth failed" }))
.toEqual([{ path: "error", reason: "unknown-key" }]);
expect(findWorkflowEventShapeViolations({ ...transitioned(), modelId: "claude-opus-5" }))
.toEqual([{ path: "modelId", reason: "unknown-key" }]);
expect(findWorkflowEventShapeViolations({ ...transitioned(), prompt: "summarize" }))
.toEqual([{ path: "prompt", reason: "unknown-key" }]);
});
it("scopes the allow-list PER TYPE — a valid key on one event is unknown on another", () => {
// `outcome` belongs to NodeCompleted, not to TaskTransitioned.
expect(findWorkflowEventShapeViolations({ ...transitioned(), outcome: "success" }))
.toEqual([{ path: "outcome", reason: "unknown-key" }]);
expect(findWorkflowEventShapeViolations({
type: "NodeCompleted", taskId: "FN-3", at: "2026-07-27T00:00:00.000Z",
nodeId: "execute", outcome: "success",
})).toEqual([]);
});
/*
FNXC:WorkflowEvents 2026-07-27-17:10 (U3, PR #2467 review — required-key half):
ALLOWED_EVENT_KEYS is a CEILING; this is the FLOOR. Without it a payload
missing `taskId` validated clean and got delivered, and a subscriber keying
derived state on `event.taskId` would write under `undefined` rather than
fail — the quiet-corruption mode this seam is supposed to be immune to.
One case per declared event type, so adding a type without its required keys
fails here rather than shipping an unvalidated payload.
*/
it("rejects a payload missing a COMMON required key", () => {
const { taskId: _taskId, ...noTaskId } = transitioned() as Record<string, unknown>;
expect(findWorkflowEventShapeViolations(noTaskId))
.toEqual([{ path: "taskId", reason: "missing-required-key" }]);
const { at: _at, ...noAt } = transitioned() as Record<string, unknown>;
expect(findWorkflowEventShapeViolations(noAt))
.toEqual([{ path: "at", reason: "missing-required-key" }]);
});
it("treats an EXPLICITLY undefined required key as missing, not as present", () => {
// `{ taskId: maybeId }` where maybeId is undefined is the same bug as
// omitting the key; the emitters' conditional spreads produce the other form.
expect(findWorkflowEventShapeViolations({ ...transitioned(), taskId: undefined }))
.toEqual([{ path: "taskId", reason: "missing-required-key" }]);
});
it("enforces the per-type required keys for every declared event type", () => {
const complete: Record<string, Record<string, unknown>> = {
TaskTransitioned: { type: "TaskTransitioned", taskId: "FN-1", at: "t", from: "todo", to: "in-progress" },
NodeEntered: { type: "NodeEntered", taskId: "FN-1", at: "t", nodeId: "execute" },
NodeCompleted: { type: "NodeCompleted", taskId: "FN-1", at: "t", nodeId: "execute", outcome: "success" },
RunSuspended: { type: "RunSuspended", taskId: "FN-1", at: "t", nodeId: "execute", reason: "capacity" },
RunResumed: { type: "RunResumed", taskId: "FN-1", at: "t", nodeId: "execute" },
};
const typeSpecificRequired: Record<string, string[]> = {
TaskTransitioned: ["from", "to"],
NodeEntered: ["nodeId"],
NodeCompleted: ["nodeId", "outcome"],
RunSuspended: ["nodeId", "reason"],
RunResumed: ["nodeId"],
};
for (const [type, payload] of Object.entries(complete)) {
// The complete payload is clean...
expect(findWorkflowEventShapeViolations(payload)).toEqual([]);
// ...and dropping any one type-specific required key is a violation.
for (const key of typeSpecificRequired[type]) {
const { [key]: _dropped, ...incomplete } = payload;
expect(findWorkflowEventShapeViolations(incomplete))
.toEqual([{ path: key, reason: "missing-required-key" }]);
}
}
});
it("keeps genuinely OPTIONAL keys optional — column, fromColumn/toColumn, runId, workflowId", () => {
// `column` is absent for a columnless node (`end`); `fromColumn`/`toColumn`
// for a suspension with no crossing; `runId`/`workflowId` are legitimately
// unresolvable at some emit sites. None may be forced into the floor.
expect(findWorkflowEventShapeViolations({ type: "NodeEntered", taskId: "FN-1", at: "t", nodeId: "end" })).toEqual([]);
expect(findWorkflowEventShapeViolations({ type: "RunSuspended", taskId: "FN-1", at: "t", nodeId: "n", reason: "capacity" })).toEqual([]);
});
it("reports BOTH an unknown key and a missing required key on one payload", () => {
const { to: _to, ...noTo } = transitioned() as Record<string, unknown>;
expect(findWorkflowEventShapeViolations({ ...noTo, error: "boom" })).toEqual([
{ path: "error", reason: "unknown-key" },
{ path: "to", reason: "missing-required-key" },
]);
});
it("rejects an unrecognised event TYPE outright rather than validating its fields", () => {
// An out-of-band type has no declared payload, so it gets no implicit
// permission to invent one.
expect(findWorkflowEventShapeViolations({ type: "TaskExploded", taskId: "FN-4", at: "x" }))
.toEqual([{ path: "type", reason: "unknown-type" }]);
});
it("allows an array of ids but rejects an array of objects", () => {
// Arrays are validated element-wise for any key the type declares; an
// UNDECLARED array key is refused by the allow-list before that runs, which
// is why both halves are asserted here.
expect(findWorkflowEventShapeViolations({ ...transitioned(), diffs: [{ field: "column" }] }))
.toEqual([{ path: "diffs", reason: "unknown-key" }]);
});
it("REFUSES a violating payload at the emit boundary so it never reaches a subscriber", async () => {
const bus = createWorkflowEventBus();
const subscriber = vi.fn();
bus.subscribe(subscriber, { name: "plugin" });
bus.emit({ ...transitioned(), to: { id: "in-review" } } as unknown as WorkflowLifecycleEvent);
await bus.drain();
// Degrades rather than throws: the emitter is post-commit, so a shape bug
// must not surface as a lifecycle failure.
expect(subscriber).not.toHaveBeenCalled();
});
});
describe("workflow event bus — reactions are non-authoritative (R5, KTD-3)", () => {
it("emitting with ZERO subscribers is a no-op that cannot throw", () => {
const bus = createWorkflowEventBus();
expect(bus.subscriberCount()).toBe(0);
expect(() => bus.emit(transitioned())).not.toThrow();
});
it("clear() drops every subscriber, so 'drop all subscribers' is expressible", async () => {
const bus = createWorkflowEventBus();
const subscriber = vi.fn();
bus.subscribe(subscriber, { name: "one" });
bus.clear();
expect(bus.subscriberCount()).toBe(0);
bus.emit(transitioned());
await bus.drain();
expect(subscriber).not.toHaveBeenCalled();
});
it("unsubscribe is idempotent — a double-off cannot remove someone else", async () => {
const bus = createWorkflowEventBus();
const survivor = vi.fn();
const off = bus.subscribe(vi.fn(), { name: "leaving" });
bus.subscribe(survivor, { name: "survivor" });
off();
off();
expect(bus.subscriberCount()).toBe(1);
bus.emit(transitioned());
await bus.drain();
expect(survivor).toHaveBeenCalledTimes(1);
});
});

157
packages/core/src/__tests__/workflow-lifecycle-traits.test.ts
View File

@@ -8,7 +8,8 @@ byte-identical on the default workflow. The custom cases prove KTD-10 fallback.
import { describe, expect, it } from "vitest";
import "../builtin-traits.js"; // register built-in traits
import { BUILTIN_CODING_WORKFLOW_IR } from "../builtin-coding-workflow-ir.js";
import { columnsWithFlag, columnHasFlag, resolveReboundTarget, resolveCompleteColumn, resolveMergeOrchestrationColumn } from "../workflow-lifecycle-traits.js";
import { columnsWithFlag, columnHasFlag, resolveReboundTarget, resolveCompleteColumn, resolveMergeOrchestrationColumn, resolveLifecycleColumns, resolveTaskLifecycleColumns } from "../workflow-lifecycle-traits.js";
import { BUILTIN_CODING_IDEAS_WORKFLOW_IR } from "../builtin-coding-ideas-workflow-ir.js";
import type { WorkflowIr } from "../workflow-ir-types.js";
describe("columnsWithFlag — builtin:coding trait→columnIds (R8)", () => {
@@ -118,3 +119,157 @@ describe("resolveCompleteColumn / resolveMergeOrchestrationColumn — U7", () =>
expect(resolveMergeOrchestrationColumn(bare)).toBeUndefined();
});
});
/*
FNXC:WorkflowLifecycleColumns 2026-07-27-09:20 (U1 — workflow-owned lifecycle):
Coverage for THE lifecycle-column resolution seam that Phases B–D convert ~207
hardcoded column literals onto. Two properties matter more than the happy path:
1. ID-INDEPENDENCE. The renamed-workflow case is the real assertion — it fails
if the resolver ever falls back to a legacy literal, which is exactly the
silent-guard failure mode this program exists to remove.
2. NO SUBSTITUTION. A workflow with no hold column must resolve `hold:
undefined`, not "the nearest thing". Substituting would turn "this workflow
has no capacity hold" into a wrong-but-plausible answer at 200 call sites.
*/
describe("resolveLifecycleColumns — U1 trait→role resolution", () => {
it("resolves the default coding workflow's roles to the legacy column ids", () => {
const columns = resolveLifecycleColumns(BUILTIN_CODING_WORKFLOW_IR);
expect(columns).toEqual({
intake: "triage",
hold: "todo",
wip: "in-progress",
review: "in-review",
complete: "done",
archived: "archived",
});
});
it("resolves Coding (Ideas) to its OWN intake column — id-independence, not a literal", () => {
const columns = resolveLifecycleColumns(BUILTIN_CODING_IDEAS_WORKFLOW_IR);
expect(columns?.intake).toBe("ideas");
// Ideas keeps `todo` as its hold column (R11's in-tree compatibility case),
// so this pair proves the resolver reads traits rather than assuming the
// default workflow's intake/hold pairing.
expect(columns?.hold).toBe("todo");
});
it("resolves a fully renamed workflow by trait, never by id", () => {
const renamed: WorkflowIr = {
version: "v2", name: "editorial",
columns: [
{ id: "backlog", name: "Backlog", traits: [{ trait: "intake" }] },
{ id: "drafting", name: "Drafting", traits: [{ trait: "hold", config: { release: "capacity" } }] },
{ id: "writing", name: "Writing", traits: [{ trait: "wip" }] },
{ id: "editorial-review", name: "Editorial review", traits: [{ trait: "merge" }] },
{ id: "published", name: "Published", traits: [{ trait: "complete" }] },
{ id: "shelved", name: "Shelved", traits: [{ trait: "archived" }] },
],
nodes: [{ id: "start", kind: "start", column: "backlog" }],
edges: [],
} as WorkflowIr;
expect(resolveLifecycleColumns(renamed)).toEqual({
intake: "backlog",
hold: "drafting",
wip: "writing",
review: "editorial-review",
complete: "published",
archived: "shelved",
});
});
it("leaves an absent role undefined instead of substituting an unrelated column", () => {
const noHold: WorkflowIr = {
version: "v2", name: "no-hold",
columns: [
{ id: "inbox", name: "Inbox", traits: [{ trait: "intake" }] },
{ id: "doing", name: "Doing", traits: [{ trait: "wip" }] },
{ id: "shipped", name: "Shipped", traits: [{ trait: "complete" }] },
],
nodes: [{ id: "start", kind: "start", column: "inbox" }],
edges: [],
} as WorkflowIr;
const columns = resolveLifecycleColumns(noHold);
expect(columns).toBeDefined();
expect(columns?.hold).toBeUndefined();
// The nearby columns are still resolved — absence is per-role, not per-workflow.
expect(columns?.intake).toBe("inbox");
expect(columns?.wip).toBe("doing");
expect(columns?.archived).toBeUndefined();
});
it("returns undefined (not a struct of undefineds) for a v1 / column-less IR", () => {
// The caller must be able to distinguish "no hold column declared" from
// "no column vocabulary at all"; only the latter licenses skip-and-log.
const v1 = { version: "v1", name: "legacy", nodes: [], edges: [] } as unknown as WorkflowIr;
expect(resolveLifecycleColumns(v1)).toBeUndefined();
});
it("picks the FIRST column carrying a role when several do", () => {
const twoHolds: WorkflowIr = {
version: "v2", name: "two-holds",
columns: [
{ id: "inbox", name: "Inbox", traits: [{ trait: "intake" }] },
{ id: "hold-a", name: "Hold A", traits: [{ trait: "hold", config: { release: "capacity" } }] },
{ id: "hold-b", name: "Hold B", traits: [{ trait: "hold", config: { release: "capacity" } }] },
],
nodes: [{ id: "start", kind: "start", column: "inbox" }],
edges: [],
} as WorkflowIr;
expect(resolveLifecycleColumns(twoHolds)?.hold).toBe("hold-a");
});
});
describe("resolveTaskLifecycleColumns — U1 store-aware form", () => {
function makeStore(overrides: Partial<Record<string, unknown>> = {}) {
const definitionReads: string[] = [];
const store = {
getTaskWorkflowSelection: (taskId: string) => ({ workflowId: taskId === "T-IDEAS" ? "wf-ideas" : "wf-custom" }),
getWorkflowDefinition: async (workflowId: string) => {
definitionReads.push(workflowId);
return {
id: workflowId,
ir: workflowId === "wf-ideas" ? BUILTIN_CODING_IDEAS_WORKFLOW_IR : BUILTIN_CODING_WORKFLOW_IR,
};
},
...overrides,
};
return { store: store as never, definitionReads };
}
it("resolves a task's roles through its workflow selection", async () => {
const { store } = makeStore();
await expect(resolveTaskLifecycleColumns(store, "T-1")).resolves.toEqual({
intake: "triage", hold: "todo", wip: "in-progress",
review: "in-review", complete: "done", archived: "archived",
});
});
it("resolves each workflow's IR ONCE per pass when the caller shares a cache", async () => {
// The reason the cache is caller-owned: a sweep over N cards on one workflow
// must read one IR, not N. Assert on the resolver's own read count.
const { store, definitionReads } = makeStore();
const cache = new Map();
await resolveTaskLifecycleColumns(store, "T-1", cache);
await resolveTaskLifecycleColumns(store, "T-2", cache);
await resolveTaskLifecycleColumns(store, "T-3", cache);
expect(definitionReads).toEqual(["wf-custom"]);
});
it("reads each DISTINCT workflow once, so a mixed-workflow sweep stays correct", async () => {
const { store, definitionReads } = makeStore();
const cache = new Map();
const first = await resolveTaskLifecycleColumns(store, "T-1", cache);
const ideas = await resolveTaskLifecycleColumns(store, "T-IDEAS", cache);
expect(first?.intake).toBe("triage");
expect(ideas?.intake).toBe("ideas");
expect(definitionReads).toEqual(["wf-custom", "wf-ideas"]);
});
it("returns undefined when the workflow resolves to no column vocabulary", async () => {
const { store } = makeStore({
getWorkflowDefinition: async () => ({ id: "wf-v1", ir: { version: "v1", name: "legacy", nodes: [], edges: [] } }),
});
await expect(resolveTaskLifecycleColumns(store, "T-1")).resolves.toBeUndefined();
});
});

197
packages/core/src/__tests__/workflow-parity.test.ts
View File

@@ -1,197 +0,0 @@
import { describe, expect, it } from "vitest";
import {
compareWorkflowRunAudits,
compareWorkflowRunObservations,
buildWorkflowObservationFromTask,
buildWorkflowObservation,
deriveStageTransitions,
DEFAULT_WORKFLOW_INVARIANTS,
type RunAuditEvent,
type WorkflowRunObservation,
} from "../index.js";
function observation(overrides: Partial<WorkflowRunObservation> = {}): WorkflowRunObservation {
return {
stageTransitions: ["triage", "execute", "review", "merge"],
terminalColumn: "done",
terminalStatus: null,
reviewVerdict: "APPROVE",
mergeOutcome: "merged",
invariants: {
fileScopeGuardOutcome: "pass",
squashMergeContractOutcome: "pass",
autoMergeTerminalUntilMergedRespected: true,
moveTaskHardCancelRespected: true,
},
...overrides,
};
}
function auditEvent(mutationType: string, target: string, phase: string): RunAuditEvent {
return {
id: `${mutationType}-${target}`,
timestamp: new Date().toISOString(),
taskId: "FN-1",
agentId: "executor",
runId: "run-1",
domain: "database",
mutationType,
target,
metadata: { phase },
};
}
describe("workflow parity", () => {
it("agrees for identical observations", () => {
const report = compareWorkflowRunObservations(observation(), observation());
expect(report).toEqual({ agree: true, diffs: [] });
});
it("reports lifecycle transition drift", () => {
const report = compareWorkflowRunObservations(
observation(),
observation({ stageTransitions: ["triage", "execute", "merge"] }),
);
expect(report.agree).toBe(false);
expect(report.diffs).toEqual(
expect.arrayContaining([expect.objectContaining({ field: "stageTransitions", category: "lifecycle" })]),
);
});
it("reports terminal status and review verdict drift", () => {
const report = compareWorkflowRunObservations(
observation(),
observation({ terminalColumn: "in-review", reviewVerdict: "REVISE" }),
);
expect(report.agree).toBe(false);
expect(report.diffs).toEqual(
expect.arrayContaining([
expect.objectContaining({ field: "terminalColumn" }),
expect.objectContaining({ field: "reviewVerdict" }),
]),
);
});
it("reports file-scope guard invariant drift", () => {
const report = compareWorkflowRunObservations(
observation(),
observation({ invariants: { ...observation().invariants, fileScopeGuardOutcome: "fail" } }),
);
expect(report.diffs).toEqual(
expect.arrayContaining([
expect.objectContaining({ field: "invariants.fileScopeGuardOutcome", category: "invariant" }),
]),
);
});
it("reports squash merge invariant drift", () => {
const report = compareWorkflowRunObservations(
observation(),
observation({ invariants: { ...observation().invariants, squashMergeContractOutcome: "blocked" } }),
);
expect(report.diffs).toEqual(
expect.arrayContaining([
expect.objectContaining({ field: "invariants.squashMergeContractOutcome", category: "invariant" }),
]),
);
});
it("reports auto-merge terminal invariant drift", () => {
const report = compareWorkflowRunObservations(
observation(),
observation({
invariants: { ...observation().invariants, autoMergeTerminalUntilMergedRespected: false },
}),
);
expect(report.diffs).toEqual(
expect.arrayContaining([
expect.objectContaining({ field: "invariants.autoMergeTerminalUntilMergedRespected", category: "invariant" }),
]),
);
});
it("reports moveTask hard-cancel invariant drift", () => {
const report = compareWorkflowRunObservations(
observation(),
observation({ invariants: { ...observation().invariants, moveTaskHardCancelRespected: false } }),
);
expect(report.diffs).toEqual(
expect.arrayContaining([
expect.objectContaining({ field: "invariants.moveTaskHardCancelRespected", category: "invariant" }),
]),
);
});
it("agrees on identical comparable run-audit slices", () => {
const events = [
auditEvent("task:move", "FN-1", "execute"),
auditEvent("task:update", "FN-1", "review"),
];
const report = compareWorkflowRunAudits(events, events);
expect(report).toEqual({ agree: true, diffs: [] });
});
it("reports run-audit drift", () => {
const legacy = [auditEvent("task:move", "FN-1", "execute")];
const interpreter = [auditEvent("task:update", "FN-2", "review")];
const report = compareWorkflowRunAudits(legacy, interpreter);
expect(report.agree).toBe(false);
expect(report.diffs).toEqual(
expect.arrayContaining([
expect.objectContaining({ field: "audit[0].mutationType", category: "audit" }),
expect.objectContaining({ field: "audit[0].target", category: "audit" }),
expect.objectContaining({ field: "audit[0].phase", category: "audit" }),
]),
);
});
});
describe("observation builders (CU-U5)", () => {
it("deriveStageTransitions maps columns to stages and collapses repeats", () => {
expect(deriveStageTransitions(["todo", "in-progress", "in-progress", "in-review", "done"]))
.toEqual(["triage", "execute", "review", "merge"]);
expect(deriveStageTransitions([])).toEqual([]);
});
it("buildWorkflowObservationFromTask reads terminal lifecycle + derives stages from columnSequence", () => {
const obs = buildWorkflowObservationFromTask(
{ column: "done", status: null, review: { verdict: "APPROVE" }, mergeDetails: { outcome: "merged" } },
{ columnSequence: ["todo", "in-progress", "in-review", "done"] },
);
expect(obs.stageTransitions).toEqual(["triage", "execute", "review", "merge"]);
expect(obs.terminalColumn).toBe("done");
expect(obs.reviewVerdict).toBe("APPROVE");
expect(obs.mergeOutcome).toBe("merged");
expect(obs.invariants).toEqual(DEFAULT_WORKFLOW_INVARIANTS);
});
it("buildWorkflowObservationFromTask infers merged from terminal column when mergeDetails absent", () => {
const obs = buildWorkflowObservationFromTask({ column: "done" });
expect(obs.mergeOutcome).toBe("merged");
expect(obs.stageTransitions).toEqual(["merge"]); // terminal-only fallback
});
it("a task and an equivalent interpreter parts observation compare as agree", () => {
const legacy = buildWorkflowObservationFromTask(
{ column: "done", review: { verdict: "APPROVE" }, mergeDetails: { outcome: "merged" } },
{ columnSequence: ["in-progress", "in-review", "done"] },
);
const interpreter = buildWorkflowObservation({
stageTransitions: ["execute", "review", "merge"],
terminalColumn: "done",
reviewVerdict: "APPROVE",
mergeOutcome: "merged",
});
expect(compareWorkflowRunObservations(legacy, interpreter).agree).toBe(true);
});
it("a divergent stage sequence surfaces an error-severity lifecycle drift", () => {
const legacy = buildWorkflowObservationFromTask({ column: "done" }, { columnSequence: ["in-progress", "in-review", "done"] });
const interpreter = buildWorkflowObservation({ stageTransitions: ["execute", "merge"], terminalColumn: "done", mergeOutcome: "merged" });
const report = compareWorkflowRunObservations(legacy, interpreter);
expect(report.agree).toBe(false);
expect(report.diffs).toEqual(
expect.arrayContaining([expect.objectContaining({ field: "stageTransitions", category: "lifecycle", severity: "error" })]),
);
});
});

43
packages/core/src/index.gate.ts
View File

@@ -407,7 +407,6 @@ export {
workflowHasColumn,
} from "./workflow-transitions.js";
export type { ColumnAdjacency } from "./workflow-transitions.js";
export { isWorkflowColumnsEnabled } from "./workflow-columns-settings.js";
// ── U8: pre-evaluated plugin gate verdicts (KTD-2) ───────────────────────────
export {
findWorkflowColumn,
@@ -1967,41 +1966,6 @@ export {
postMergeVerificationOptionalGroupNode,
} from "./builtin-post-merge-group.js";
export type { PostMergeOptionalGroupSpec } from "./builtin-post-merge-group.js";
export {
WORKFLOW_COMPARABLE_AUDIT_MUTATIONS,
WORKFLOW_PARITY_OBSERVED_MUTATION,
WORKFLOW_PARITY_DRIFT_MUTATION,
compareWorkflowRunAudits,
compareWorkflowRunObservations,
extractWorkflowAuditObservations,
DEFAULT_WORKFLOW_INVARIANTS,
deriveStageTransitions,
buildWorkflowObservationFromTask,
buildWorkflowObservation,
checkTransitionParity,
countDualAcceptDisagreements,
computeWorkflowColumnsGraduationReport,
DUAL_ACCEPT_PARITY_MUTATIONS,
} from "./workflow-parity.js";
export type {
WorkflowAuditObservation,
WorkflowParityDiff,
WorkflowParityDiffCategory,
WorkflowParityDiffSeverity,
WorkflowParityDriftReport,
WorkflowReliabilityInvariantSignals,
WorkflowRunObservation,
WorkflowStage,
WorkflowObservationTaskInput,
WorkflowObservationBuildOptions,
WorkflowObservationParts,
WorkflowParitySummary,
TransitionParityDiff,
TransitionParityReport,
DualAcceptDisagreementReport,
WorkflowColumnsGraduationReport,
GraduationReportInputs,
} from "./workflow-parity.js";
export { isResearchExperimentalEnabled, resolveResearchSettings } from "./research-settings.js";
export type { ResolvedResearchSettings } from "./research-settings.js";
export { isEvalsExperimentalEnabled, resolveEvalSettings } from "./eval-settings.js";
@@ -2280,7 +2244,12 @@ Cutover (IR-driven lifecycle) barrel sync — same failure class as the classify
*/
export { resolveCreationColumn } from "./workflow-ir.js";
export { resolveWipBudgetColumns } from "./workflow-capacity.js";
export { columnsWithFlag, columnHasFlag, resolveReboundTarget, resolveCompleteColumn, resolveMergeOrchestrationColumn } from "./workflow-lifecycle-traits.js";
export { createWorkflowEventBus, getWorkflowEventBus, emitWorkflowLifecycleEvent, resetWorkflowEventBusForTesting } from "./workflow-events.js";
export type { WorkflowEventBus, WorkflowEventSubscriber, WorkflowEventSubscription } from "./workflow-events.js";
export { findWorkflowEventShapeViolations, isIdsOnlyWorkflowEvent, MAX_ID_VALUE_LENGTH } from "./types/workflow-events.js";
export type { WorkflowLifecycleEvent, WorkflowLifecycleEventType, WorkflowLifecycleEventBase, TaskTransitionedEvent, NodeEnteredEvent, NodeCompletedEvent, RunSuspendedEvent, RunResumedEvent, WorkflowEventShapeViolation } from "./types/workflow-events.js";
export { columnsWithFlag, columnHasFlag, resolveReboundTarget, resolveCompleteColumn, resolveMergeOrchestrationColumn, resolveLifecycleColumns, resolveTaskLifecycleColumns } from "./workflow-lifecycle-traits.js";
export type { LifecycleColumns } from "./workflow-lifecycle-traits.js";
export { resolveReviewLevelSteps, applyReviewLevelPreset } from "./review-level-preset.js";
export { LEGACY_STATUS_ADOPTION, resolveLegacyStatusAdoption, resolveReviewLevelBackfill, planLegacyAdoption, resolveOrphanedPendingStepResults, type LegacyAdoptionPlan, type LegacyAdoptionCandidate, type LegacyAdoptionAction, type LegacyAdoptionKind } from "./legacy-adoption.js";
export { hashWorkflowIr, computeWorkflowIrPin, detectWorkflowDrift, type WorkflowIrPin } from "./workflow-ir-resolver.js";

43
packages/core/src/index.ts
View File

@@ -438,7 +438,6 @@ export {
workflowHasColumn,
} from "./workflow-transitions.js";
export type { ColumnAdjacency } from "./workflow-transitions.js";
export { isWorkflowColumnsEnabled } from "./workflow-columns-settings.js";
// ── U8: pre-evaluated plugin gate verdicts (KTD-2) ───────────────────────────
export {
findWorkflowColumn,
@@ -447,7 +446,12 @@ export {
export type { PluginGateVerdict, ColumnPluginGate } from "./plugin-gate-verdict.js";
// ── U6: workflow capacity (WIP) resolution shared by store + sweep ───────────
export { resolveColumnCapacity, resolveWipBudgetColumns, DEFAULT_WORKFLOW_POOL_ID } from "./workflow-capacity.js";
export { columnsWithFlag, columnHasFlag, resolveReboundTarget, resolveCompleteColumn, resolveMergeOrchestrationColumn } from "./workflow-lifecycle-traits.js";
export { createWorkflowEventBus, getWorkflowEventBus, emitWorkflowLifecycleEvent, resetWorkflowEventBusForTesting } from "./workflow-events.js";
export type { WorkflowEventBus, WorkflowEventSubscriber, WorkflowEventSubscription } from "./workflow-events.js";
export { findWorkflowEventShapeViolations, isIdsOnlyWorkflowEvent, MAX_ID_VALUE_LENGTH } from "./types/workflow-events.js";
export type { WorkflowLifecycleEvent, WorkflowLifecycleEventType, WorkflowLifecycleEventBase, TaskTransitionedEvent, NodeEnteredEvent, NodeCompletedEvent, RunSuspendedEvent, RunResumedEvent, WorkflowEventShapeViolation } from "./types/workflow-events.js";
export { columnsWithFlag, columnHasFlag, resolveReboundTarget, resolveCompleteColumn, resolveMergeOrchestrationColumn, resolveLifecycleColumns, resolveTaskLifecycleColumns } from "./workflow-lifecycle-traits.js";
export type { LifecycleColumns } from "./workflow-lifecycle-traits.js";
export { resolveReviewLevelSteps, applyReviewLevelPreset } from "./review-level-preset.js";
export {
LEGACY_STATUS_ADOPTION,
@@ -2136,41 +2140,6 @@ export {
postMergeVerificationOptionalGroupNode,
} from "./builtin-post-merge-group.js";
export type { PostMergeOptionalGroupSpec } from "./builtin-post-merge-group.js";
export {
WORKFLOW_COMPARABLE_AUDIT_MUTATIONS,
WORKFLOW_PARITY_OBSERVED_MUTATION,
WORKFLOW_PARITY_DRIFT_MUTATION,
compareWorkflowRunAudits,
compareWorkflowRunObservations,
extractWorkflowAuditObservations,
DEFAULT_WORKFLOW_INVARIANTS,
deriveStageTransitions,
buildWorkflowObservationFromTask,
buildWorkflowObservation,
checkTransitionParity,
countDualAcceptDisagreements,
computeWorkflowColumnsGraduationReport,
DUAL_ACCEPT_PARITY_MUTATIONS,
} from "./workflow-parity.js";
export type {
WorkflowAuditObservation,
WorkflowParityDiff,
WorkflowParityDiffCategory,
WorkflowParityDiffSeverity,
WorkflowParityDriftReport,
WorkflowReliabilityInvariantSignals,
WorkflowRunObservation,
WorkflowStage,
WorkflowObservationTaskInput,
WorkflowObservationBuildOptions,
WorkflowObservationParts,
WorkflowParitySummary,
TransitionParityDiff,
TransitionParityReport,
DualAcceptDisagreementReport,
WorkflowColumnsGraduationReport,
GraduationReportInputs,
} from "./workflow-parity.js";
export { isResearchExperimentalEnabled, resolveResearchSettings } from "./research-settings.js";
export type { ResolvedResearchSettings } from "./research-settings.js";
export { isEvalsExperimentalEnabled, resolveEvalSettings } from "./eval-settings.js";

21
packages/core/src/store.ts
View File

@@ -56,7 +56,6 @@ import "./builtin-traits.js";
// the `step-headings` parser through the registry (proving the registry path),
// staying byte-identical with the direct extracted function.
import type { StoredWorkflowRow, WorkflowDefinition, WorkflowDefinitionInput, WorkflowDefinitionUpdate, WorkflowNodeLayout } from "./workflow-definition-types.js";
import { type WorkflowParitySummary, type WorkflowColumnsGraduationReport } from "./workflow-parity.js";
/** Tags WorkflowStep rows materialized by compiling a workflow so they can be
* filtered out of the user-facing step manager and cleaned up on re-selection. */
@@ -99,14 +98,14 @@ import { moveTaskImpl, moveTaskIfImpl, handoffToReviewImpl, moveTaskInternalImpl
import { recordGoalCitationsImpl, insertTaskWithFtsRecoveryImpl2, assertTaskIdAvailableImpl, atomicWriteTaskJsonImpl2, createTaskWithDistributedReservationImpl, toStoredWorkflowStepImpl, ensureWorkflowStepForTemplateImpl, resolveEnabledWorkflowStepsImpl, setTaskBranchGroupImpl, getTaskColumnsImpl, prepareWorkflowMovePolicyPreflightImpl, updateTaskCustomFieldsImpl, listWorkflowPromptOverridesForProjectImpl, listWorkflowWorkItemsForTaskImpl, listDueWorkflowWorkItemsImpl, rewriteBlockedByResidueDependentsForRemovalImpl, getAllDocumentsImpl, deleteWorkflowStepImpl, toWorkflowDefinitionImpl, materializeDefaultWorkflowStepsImpl, reconcileTaskCustomFieldsForSchemaImpl, getTaskMovedCountsByDayImpl, getGoalStoreImpl, upsertTaskCommitAssociationImpl } from "./task-store/workflow-task-create-ops.js";
import { applyLegacyWorkflowStepOverridesImpl, archiveDbImpl, assertNoDependencyCycleImpl, atomicCreateTaskJsonImpl, buildActiveTaskDependencyLookupImpl, buildArchivedAgentLogFieldsImpl, buildTaskIdIntegrityFallbackReportImpl, createBranchGroupImpl, dbImpl, detectAndCacheTaskIdIntegrityReportImpl, findLiveDependentsImpl, findLiveLineageChildrenImpl, getLegacyWorkflowStepSnapshotImpl, getMalformedTaskMetadataReasonImpl, getMergeQueuedTaskIdsAsyncImpl, insertRunAuditEventRowImpl, insertTaskImpl, invokeTaskCreatedHookImpl, isTaskArchivedAsyncImpl, isTaskArchivedImpl, isTaskIdPresentInArchivedTasksTableAsyncImpl, isTaskIdPresentInArchivedTasksTableImpl, logTaskCreateConflictImpl, maybeResolveTombstonedTaskIdImpl, mergeTaskIdIntegrityReportsImpl, optionalGroupIdSetImpl, patchTaskRowInTransactionImpl, readConfigFastImpl, readConfigImpl, readPromptForArchiveImpl, readTaskFromDbImpl, reconcileDistributedTaskIdStateOnOpenImpl, recordActivityFromListenerImpl, recordDependencyCycleRejectedAuditImpl, refreshTaskIdIntegrityReportImpl, resolveLocalNodeIdForTaskAllocationImpl, runTaskFtsWriteWithRecoveryImpl, scanAndRecordCitationsImpl, taskIdExistsAnywhereImpl, throwSoftDeletedWriteBlockedImpl, toBuiltInWorkflowStepImpl, trackDeferredTaskCreatedWorkImpl, upsertTaskImpl, withConfigLockImpl, withTaskLockImpl, withWorktreeAllocationLockImpl } from "./task-store/task-id-integrity.js";
import { claimNextToolFailureRetryImpl, createTaskVerificationRequestImpl, claimTaskVerificationRequestImpl, finishTaskVerificationRequestImpl, clearNearDuplicateReferencesToFailSoftImpl, clearWorkflowRunStepInstancesAsyncImpl, clearWorkflowRunStepInstancesImpl, computeMovedSettingsTargetWorkflowIdsImpl, ensureBranchGroupForSourceImpl, ensurePrEntityForSourceImpl, findRecentTasksByContentFingerprintImpl, getActiveMergingTaskImpl, getActivePrEntityBySourceImpl, getBranchGroupByBranchNameImpl, getBranchGroupBySourceImpl, getBranchGroupImpl, getBranchProgressByTaskImpl, getMutationsForRunImpl, getPrEntityByNumberImpl, getPrEntityImpl, getPrThreadStateImpl, getTasksByAssignedAgentImpl, getWorkflowPromptOverridesAsyncImpl, getWorkflowSettingValuesAsyncImpl, getWorkflowSettingValuesImpl, getWorkflowSettingsProjectIdImpl, getWorkflowWorkItemImpl, insertCompletionHandoffWorkflowWorkAuditImpl, listActivePrEntitiesImpl, listBranchGroupsImpl, listPrThreadStatesImpl, listTasksByBranchGroupImpl, listWorkflowSettingValuesForProjectImpl, loadWorkflowRunBranchesImpl, loadWorkflowRunStepInstancesAsyncImpl, loadWorkflowRunStepInstancesImpl, markToolFailureRetryExhaustedAuditImpl, mergeCustomFieldPatchImpl, normalizeMergeRequestStateImpl, normalizeWorkflowWorkItemKindImpl, normalizeWorkflowWorkItemStateImpl, parseWorkflowPromptOverrideJsonImpl, recordPrThreadOutcomeImpl, resetAllStepsToPendingImpl, resetPromptCheckboxesImpl, resolveWorkflowMoveActorImpl, resolveWorkflowSettingDeclarationsImpl, saveWorkflowRunStepInstanceAsyncImpl, saveWorkflowRunStepInstanceImpl, transitionMergeRequestStateImpl, transitionWorkflowWorkItemSyncImpl, updateTaskImpl, updateWorkflowPromptOverridesImpl, upsertMergeRequestRecordImpl, workflowStateForMergeRequestStateImpl } from "./task-store/branch-and-pr-entities.js";
import { addPrInfoImpl, addSteeringCommentImpl, archiveAllDoneImpl, cleanupStaleMergeQueueRowsImpl, clearCompletionHandoffAcceptedMarkerImpl, clearDoneTransientFieldsImpl, clearStaleExecutionStartBranchReferencesImpl, computeWorkflowColumnsGraduationReportImpl, deleteTaskCommentImpl, deleteTaskDocumentImpl, emitUsageEventImpl, enqueueMergeQueueImpl, getAgentLogCountImpl, getAgentLogsImpl, getArtifactImpl, getArtifactsImpl, getAttachmentImpl, getCompletionHandoffAcceptedMarkerImpl, getTaskDocumentImpl, getTaskDocumentRevisionsImpl, getTaskDocumentsImpl, insertArtifactRowImpl, linkGithubIssueImpl, listWorkflowWorkItemsForTaskSyncImpl, moveToDoneImpl, parseDependenciesFromPromptImpl, parseFileScopeFromPromptImpl, parseStepsFromPromptImpl, peekMergeQueueHeadImpl, peekMergeQueueImpl, readPreArchiveColumnFromTaskFileImpl, recordPluginActivationImpl, recordRunAuditEventBackendImpl, removePrInfoByNumberImpl, resolvePrimaryPrInfoImpl, resolveUnarchiveTargetColumnImpl, rewriteLineageChildrenForRemovalImpl, runGitCommandImpl, stopWatchingImpl, syncAgentTaskLinkOnReassignmentImpl, updateArtifactImpl, updateGithubTrackingImpl, updatePrInfoByNumberImpl, updateTaskCommentImpl, upsertPrInfoByNumberImpl, writeArtifactDataImpl } from "./task-store/task-artifacts-ops.js";
import { addPrInfoImpl, addSteeringCommentImpl, archiveAllDoneImpl, cleanupStaleMergeQueueRowsImpl, clearCompletionHandoffAcceptedMarkerImpl, clearDoneTransientFieldsImpl, clearStaleExecutionStartBranchReferencesImpl, deleteTaskCommentImpl, deleteTaskDocumentImpl, emitUsageEventImpl, enqueueMergeQueueImpl, getAgentLogCountImpl, getAgentLogsImpl, getArtifactImpl, getArtifactsImpl, getAttachmentImpl, getCompletionHandoffAcceptedMarkerImpl, getTaskDocumentImpl, getTaskDocumentRevisionsImpl, getTaskDocumentsImpl, insertArtifactRowImpl, linkGithubIssueImpl, listWorkflowWorkItemsForTaskSyncImpl, moveToDoneImpl, parseDependenciesFromPromptImpl, parseFileScopeFromPromptImpl, parseStepsFromPromptImpl, peekMergeQueueHeadImpl, peekMergeQueueImpl, readPreArchiveColumnFromTaskFileImpl, recordPluginActivationImpl, recordRunAuditEventBackendImpl, removePrInfoByNumberImpl, resolvePrimaryPrInfoImpl, resolveUnarchiveTargetColumnImpl, rewriteLineageChildrenForRemovalImpl, runGitCommandImpl, stopWatchingImpl, syncAgentTaskLinkOnReassignmentImpl, updateArtifactImpl, updateGithubTrackingImpl, updatePrInfoByNumberImpl, updateTaskCommentImpl, upsertPrInfoByNumberImpl, writeArtifactDataImpl } from "./task-store/task-artifacts-ops.js";
import { approveCliAutonomyImpl, approveWorkflowCliCommandImpl, cleanupOrphanedMaterializedStepsImpl, consumePluginGateVerdictsImpl, getAgentLogsByTimeRangeImpl, getDatabaseHealthImpl, getDistributedTaskIdAllocatorImpl, getExperimentSessionStoreImpl, getInReviewDurationEventsImpl, getMissionStoreImpl, getIdeationStoreImpl, getPluginStoreImpl, getSecretsStoreImpl, getSettingsSyncImpl, getTaskMergedTaskIdsImpl, getTaskWorkflowSelectionImpl, getImportTranslationImpl, recordImportTranslationImpl, pruneImportTranslationsImpl, type ImportTranslationCacheKey, type ImportTranslationCacheEntry, getVerificationCacheHitImpl, getWorkflowDefinitionImpl, healthCheckImpl, importLegacyAgentLogsOnceImpl, insertWorkflowDefinitionSyncImpl, isCliAutonomyApprovedImpl, isPluginInstalledImpl, isWorkflowCliCommandApprovedImpl, listWorkflowDefinitionsImpl, materializeExplicitWorkflowStepsImpl, materializeWorkflowStepsImpl, migrateActiveArchivedTasksToArchiveDbImpl, migrateLegacyArchiveEntriesToArchiveDbImpl, nextWorkflowDefinitionIdImpl, occupantsByColumnForWorkflowImpl, parseWorkflowLayoutImpl, pruneAgentLogFilesImpl, purgeTaskWorkflowSelectionRowsImpl, readAllWorkflowDefinitionsImpl, readRawProjectSettingsImpl, recordPluginGateVerdictImpl, recordVerificationCachePassImpl, removeMaterializedSelectionImpl, resolvePluginWorkflowStepImpl, resolveTaskWorkflowIrSyncImpl, revokeCliAutonomyImpl, selectTaskWorkflowAndReconcileImpl, writeTaskWorkflowSelectionImpl, getTaskWorkflowSelectionAsyncImpl, } from "./task-store/workflow-definitions.js";
import { getTaskCommitAssociationsByLineageIdImpl, replaceLegacyTaskCommitAssociationsImpl } from "./task-store/task-commit-associations.js";
import { findRecentTasksBySourceParentTaskIdImpl } from "./task-store/branch-and-pr-entities.js";
import { addTaskCommentImpl, applyBuiltInPromptOverridesAsyncImpl, applyBuiltInPromptOverridesSyncImpl, areAllDependenciesDoneImpl, artifactStoredNameImpl, assertWorkflowIrTraitsValidImpl, clearActivityLogImpl, clearTaskWorkflowSelectionImpl, deleteTaskByIdImpl, getDefaultWorkflowIdImpl, resolveOriginWorkflowOverrideIdImpl, type TaskOriginWorkflowKind, getInsightStoreImpl, getMergeQueuedTaskIdsImpl, getMergeRequestRecordImpl, getMergeRequestRecordAsyncImpl, getResearchStoreImpl, getTaskIdFromDirImpl, getTodoStoreImpl, getWorkflowWorkItemByIdentityImpl, hasActiveTaskImpl, invalidateConfigCacheAfterMigrationImpl, isTaskIdConflictErrorImpl, listLegacyAutoMergeStampCandidatesImpl, readTaskRowFromDbImpl, recordBranchGroupMemberLandedImpl, refreshDatabaseHealthAsyncImpl, refreshDatabaseHealthImpl, resolveEffectiveWorkflowIdSyncImpl, resolveTaskCustomFieldDefsSyncImpl, resolveWorkflowBypassGuardsImpl, serializeConfigForDiskImpl, setPluginWorkflowStepTemplatesImpl, shouldSkipWorkflowMovePoliciesImpl, suppressWatcherImpl, upsertTaskWithFtsRecoveryImpl } from "./task-store/task-store-helpers.js";
import { getTaskSelectClauseImpl2, createTaskPersistSerializationContextImpl, getTaskPersistValuesImpl, getTaskPatchDescriptorsImpl, normalizeTaskFromDiskImpl, writeTaskJsonFileImpl, rowToPrEntityImpl, generatePrEntityIdImpl, readTaskForMoveImpl, rowToMergeQueueEntryImpl, rowToMergeRequestRecordImpl, rowToCompletionHandoffMarkerImpl, rowToWorkflowWorkItemImpl, rowToRunAuditEventImpl } from "./task-store/task-row-mappers.js";
import { getTaskSelectClauseWithActivityLogLimitImpl, getChangedTaskColumnsImpl, getSoftDeletedWriteConflictImpl, readTaskJsonImpl, writeConfigImpl, _maybeAutoArchiveSameAgentDuplicateBackendImpl, updateBranchGroupImpl, updatePrEntityImpl, listTasksForGithubTrackingReconcileImpl, listTasksForGitlabTrackingReconcileImpl, renewCheckoutLeaseImpl, updateTaskAtomicImpl, getWorkflowPromptOverridesImpl, updateWorkflowSettingValuesImpl, rollbackConfigurationImpl, cancelActiveWorkflowWorkItemsForTaskImpl, setCompletionHandoffAcceptedMarkerImpl, reconcileLegacyAutoMergeStampsImpl, recoverExpiredMergeQueueLeasesImpl, rewriteDependentsForRemovalImpl, cleanupBranchForTaskImpl, addAttachmentImpl, deleteAttachmentImpl, registerArtifactImpl, updatePrInfoImpl, unlinkGithubIssueImpl, cleanupArchivedTasksImpl, generatePromptFromArchiveEntryImpl, listWorkflowOccupantTaskIdsImpl, evacuateCustomColumnsToLegacyImpl, listApprovedCliAutonomyAdaptersImpl, closeImpl, getActivityLogImpl } from "./task-store/task-mutation-ops.js";
import { getOrCreateForProjectImpl, listGoalCitationsImpl, atomicWriteTaskJsonWithAuditImpl, duplicateTaskImpl, listStrandedRefinementsImpl, tryClaimCheckoutImpl, evaluateWorkflowMovePoliciesImpl, recordRunAuditEventImpl, getRunAuditEventsImpl, getWorkflowParitySummaryImpl, dequeueMergeQueueOnColumnExitImpl, updateIssueInfoImpl, listWorkflowStepsImpl, getWorkflowStepImpl, createWorkflowDefinitionImpl, countActiveInCapacitySlotSyncImpl, countActiveInCapacitySlotAsyncImpl, generateSpecifiedPromptImpl, recordActivityImpl, getEvalStoreImpl } from "./task-store/project-store-ops.js";
import { getOrCreateForProjectImpl, listGoalCitationsImpl, atomicWriteTaskJsonWithAuditImpl, duplicateTaskImpl, listStrandedRefinementsImpl, tryClaimCheckoutImpl, evaluateWorkflowMovePoliciesImpl, recordRunAuditEventImpl, getRunAuditEventsImpl, dequeueMergeQueueOnColumnExitImpl, updateIssueInfoImpl, listWorkflowStepsImpl, getWorkflowStepImpl, createWorkflowDefinitionImpl, countActiveInCapacitySlotSyncImpl, countActiveInCapacitySlotAsyncImpl, generateSpecifiedPromptImpl, recordActivityImpl, getEvalStoreImpl } from "./task-store/project-store-ops.js";
import { markLegacyAutoMergeStampsOnceImpl, appendAgentLogImpl, importLegacyAgentLogsImpl, cleanupNoOpTaskMovedActivityRowsOnceImpl, runWorkflowColumnsIntegrityPassImpl, backfillCommitAssociationDiffStatsImpl } from "./task-store/workflow-integrity.js";
import { saveWorkflowRunBranchImpl, clearNearDuplicateReferencesToImpl, selectNextTaskForAgentImpl, pauseTaskImpl, clearLinkedAgentTaskIdsImpl, listArtifactsImpl, rehomeOccupantImpl } from "./task-store/branch-group-ops.js";
import { taskToArchiveEntryImpl, deleteTaskBackendImpl, deleteTaskIfBackendImpl, archiveTaskBackendImpl, unarchiveTaskImpl, restoreFromArchiveImpl, listArchivedTasksImpl } from "./task-store/archive-lifecycle-2.js";
@@ -1647,14 +1646,14 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
if (!this.asyncLayer) return { reconciled: 0 };
return reconcileSoftDeletedColumnDriftAsync(this.asyncLayer, recordAudit);
}
async getWorkflowParitySummary(options: { since?: string; limit?: number } = {}): Promise<WorkflowParitySummary> {
return getWorkflowParitySummaryImpl(this, options);
}
/** Aggregate the `workflowColumns` flag default-flip criteria (U12, KTD-8) into */
async computeWorkflowColumnsGraduationReport( options: { since?: string; limit?: number } = {}, ): Promise<WorkflowColumnsGraduationReport> {
return computeWorkflowColumnsGraduationReportImpl(this, options);
}
/*
FNXC:WorkflowColumns 2026-07-27-10:05 (U2 / R9):
`getWorkflowParitySummary` and `computeWorkflowColumnsGraduationReport` are
DELETED with `workflow-parity.ts`. They aggregated the pre-cutover dual-observe
contract, whose emitter (`workflow-parity-observer.ts`) is already a tombstone —
so both always returned an empty report over run-audit rows nothing writes, and
neither had a caller outside this class.
*/
/**
* FNXC:RuntimeLifecycleAsync 2026-06-24-11:10:

42
packages/core/src/task-store/moves.ts
View File

@@ -13,6 +13,7 @@ import {eq, sql} from "drizzle-orm";
import type {Task, Column, ColumnId, HandoffToReviewOptions} from "../types.js";
import {VALID_TRANSITIONS, COLUMNS} from "../types.js";
import {serializeWorkflowIr} from "../workflow-ir.js";
import {emitWorkflowLifecycleEvent} from "../workflow-events.js";
import {resolveAllowedColumns, workflowHasColumn} from "../workflow-transitions.js";
import {isBuiltinWorkflowId, getBuiltinWorkflow, resolveDefaultWorkflowIr, DEFAULT_WORKFLOW_ID} from "../builtin-workflows.js";
import {parseWorkflowIr} from "../workflow-ir.js";
@@ -1212,6 +1213,47 @@ export async function moveTaskInternalImpl(store: TaskStore, id: string, toColum
if (fromColumn !== toColumn) {
store.emit("task:moved", { task, from: fromColumn, to: toColumn, source: moveSource });
/*
FNXC:WorkflowEvents 2026-07-27-11:45 (U3 / R5, R6):
THE post-commit emit point for lifecycle transitions. Its position is the
contract, not a detail: everything above has committed (the row upsert,
the capacity reservation, the in-transaction outbox writes), so an emitted
`TaskTransitioned` implies a durable transition and a rolled-back
transaction emits nothing at all — the transaction throws out of the
`layer.transactionImmediate` block long before reaching here.
It sits beside the existing `task:moved` store event rather than replacing
it: `task:moved` carries the whole task object to in-process listeners
(triage's column-wake handler), which the ids-only rule forbids on the
bus. U7 migrates that listener onto the bus as a subscriber; until then
the two coexist and neither is authoritative — a lifecycle decision reads
the task row, never an event.
Fire-and-forget by construction: `emit` never throws and never awaits
subscribers, so no subscriber can slow, fail, or reorder a transition.
*/
emitWorkflowLifecycleEvent({
type: "TaskTransitioned",
taskId: id,
at: movedAt,
from: fromColumn,
to: toColumn,
moveSource,
...(internal.runContext?.runId ? { runId: internal.runContext.runId } : {}),
/*
FNXC:WorkflowEvents 2026-07-27-15:10 (U3, PR #2467 review):
OMIT rather than guess. `effectiveWorkflowIdForMove` reads the task's real
selection only when `useWorkflow` is true; otherwise it is hardcoded to
`builtin:coding`. That compat flag is off for effectively every real
project (see the note at the flag-OFF adjacency branch above), so
emitting it unconditionally would stamp `builtin:coding` onto moves of
tasks on a custom workflow — a wrong value baked into a brand-new wire
field, latent only because no subscriber reads it yet. An absent
`workflowId` means "not resolved here"; a subscriber that needs it reads
the selection itself.
*/
...(useWorkflow ? { workflowId: effectiveWorkflowIdForMove } : {}),
});
}
if (toColumn === "done") {
await store.clearNearDuplicateReferencesToFailSoft(id, {

45
packages/core/src/task-store/project-store-ops.ts
View File

@@ -21,7 +21,6 @@ import {getWorkflowExtensionRegistry} from "../workflow-extension-registry.js";
import type {WorkflowMovePolicyInput} from "../workflow-extension-types.js";
import "../builtin-traits.js";
import {normalizeWorkflowIcon, type WorkflowDefinition, type WorkflowDefinitionInput} from "../workflow-definition-types.js";
import {WORKFLOW_PARITY_OBSERVED_MUTATION, WORKFLOW_PARITY_DRIFT_MUTATION, type WorkflowParityDiff, type WorkflowParitySummary} from "../workflow-parity.js";
import {normalizeTaskPriority} from "../task-priority.js";
import type {AsyncDataLayer, DbTransaction} from "../postgres/data-layer.js";
import {recordRunAuditEventWithinTransaction} from "../postgres/data-layer.js";
@@ -503,50 +502,6 @@ export function getRunAuditEventsImpl(store: TaskStore, options: RunAuditEventFi
return rows.map((row) => store.rowToRunAuditEvent(row));
}
export async function getWorkflowParitySummaryImpl(store: TaskStore, options: { since?: string; limit?: number } = {}): Promise<WorkflowParitySummary> {
const limit = options.limit ?? 1000;
const observed = await store.getRunAuditEventsAsync({
domain: "database",
mutationType: WORKFLOW_PARITY_OBSERVED_MUTATION as unknown as RunAuditEvent["mutationType"],
startTime: options.since,
limit,
});
const driftEvents = await store.getRunAuditEventsAsync({
domain: "database",
mutationType: WORKFLOW_PARITY_DRIFT_MUTATION as unknown as RunAuditEvent["mutationType"],
startTime: options.since,
limit,
});
let agreed = 0;
for (const event of observed) {
if (event.metadata?.agree === true) agreed += 1;
}
const driftFieldCounts: Record<string, number> = {};
const recentDrift: WorkflowParitySummary["recentDrift"] = [];
for (const event of driftEvents) {
const diffs = Array.isArray(event.metadata?.diffs)
? (event.metadata.diffs as WorkflowParityDiff[])
: [];
for (const diff of diffs) {
driftFieldCounts[diff.field] = (driftFieldCounts[diff.field] ?? 0) + 1;
}
if (recentDrift.length < 20) {
recentDrift.push({ taskId: event.taskId ?? event.target, timestamp: event.timestamp, diffs });
}
}
return {
observed: observed.length,
agreed,
drift: driftEvents.length,
agreeRate: observed.length > 0 ? agreed / observed.length : 0,
driftFieldCounts,
recentDrift,
};
}
export function dequeueMergeQueueOnColumnExitImpl(store: TaskStore, taskId: string, previousColumn: ColumnId, nextColumn: ColumnId, now: string): void {
if (previousColumn !== "in-review" || nextColumn === "in-review") {
return;

27
packages/core/src/task-store/task-artifacts-ops.ts
View File

@@ -11,7 +11,6 @@
import { TaskStore } from "../store.js";
import { countAgentLogEntries, readAgentLogEntries } from "../agent-log-file-store.js";
import { BUILTIN_CODING_WORKFLOW_IR } from "../builtin-coding-workflow-ir.js";
import { toJsonNullable } from "../db.js";
import { DbTransaction, recordRunAuditEventWithinTransaction } from "../postgres/data-layer.js";
import { and, eq, inArray, isNull, ne } from "drizzle-orm";
@@ -25,9 +24,8 @@ import { enqueueMergeQueue as enqueueMergeQueueAsync, peekMergeQueue as peekMerg
import { clearCompletionHandoffMarker as clearCompletionHandoffMarkerAsync, getCompletionHandoffMarker as getCompletionHandoffMarkerAsync } from "./async-workflow-workitems.js";
import { extractEffectiveWriteScopeFromPrompt } from "../file-scope-classification.js";
import { ArtifactRow, WorkflowWorkItemRow } from "./row-types.js";
import { AgentLogEntry, Artifact, ArtifactCreateInput, Column, CompletionHandoffMarker, MergeQueueEnqueueOptions, MergeQueueEntry, PluginActivation, PluginActivationInput, RunAuditEvent, RunMutationContext, Task, TaskDocument, TaskDocumentRevision, WorkflowWorkItem, WorkflowWorkItemKind, isColumn } from "../types.js";
import { AgentLogEntry, Artifact, ArtifactCreateInput, Column, CompletionHandoffMarker, MergeQueueEnqueueOptions, MergeQueueEntry, PluginActivation, PluginActivationInput, RunMutationContext, Task, TaskDocument, TaskDocumentRevision, WorkflowWorkItem, WorkflowWorkItemKind, isColumn } from "../types.js";
import type { UsageEventInput } from "../usage-events.js";
import { DUAL_ACCEPT_PARITY_MUTATIONS, type WorkflowColumnsGraduationReport, computeWorkflowColumnsGraduationReport } from "../workflow-parity.js";
import { existsSync } from "node:fs";
import { mkdir, readFile, writeFile } from "node:fs/promises";
import { join } from "node:path";
@@ -79,29 +77,6 @@ export async function recordPluginActivationImpl(store: TaskStore, input: Plugin
return recordPluginActivationAsync(layer.db, input);
}
export async function computeWorkflowColumnsGraduationReportImpl(store: TaskStore,
options: { since?: string; limit?: number } = {},
): Promise<WorkflowColumnsGraduationReport> {
const limit = options.limit ?? 1000;
const parity = await store.getWorkflowParitySummary(options);
const dualAcceptEvents: RunAuditEvent[] = [];
for (const mutationType of DUAL_ACCEPT_PARITY_MUTATIONS) {
dualAcceptEvents.push(
...await store.getRunAuditEventsAsync({
domain: "database",
mutationType: mutationType as unknown as RunAuditEvent["mutationType"],
startTime: options.since,
limit,
}),
);
}
return computeWorkflowColumnsGraduationReport({
parity,
defaultWorkflowIr: BUILTIN_CODING_WORKFLOW_IR,
dualAcceptEvents,
});
}
export async function enqueueMergeQueueImpl(store: TaskStore, taskId: string, opts: MergeQueueEnqueueOptions = {}): Promise<MergeQueueEntry> {
/*
FNXC:SqliteDualPathCleanup 2026-07-26-14:05:

246
packages/core/src/types/workflow-events.ts Normal file
View File

@@ -0,0 +1,246 @@
/*
FNXC:WorkflowEvents 2026-07-27-11:00 (U3 / R5, R6 — workflow-owned lifecycle):
The lifecycle event vocabulary. One typed announcement per graph/lifecycle seam,
carrying IDS AND OUTCOMES ONLY — the same discipline run-audit metadata follows,
for the same reason: these payloads are persisted, logged, and forwarded to plugin
subscribers, so prose, prompt text, model ids, and object bodies must never enter
them.
WHY THESE ARE REACTIONS AND NOT THE TRANSITION (KTD-3, settled with the operator
over event-sourced lifecycle): the transition itself commits transactionally —
guards, capacity reservation, and the move together — and the event fires AFTER.
Making the bus authoritative would put capacity and move atomicity behind
eventual consistency, which is precisely the double-release and crash-stranding
class this program exists to remove.
THE THREE INVARIANTS every consumer may rely on:
1. No subscriber performs a lifecycle transition.
2. No subscriber is the only record of durable work — the transactional outbox
owns that (a work item written INSIDE the transition transaction).
3. A dropped event costs a REACTION (a notification, a board refresh, an
analytics row), never a state change or a unit of work.
Corollary for authors: because a dropped event is tolerable by construction,
emission is fire-and-forget and a throwing subscriber is isolated. If you find
yourself wanting delivery guarantees from this bus, the work belongs in the
outbox instead.
*/
/** The lifecycle seams that announce themselves. */
export type WorkflowLifecycleEventType =
| "TaskTransitioned"
| "NodeEntered"
| "NodeCompleted"
| "RunSuspended"
| "RunResumed";
/** Fields every lifecycle event carries. */
export interface WorkflowLifecycleEventBase {
type: WorkflowLifecycleEventType;
/** The task the seam belongs to. */
taskId: string;
/** ISO timestamp of the seam (the committed move time, not the emit time). */
at: string;
/** The graph run this seam belongs to, when one is in scope. */
runId?: string;
/** The workflow governing the task, when resolved. */
workflowId?: string;
}
/** A committed lifecycle column change. Emitted from the single post-commit
* point in `moveTaskInternalImpl`, so its existence implies durability. */
export interface TaskTransitionedEvent extends WorkflowLifecycleEventBase {
type: "TaskTransitioned";
from: string;
to: string;
/** The graph node whose entry caused the crossing, when graph-driven. */
nodeId?: string;
/** Who asked for the move ("user" / "engine" / …) — an enum-ish id, not prose. */
moveSource?: string;
}
/** Graph traversal entered a node. */
export interface NodeEnteredEvent extends WorkflowLifecycleEventBase {
type: "NodeEntered";
nodeId: string;
/** The node's declared column, absent for a columnless node (e.g. `end`). */
column?: string;
}
/** A node finished with a routing outcome ("success" / "failure" / …). */
export interface NodeCompletedEvent extends WorkflowLifecycleEventBase {
type: "NodeCompleted";
nodeId: string;
outcome: string;
}
/** A run parked at a seam it cannot cross yet (capacity, manual hold). */
export interface RunSuspendedEvent extends WorkflowLifecycleEventBase {
type: "RunSuspended";
nodeId: string;
/** Enum-ish suspension reason ("capacity", …) — never an error message. */
reason: string;
fromColumn?: string;
toColumn?: string;
}
/** A previously suspended run resumed. */
export interface RunResumedEvent extends WorkflowLifecycleEventBase {
type: "RunResumed";
nodeId: string;
/** Who released it ("scheduler", "operator", …). */
releasedBy?: string;
}
export type WorkflowLifecycleEvent =
| TaskTransitionedEvent
| NodeEnteredEvent
| NodeCompletedEvent
| RunSuspendedEvent
| RunResumedEvent;
/*
FNXC:WorkflowEvents 2026-07-27-11:05 (U3):
IDS-ONLY ENFORCEMENT IS A TEST, NOT A CONVENTION. run-audit's equivalent rule
lives only in prose and has been violated repeatedly (each violation caught in
review, if at all). These payloads reach plugin subscribers, so the rule is
mechanised here and asserted by `workflow-events.test.ts`.
The rule has TWO halves, and the second is the one that matters (PR #2467 review
— CodeRabbit, major):
a. VALUE shape. Every value is a scalar (string / number / boolean) or an array
of scalars; a string is at most MAX_ID_VALUE_LENGTH characters and contains
no newline. This catches a spread task row and a multi-line stack trace.
b. KEY allow-list, per event type. Value shape ALONE is not enough: a short
`error: "auth failed"`, a `prompt: "summarize"`, or a `modelId` is a
perfectly good scalar and would sail through. Since these payloads reach
plugin subscribers, an unknown key is refused outright — the declared
interfaces above are the whole permitted surface, so adding a field means
adding it here, deliberately, rather than discovering it in a log.
c. REQUIRED keys, per event type. The allow-list is a ceiling; this is the
floor. Without it a payload missing `taskId` validates clean and gets
delivered — and a subscriber keying derived state on `event.taskId` then
writes under `undefined` rather than failing, which is the quiet-corruption
mode this whole seam is supposed to be immune to. An absent required key is
a producer bug, so it is caught at the emit boundary rather than at each of
N subscribers.
An unknown TYPE is itself a violation: a caller inventing an event out of band
gets no implicit permission to invent its payload either.
*/
export const MAX_ID_VALUE_LENGTH = 200;
/** Keys every lifecycle event may carry. */
const COMMON_EVENT_KEYS = ["type", "taskId", "at", "runId", "workflowId"] as const;
/** Keys every lifecycle event MUST carry. `runId`/`workflowId` are deliberately
* absent — both are legitimately unresolvable at some emit sites, and U3's own
* `workflowId` fix omits rather than guesses. */
const COMMON_REQUIRED_EVENT_KEYS = ["type", "taskId", "at"] as const;
/** The per-type permitted key surface — the declared interfaces above, encoded.
* A key absent from its type's list is refused, not merely value-checked. */
const ALLOWED_EVENT_KEYS: Record<WorkflowLifecycleEventType, readonly string[]> = {
TaskTransitioned: [...COMMON_EVENT_KEYS, "from", "to", "nodeId", "moveSource"],
NodeEntered: [...COMMON_EVENT_KEYS, "nodeId", "column"],
NodeCompleted: [...COMMON_EVENT_KEYS, "nodeId", "outcome"],
RunSuspended: [...COMMON_EVENT_KEYS, "nodeId", "reason", "fromColumn", "toColumn"],
RunResumed: [...COMMON_EVENT_KEYS, "nodeId", "releasedBy"],
};
/*
The non-optional fields of each interface above. Kept as a sibling literal rather
than derived from ALLOWED_EVENT_KEYS because required-ness is exactly the part a
type cannot express at runtime — `column` on NodeEntered and `fromColumn`/
`toColumn` on RunSuspended are allowed but genuinely optional (a columnless node
has no column), so the two lists differ on purpose.
*/
const REQUIRED_EVENT_KEYS: Record<WorkflowLifecycleEventType, readonly string[]> = {
TaskTransitioned: [...COMMON_REQUIRED_EVENT_KEYS, "from", "to"],
NodeEntered: [...COMMON_REQUIRED_EVENT_KEYS, "nodeId"],
NodeCompleted: [...COMMON_REQUIRED_EVENT_KEYS, "nodeId", "outcome"],
RunSuspended: [...COMMON_REQUIRED_EVENT_KEYS, "nodeId", "reason"],
RunResumed: [...COMMON_REQUIRED_EVENT_KEYS, "nodeId"],
};
/** A single ids-only rule violation. `path` locates it for the failure message. */
export interface WorkflowEventShapeViolation {
path: string;
reason:
| "object-body"
| "prose-string"
| "unsupported-type"
| "unknown-key"
| "unknown-type"
| "missing-required-key";
}
function checkScalar(path: string, value: unknown, out: WorkflowEventShapeViolation[]): void {
if (value === undefined || value === null) return;
if (typeof value === "number" || typeof value === "boolean") return;
if (typeof value === "string") {
if (value.length > MAX_ID_VALUE_LENGTH || value.includes("\n")) {
out.push({ path, reason: "prose-string" });
}
return;
}
if (typeof value === "object") {
out.push({ path, reason: "object-body" });
return;
}
out.push({ path, reason: "unsupported-type" });
}
/**
* Report every way `event` violates the ids/outcomes-only rule. Empty array ⇒
* the payload is safe to emit, persist, and hand to a plugin subscriber.
*/
export function findWorkflowEventShapeViolations(event: unknown): WorkflowEventShapeViolation[] {
const violations: WorkflowEventShapeViolation[] = [];
if (event === null || typeof event !== "object" || Array.isArray(event)) {
return [{ path: "<root>", reason: "unsupported-type" }];
}
const record = event as Record<string, unknown>;
const allowed = ALLOWED_EVENT_KEYS[record.type as WorkflowLifecycleEventType];
if (!allowed) {
// An unrecognised type has no declared payload, so nothing about it can be
// validated — refuse it rather than fall through to value-shape checks that
// would wave through any scalar field it carries.
return [{ path: "type", reason: "unknown-type" }];
}
for (const [key, value] of Object.entries(record)) {
if (!allowed.includes(key)) {
// The half that actually protects subscribers: `error`, `prompt`, and
// `modelId` are all valid scalars and are all refused here.
violations.push({ path: key, reason: "unknown-key" });
continue;
}
if (Array.isArray(value)) {
value.forEach((entry, i) => checkScalar(`${key}[${i}]`, entry, violations));
continue;
}
checkScalar(key, value, violations);
}
/*
The FLOOR. `undefined` counts as missing, not present: the emitters build
payloads with conditional spreads, so a field that failed to resolve is absent
rather than explicitly undefined — but a caller writing `{ taskId: maybeId }`
produces the explicitly-undefined form, and both are the same bug. Reported
after the per-key pass so a payload that is both malformed and incomplete
surfaces every reason at once.
*/
for (const key of REQUIRED_EVENT_KEYS[record.type as WorkflowLifecycleEventType]) {
if (record[key] === undefined || record[key] === null) {
violations.push({ path: key, reason: "missing-required-key" });
}
}
return violations;
}
/** Convenience predicate over `findWorkflowEventShapeViolations`. */
export function isIdsOnlyWorkflowEvent(event: unknown): boolean {
return findWorkflowEventShapeViolations(event).length === 0;
}

13
packages/core/src/workflow-columns-settings.ts
View File

@@ -1,13 +0,0 @@
import type { Settings } from "./types.js";
/**
* Resolve whether workflow-defined columns are active for a settings snapshot.
*
* FNXC:WorkflowColumns 2026-06-22-18:00:
* Workflow columns graduated from the experimental runtime flag. Public runtime checks must treat stale persisted false values as enabled so engine scheduling and dashboard callers do not reactivate the retired legacy dispatcher.
*/
export function isWorkflowColumnsEnabled(
_settings: Pick<Settings, "experimentalFeatures"> | undefined,
): boolean {
return true;
}

169
packages/core/src/workflow-events.ts Normal file
View File

@@ -0,0 +1,169 @@
/*
FNXC:WorkflowEvents 2026-07-27-11:20 (U3 / R5, R6 — workflow-owned lifecycle):
THE post-commit lifecycle bus: one place transitions are announced, one registry
of subscribers. Later units move imperative cross-service calls behind it —
today `executor.ts` is 21k lines largely because it is the junction box every
lane routes its reactions through.
WHAT THIS BUS IS NOT. It is not a queue, not a transaction participant, and not
a delivery guarantee. Durable follow-on work uses the TRANSACTIONAL OUTBOX —
a `workflow_work_items` row written INSIDE the transition transaction (the shape
`createCompletionHandoffWorkflowWork` already uses). "Emit after commit, let a
subscriber enqueue the work" has a crash window: a process that dies between the
commit and the subscriber leaves no event AND no work-item row, so required work
is skipped permanently with nothing to recover from. Writing the item in the
transaction closes that window — a crash between commit and emit then costs at
most a notification.
Consequently EMISSION IS DELIBERATELY LOSSY AND ISOLATED:
- a throwing subscriber is caught, logged, and cannot roll back the
transition or stop the other subscribers;
- a rejected async subscriber is caught the same way;
- emission never blocks the caller's return, but events are DELIVERED IN
COMMIT ORDER (see the serial chain below).
ORDERING. Subscribers may be async, so a naive `for (…) void fn(e)` would let a
slow subscriber on transition #1 interleave with transition #2. Every emit is
appended to a single promise chain, so subscriber invocations for two transitions
on one task run in the order the transitions committed. That property is what
lets a subscriber maintain derived state (a board projection, a counter) without
its own sequencing.
TESTABILITY. `emit` is fire-and-forget; `drain()` awaits the current chain so a
test can assert on delivery without polling. Production code must not await
`drain()` on a lifecycle path — doing so would make a subscriber able to slow a
transition, which is the coupling the bus exists to remove.
*/
import { createLogger } from "./logger.js";
import {
findWorkflowEventShapeViolations,
type WorkflowLifecycleEvent,
} from "./types/workflow-events.js";
const eventLog = createLogger("workflow-events");
/** A post-commit reaction. Must not perform a lifecycle transition (R5). */
export type WorkflowEventSubscriber = (event: WorkflowLifecycleEvent) => void | Promise<void>;
export interface WorkflowEventSubscription {
/** Diagnostic name used in isolation warnings; defaults to "anonymous". */
name?: string;
}
export interface WorkflowEventBus {
/** Register a subscriber. Returns an unsubscribe function. */
subscribe(subscriber: WorkflowEventSubscriber, options?: WorkflowEventSubscription): () => void;
/** Announce a COMMITTED seam. Fire-and-forget; never throws. */
emit(event: WorkflowLifecycleEvent): void;
/** Await delivery of everything emitted so far. Test/shutdown seam only. */
drain(): Promise<void>;
/** Drop every subscriber. */
clear(): void;
subscriberCount(): number;
}
export function createWorkflowEventBus(): WorkflowEventBus {
const subscribers = new Map<symbol, { fn: WorkflowEventSubscriber; name: string }>();
// The serial delivery chain — see ORDERING above.
let chain: Promise<void> = Promise.resolve();
const deliver = async (event: WorkflowLifecycleEvent): Promise<void> => {
// Snapshot: a subscriber that unsubscribes (or one registered) mid-delivery
// must not mutate the iteration for the event already in flight.
for (const { fn, name } of [...subscribers.values()]) {
try {
await fn(event);
} catch (err) {
/*
FNXC:WorkflowEvents 2026-07-27-11:25 (U3 / R5):
ISOLATION. Logged and swallowed — never rethrown. A subscriber is a
reaction; letting one fail the emit would let a plugin's bug become a
lifecycle fault, and the transition it is reacting to has ALREADY
committed, so there is nothing left to roll back even if we wanted to.
*/
eventLog.warn("workflow event subscriber threw (isolated)", {
phase: "workflow-events:deliver",
subscriber: name,
eventType: event.type,
taskId: event.taskId,
error: err instanceof Error ? err.message : String(err),
});
}
}
};
return {
subscribe(subscriber, options) {
const key = Symbol("workflow-event-subscriber");
subscribers.set(key, { fn: subscriber, name: options?.name ?? "anonymous" });
return () => {
subscribers.delete(key);
};
},
emit(event) {
/*
FNXC:WorkflowEvents 2026-07-27-11:30 (U3):
The ids-only rule is enforced at the EMIT boundary, not at the subscriber,
so a violating payload never reaches a plugin or a log sink. It degrades
rather than throws: the emitter is on a post-commit path where throwing
would surface a shape bug as a lifecycle failure. The test suite asserts
the violation is DETECTED; production merely refuses to forward it.
*/
const violations = findWorkflowEventShapeViolations(event);
if (violations.length > 0) {
eventLog.warn("workflow event dropped — payload is not ids/outcomes-only", {
phase: "workflow-events:emit",
eventType: (event as { type?: string })?.type,
violations: violations.map((v) => `${v.path}:${v.reason}`),
});
return;
}
if (subscribers.size === 0) return;
chain = chain.then(() => deliver(event));
},
drain() {
return chain;
},
clear() {
subscribers.clear();
},
subscriberCount() {
return subscribers.size;
},
};
}
/*
FNXC:WorkflowEvents 2026-07-27-11:35 (U3):
Process-global default bus. The emitters (`moveTaskInternalImpl`, the graph's
column-boundary controller) are deep inside call paths with no place to thread a
bus handle, and the subscribers (engine lane services, plugins) register at
process start — the same shape as the existing `store.on/off` seam this bus
generalises. A per-store bus would need plumbing through every one of those call
sites for no isolation benefit: subscribers are already isolated from each other
and from the transition.
`resetWorkflowEventBusForTesting` exists so a suite cannot leak subscribers into
the next one; production must never call it.
*/
let globalBus: WorkflowEventBus | undefined;
export function getWorkflowEventBus(): WorkflowEventBus {
globalBus ??= createWorkflowEventBus();
return globalBus;
}
/** Emit onto the global bus. The single call the emitters use. */
export function emitWorkflowLifecycleEvent(event: WorkflowLifecycleEvent): void {
getWorkflowEventBus().emit(event);
}
/** @internal test-only — drop all subscribers and reset the delivery chain. */
export function resetWorkflowEventBusForTesting(): void {
globalBus = createWorkflowEventBus();
}

114
packages/core/src/workflow-lifecycle-traits.ts
View File

@@ -22,6 +22,7 @@ resolves to the same columns the old literals named):
import type { WorkflowIr, WorkflowIrColumn } from "./workflow-ir-types.js";
import type { TraitFlags } from "./trait-types.js";
import { getTraitRegistry } from "./trait-registry.js";
import { resolveWorkflowIrForTask, type WorkflowIrResolverStore } from "./workflow-ir-resolver.js";
/** The v2 column list, or [] for a v1/column-less IR. */
function columnsOf(ir: WorkflowIr): WorkflowIrColumn[] {
@@ -84,3 +85,116 @@ export function resolveReboundTarget(ir: WorkflowIr): string | undefined {
if (intake) return intake.id;
return columns[0].id;
}
/*
FNXC:WorkflowLifecycleColumns 2026-07-27-09:10 (U1 / KTD-2 — workflow-owned lifecycle):
THE lifecycle-column resolution seam. ~207 production sites decide the lifecycle by
comparing `task.column` against a hardcoded id ("todo", "in-progress", …). Those guards
do not FAIL when the column moves underneath them — they silently stop matching, which
disables a recovery path with a green suite. Phases B–D convert those sites onto the two
functions below, so conversion is mechanical rather than a per-site IR plumbing exercise.
Why a single struct rather than six separate lookups: most call sites need two or three
lifecycle columns at once (a sweep gated on the hold column that rebounds into it, a
release path comparing hold against wip). Resolving them together keeps one IR read and
one cache entry per workflow.
Trait → role mapping (the trait vocabulary is the source of truth, not these names):
intake → `intake` where new cards land
hold → `hold` passive dwell with a release condition (capacity)
wip → `countsTowardWip` occupies an implementation slot
review → `mergeOrchestration` the merge/PR orchestration lane
complete → `complete` terminal success
archived → `archived` globally archived
CONSERVATIVE-ON-UNRESOLVABLE (deliberate): a v1 / column-less IR resolves to `undefined`
for the WHOLE struct, not to a struct of undefined roles. The distinction matters — a
caller must be able to tell "this workflow declares no hold column" (hold: undefined,
struct present) apart from "this workflow has no column vocabulary at all" (undefined).
The first is a real workflow shape to honor; the second means the caller has no basis to
decide and must skip-and-log rather than guess a legacy literal.
*/
export interface LifecycleColumns {
/** Where new cards land. */
intake: string | undefined;
/** Passive dwell column with a release condition (capacity hold). */
hold: string | undefined;
/** Occupies an implementation/WIP slot. */
wip: string | undefined;
/** The merge/PR orchestration lane. */
review: string | undefined;
/** Terminal-success column. */
complete: string | undefined;
/** Globally archived column. */
archived: string | undefined;
}
/** The trait carrying each lifecycle role. Declared once so the roles and the
* trait vocabulary cannot drift apart silently. */
const LIFECYCLE_ROLE_FLAGS: Record<keyof LifecycleColumns, keyof TraitFlags> = {
intake: "intake",
hold: "hold",
wip: "countsTowardWip",
review: "mergeOrchestration",
complete: "complete",
archived: "archived",
};
/**
* Resolve an IR's lifecycle columns by trait — the FIRST column carrying each
* trait, in declared column order. A role no column carries is `undefined`
* (never substituted from an unrelated column).
*
* Returns `undefined` for a v1 / column-less IR: there is no column vocabulary
* to resolve, so the caller has no workflow-derived answer to act on.
*/
export function resolveLifecycleColumns(ir: WorkflowIr): LifecycleColumns | undefined {
const columns = columnsOf(ir);
if (columns.length === 0) return undefined;
const registry = getTraitRegistry();
/*
FNXC:WorkflowLifecycleColumns 2026-07-27-15:40 (U1, PR #2467 review):
Resolve each column's flags ONCE. A per-role `columns.find(...)` re-resolved
every column's traits per role — up to 6N resolutions — and this function is
not memoized, so a Phase B sweep sharing an IR cache across 400 cards would
still pay it per card (the cache holds the IR, not the resolved struct).
*/
const resolved = columns.map((c) => ({ id: c.id, flags: registry.resolveColumnFlags(c) }));
const first = (flag: keyof TraitFlags): string | undefined =>
resolved.find((c) => c.flags[flag] === true)?.id;
return {
intake: first(LIFECYCLE_ROLE_FLAGS.intake),
hold: first(LIFECYCLE_ROLE_FLAGS.hold),
wip: first(LIFECYCLE_ROLE_FLAGS.wip),
review: first(LIFECYCLE_ROLE_FLAGS.review),
complete: first(LIFECYCLE_ROLE_FLAGS.complete),
archived: first(LIFECYCLE_ROLE_FLAGS.archived),
};
}
/**
* Store-aware form: resolve a TASK's lifecycle columns through its workflow
* selection.
*
* `cache` is CALLER-OWNED on purpose. A self-healing pass over 400 cards spanning
* three workflows must read three IRs, not 400 — the caller allocates one map per
* sweep and hands it to every resolution in that pass (the shape the periodic
* sweep's existing `irCache` already uses). A module-level cache would instead
* have to guess when a mid-flight workflow edit invalidates it.
*
* Returns `undefined` when the workflow cannot be resolved to a column
* vocabulary — callers keep conservative behavior (skip and log) rather than
* falling back to a legacy literal.
*/
export async function resolveTaskLifecycleColumns(
store: WorkflowIrResolverStore,
taskId: string,
cache?: Map<string, WorkflowIr>,
): Promise<LifecycleColumns | undefined> {
try {
const ir = await resolveWorkflowIrForTask(store, taskId, cache);
return resolveLifecycleColumns(ir);
} catch {
return undefined;
}
}

485
packages/core/src/workflow-parity.ts
View File

@@ -1,485 +0,0 @@
import type { Column, RunAuditEvent } from "./types.js";
import { VALID_TRANSITIONS } from "./types.js";
import type { WorkflowIr } from "./workflow-ir-types.js";
import { resolveAllowedColumns, workflowHasColumn } from "./workflow-transitions.js";
export const WORKFLOW_PARITY_OBSERVED_MUTATION = "workflow:parity-observed" as const;
export const WORKFLOW_PARITY_DRIFT_MUTATION = "workflow:parity-drift" as const;
export type WorkflowStage = "triage" | "execute" | "review" | "merge";
export type WorkflowParityDiffCategory = "lifecycle" | "invariant" | "audit";
export type WorkflowParityDiffSeverity = "info" | "warning" | "error";
export interface WorkflowReliabilityInvariantSignals {
fileScopeGuardOutcome: string | null;
squashMergeContractOutcome: string | null;
autoMergeTerminalUntilMergedRespected: boolean;
moveTaskHardCancelRespected: boolean;
}
/**
* Observe-only workflow snapshot used for parity checks.
* Legacy remains authoritative; interpreter observations are advisory diagnostics only.
*/
export interface WorkflowRunObservation {
stageTransitions: WorkflowStage[];
terminalColumn: string | null;
terminalStatus: string | null;
reviewVerdict: string | null;
mergeOutcome: string | null;
invariants: WorkflowReliabilityInvariantSignals;
}
export interface WorkflowParityDiff {
field: string;
legacy: unknown;
interpreter: unknown;
category: WorkflowParityDiffCategory;
severity: WorkflowParityDiffSeverity;
}
export interface WorkflowParityDriftReport {
agree: boolean;
diffs: WorkflowParityDiff[];
}
function isEqualScalarArray(left: readonly string[], right: readonly string[]): boolean {
if (left.length !== right.length) return false;
return left.every((value, index) => value === right[index]);
}
function pushDiff(
diffs: WorkflowParityDiff[],
field: string,
legacy: unknown,
interpreter: unknown,
category: WorkflowParityDiffCategory,
severity: WorkflowParityDiffSeverity = "warning",
): void {
diffs.push({ field, legacy, interpreter, category, severity });
}
/**
* Pure observation comparison contract for dual-observe shadow checks.
* Legacy observation is authoritative; interpreter drift is diagnostics only.
*/
export function compareWorkflowRunObservations(
legacy: WorkflowRunObservation,
interpreter: WorkflowRunObservation,
): WorkflowParityDriftReport {
const diffs: WorkflowParityDiff[] = [];
if (!isEqualScalarArray(legacy.stageTransitions, interpreter.stageTransitions)) {
pushDiff(
diffs,
"stageTransitions",
legacy.stageTransitions,
interpreter.stageTransitions,
"lifecycle",
"error",
);
}
const lifecycleChecks: Array<[field: string, legacyValue: unknown, interpreterValue: unknown]> = [
["terminalColumn", legacy.terminalColumn, interpreter.terminalColumn],
["terminalStatus", legacy.terminalStatus, interpreter.terminalStatus],
["reviewVerdict", legacy.reviewVerdict, interpreter.reviewVerdict],
["mergeOutcome", legacy.mergeOutcome, interpreter.mergeOutcome],
];
for (const [field, legacyValue, interpreterValue] of lifecycleChecks) {
if (legacyValue !== interpreterValue) {
pushDiff(diffs, field, legacyValue, interpreterValue, "lifecycle", "error");
}
}
const invariantChecks: Array<[field: string, legacyValue: unknown, interpreterValue: unknown]> = [
[
"invariants.fileScopeGuardOutcome",
legacy.invariants.fileScopeGuardOutcome,
interpreter.invariants.fileScopeGuardOutcome,
],
[
"invariants.squashMergeContractOutcome",
legacy.invariants.squashMergeContractOutcome,
interpreter.invariants.squashMergeContractOutcome,
],
[
"invariants.autoMergeTerminalUntilMergedRespected",
legacy.invariants.autoMergeTerminalUntilMergedRespected,
interpreter.invariants.autoMergeTerminalUntilMergedRespected,
],
[
"invariants.moveTaskHardCancelRespected",
legacy.invariants.moveTaskHardCancelRespected,
interpreter.invariants.moveTaskHardCancelRespected,
],
];
for (const [field, legacyValue, interpreterValue] of invariantChecks) {
if (legacyValue !== interpreterValue) {
pushDiff(diffs, field, legacyValue, interpreterValue, "invariant", "error");
}
}
return {
agree: diffs.length === 0,
diffs,
};
}
export const WORKFLOW_COMPARABLE_AUDIT_MUTATIONS = [
"task:move",
"task:update",
"task:pause",
"task:unpause",
"task:dependency:add",
"merge:request-enqueued",
"merge:dependency-parity-diff",
"merge:lease-parity-diff",
] as const;
const WORKFLOW_COMPARABLE_AUDIT_MUTATION_SET = new Set<string>(WORKFLOW_COMPARABLE_AUDIT_MUTATIONS);
export interface WorkflowAuditObservation {
mutationType: string;
target: string;
phase: string | null;
}
export function extractWorkflowAuditObservations(events: readonly RunAuditEvent[]): WorkflowAuditObservation[] {
return events
.filter(
(event) =>
event.domain === "database"
&& WORKFLOW_COMPARABLE_AUDIT_MUTATION_SET.has(String(event.mutationType)),
)
.map((event) => ({
mutationType: String(event.mutationType),
target: event.target,
phase: typeof event.metadata?.phase === "string" ? event.metadata.phase : null,
}));
}
export function compareWorkflowRunAudits(
legacyEvents: readonly RunAuditEvent[],
interpreterEvents: readonly RunAuditEvent[],
): WorkflowParityDriftReport {
const legacy = extractWorkflowAuditObservations(legacyEvents);
const interpreter = extractWorkflowAuditObservations(interpreterEvents);
const diffs: WorkflowParityDiff[] = [];
if (legacy.length !== interpreter.length) {
pushDiff(diffs, "audit.length", legacy.length, interpreter.length, "audit");
}
const count = Math.max(legacy.length, interpreter.length);
for (let index = 0; index < count; index += 1) {
const left = legacy[index];
const right = interpreter[index];
if (!left || !right) {
pushDiff(diffs, `audit[${index}]`, left ?? null, right ?? null, "audit");
continue;
}
if (left.mutationType !== right.mutationType) {
pushDiff(
diffs,
`audit[${index}].mutationType`,
left.mutationType,
right.mutationType,
"audit",
);
}
if (left.target !== right.target) {
pushDiff(diffs, `audit[${index}].target`, left.target, right.target, "audit");
}
if (left.phase !== right.phase) {
pushDiff(diffs, `audit[${index}].phase`, left.phase, right.phase, "audit");
}
}
return {
agree: diffs.length === 0,
diffs,
};
}
// ── Observation builders (CU-U5) ─────────────────────────────────────────────
// Construct a WorkflowRunObservation from real run data so the legacy and
// interpreter sides can be compared without either side hand-rolling the shape.
/** Conservative defaults: a run that didn't signal an invariant is assumed to
* have respected the terminal/cancel contracts (the common, non-drift case). */
export const DEFAULT_WORKFLOW_INVARIANTS: WorkflowReliabilityInvariantSignals = {
fileScopeGuardOutcome: null,
squashMergeContractOutcome: null,
autoMergeTerminalUntilMergedRespected: true,
moveTaskHardCancelRespected: true,
};
const COLUMN_TO_STAGE: Record<string, WorkflowStage | undefined> = {
triage: "triage",
todo: "triage",
"in-progress": "execute",
"in-review": "review",
done: "merge",
};
/**
* Map the ordered list of columns a run passed through to workflow stages,
* collapsing consecutive repeats. This is how the legacy side derives its
* stageTransitions — from the real task-move history rather than a guess.
*/
export function deriveStageTransitions(columnSequence: readonly string[]): WorkflowStage[] {
const stages: WorkflowStage[] = [];
for (const column of columnSequence) {
const stage = COLUMN_TO_STAGE[column];
if (stage && stages[stages.length - 1] !== stage) stages.push(stage);
}
return stages;
}
export interface WorkflowObservationTaskInput {
column: string;
status?: string | null;
review?: { verdict?: string } | null;
mergeDetails?: { outcome?: string } | null;
}
export interface WorkflowObservationBuildOptions {
/** Explicit stage sequence (wins over columnSequence). */
stageTransitions?: readonly WorkflowStage[];
/** Ordered columns the run passed through; mapped to stages when stageTransitions is absent. */
columnSequence?: readonly string[];
invariants?: Partial<WorkflowReliabilityInvariantSignals>;
}
/**
* Build a parity observation from a task's terminal persisted state (the legacy
* authoritative side). stageTransitions come from the caller's recorded column
* history when available, else fall back to the terminal column alone.
*/
export function buildWorkflowObservationFromTask(
task: WorkflowObservationTaskInput,
options?: WorkflowObservationBuildOptions,
): WorkflowRunObservation {
const stageTransitions = options?.stageTransitions
? [...options.stageTransitions]
: deriveStageTransitions(options?.columnSequence ?? [task.column]);
return {
stageTransitions,
terminalColumn: task.column ?? null,
terminalStatus: task.status ?? null,
reviewVerdict: task.review?.verdict ?? null,
mergeOutcome: task.mergeDetails?.outcome ?? (task.column === "done" ? "merged" : null),
invariants: { ...DEFAULT_WORKFLOW_INVARIANTS, ...options?.invariants },
};
}
/** Aggregate of dual-observe parity audit events — the graduation signal. */
export interface WorkflowParitySummary {
/** Total `workflow:parity-observed` events in scope. */
observed: number;
/** Of those, how many reported agree=true. */
agreed: number;
/** Total `workflow:parity-drift` events in scope. */
drift: number;
/** agreed / observed in [0,1]; 0 when nothing observed yet. */
agreeRate: number;
/** Count of drift occurrences per observation field, most-divergent first. */
driftFieldCounts: Record<string, number>;
/** Most recent drift events (capped) for inspection. */
recentDrift: Array<{ taskId: string; timestamp: string; diffs: WorkflowParityDiff[] }>;
}
export interface WorkflowObservationParts {
stageTransitions: readonly WorkflowStage[];
terminalColumn?: string | null;
terminalStatus?: string | null;
reviewVerdict?: string | null;
mergeOutcome?: string | null;
invariants?: Partial<WorkflowReliabilityInvariantSignals>;
}
/**
* Build a parity observation from explicit parts (the interpreter/shadow side
* assembles these from its graph-walk result).
*/
export function buildWorkflowObservation(parts: WorkflowObservationParts): WorkflowRunObservation {
return {
stageTransitions: [...parts.stageTransitions],
terminalColumn: parts.terminalColumn ?? null,
terminalStatus: parts.terminalStatus ?? null,
reviewVerdict: parts.reviewVerdict ?? null,
mergeOutcome: parts.mergeOutcome ?? null,
invariants: { ...DEFAULT_WORKFLOW_INVARIANTS, ...parts.invariants },
};
}
// ── Transition parity (U12) ──────────────────────────────────────────────────
//
// The transition-parity suite (U4) proves, as a unit test, that the default
// workflow's resolved column adjacency equals the legacy VALID_TRANSITIONS
// graph. U12 surfaces the SAME comparison as a runtime check so the graduation
// gate can re-evaluate it against whatever IR is actually resolved for the
// default workflow in the field (not just the static fixture), catching a
// deliberately or accidentally drifted default-workflow adjacency.
/** One adjacency disagreement between the legacy graph and the resolved IR. */
export interface TransitionParityDiff {
/** The `from` column whose allowed-set diverged. */
from: string;
/** Allowed targets per the legacy VALID_TRANSITIONS graph. */
legacyAllowed: string[];
/** Allowed targets per the resolved workflow IR column graph. */
resolvedAllowed: string[];
}
export interface TransitionParityReport {
/** True when every legacy column's allowed-set matches the resolved IR's. */
agree: boolean;
/** Per-column adjacency disagreements (empty when `agree`). */
diffs: TransitionParityDiff[];
}
const LEGACY_COLUMNS = Object.keys(VALID_TRANSITIONS) as Column[];
function sortedUnique(values: readonly string[]): string[] {
return [...new Set(values)].sort();
}
/**
* Compare the default-workflow IR's resolved column adjacency against the legacy
* VALID_TRANSITIONS graph (R12 transition parity, machine-checked). For every
* legacy column, the resolved allowed-set must equal the legacy allowed-set
* exactly (allowed AND rejected). The IR must also recognize every legacy
* column. Any divergence is a graduation blocker.
*/
export function checkTransitionParity(ir: WorkflowIr): TransitionParityReport {
const diffs: TransitionParityDiff[] = [];
for (const from of LEGACY_COLUMNS) {
const legacyAllowed = sortedUnique(VALID_TRANSITIONS[from]);
// A column the resolved IR doesn't even define diverges by construction.
const resolvedAllowed = workflowHasColumn(ir, from)
? sortedUnique(resolveAllowedColumns(ir, from))
: [];
const equal =
legacyAllowed.length === resolvedAllowed.length &&
legacyAllowed.every((value, index) => value === resolvedAllowed[index]);
if (!equal) diffs.push({ from, legacyAllowed, resolvedAllowed });
}
return { agree: diffs.length === 0, diffs };
}
// ── Dual-accept disagreement counter (U12) ───────────────────────────────────
//
// U6 logs `merge:dependency-parity-diff` audits whenever the explicit handoff
// marker and the complete-flag column disagree during the FN-5719 dual-accept
// window. The window CLOSES at graduation, so any disagreement above zero over
// the observation period blocks the flip. This surfaces the count (and the
// lease-parity counterpart) from the audit trail as a graduation signal.
export const DUAL_ACCEPT_PARITY_MUTATIONS = [
"merge:dependency-parity-diff",
"merge:lease-parity-diff",
] as const;
const DUAL_ACCEPT_PARITY_MUTATION_SET = new Set<string>(DUAL_ACCEPT_PARITY_MUTATIONS);
export interface DualAcceptDisagreementReport {
/** Total dual-accept disagreement audit events in scope. */
total: number;
/** Count per mutation type (dependency vs lease parity diff). */
byMutationType: Record<string, number>;
}
/**
* Count the dual-accept marker/column disagreement audits (U6) in scope. Pure
* over the supplied events so the store can feed it whatever audit window the
* graduation report observes.
*/
export function countDualAcceptDisagreements(
events: readonly RunAuditEvent[],
): DualAcceptDisagreementReport {
const byMutationType: Record<string, number> = {};
let total = 0;
for (const event of events) {
const type = String(event.mutationType);
if (event.domain !== "database" || !DUAL_ACCEPT_PARITY_MUTATION_SET.has(type)) continue;
byMutationType[type] = (byMutationType[type] ?? 0) + 1;
total += 1;
}
return { total, byMutationType };
}
// ── Graduation report (U12) ──────────────────────────────────────────────────
//
// The flag default-flip criteria, aggregated into one report (KTD-8). The flip
// is a FIELD decision — this report is the GATE, not the trigger. `ready` is
// true only when ALL of:
// - the five-invariant dual-observe parity shows zero drift (drift === 0) over
// a non-empty observation window;
// - the default workflow's transition parity holds (no adjacency drift);
// - zero dual-accept marker/column disagreements over the window.
export interface WorkflowColumnsGraduationReport {
/** Five-invariant dual-observe parity (from the audit trail). */
parity: WorkflowParitySummary;
/** Default-workflow transition-graph parity vs VALID_TRANSITIONS. */
transitionParity: TransitionParityReport;
/** Dual-accept marker/column disagreement count (U6). */
dualAccept: DualAcceptDisagreementReport;
/** True only when every gate passes — the flag is eligible to default on. */
ready: boolean;
/** Human-readable blockers when not ready (empty when ready). */
blockers: string[];
}
export interface GraduationReportInputs {
/** Dual-observe parity summary (e.g. `store.getWorkflowParitySummary()`). */
parity: WorkflowParitySummary;
/** The resolved default-workflow IR to transition-parity-check. */
defaultWorkflowIr: WorkflowIr;
/** Audit events in the observation window for dual-accept counting. */
dualAcceptEvents: readonly RunAuditEvent[];
}
/**
* Aggregate the flag default-flip criteria into a single graduation report
* (U12, absorbing plan 002's M-D). Pure: the caller assembles the inputs from
* the store's audit trail and resolved default workflow, and decides whether to
* flip the flag — this function only computes the gate.
*/
export function computeWorkflowColumnsGraduationReport(
inputs: GraduationReportInputs,
): WorkflowColumnsGraduationReport {
const { parity, defaultWorkflowIr, dualAcceptEvents } = inputs;
const transitionParity = checkTransitionParity(defaultWorkflowIr);
const dualAccept = countDualAcceptDisagreements(dualAcceptEvents);
const blockers: string[] = [];
if (parity.observed === 0) {
blockers.push("no parity observations recorded yet (observation window empty)");
}
if (parity.drift > 0) {
blockers.push(`five-invariant parity drift observed (${parity.drift} drift events)`);
}
if (!transitionParity.agree) {
const cols = transitionParity.diffs.map((d) => d.from).join(", ");
blockers.push(`default-workflow transition parity drifted (columns: ${cols})`);
}
if (dualAccept.total > 0) {
blockers.push(`dual-accept marker/column disagreements above zero (${dualAccept.total})`);
}
return {
parity,
transitionParity,
dualAccept,
ready: blockers.length === 0,
blockers,
};
}

33
packages/dashboard/src/routes/board-workflows.ts
View File

@@ -22,7 +22,6 @@ import {
BUILTIN_CODING_WORKFLOW_IR,
getBuiltinWorkflow,
isBuiltinWorkflowId,
isWorkflowColumnsEnabled,
parseWorkflowIr,
resolveColumnFlags,
resolveWorkflowIrById,
@@ -157,9 +156,15 @@ async function describeWorkflow(
/**
* Build the board-workflows payload for the given task ids. Resolves each task's
* workflow selection (null → the default workflow lane) and assembles the
* deduplicated set of referenced workflow definitions. Returns
* `{ flagEnabled: false, ... }` (empty maps) when the flag is OFF so the route
* can return early and the client renders the legacy board.
* deduplicated set of referenced workflow definitions.
*
* FNXC:WorkflowColumns 2026-07-27-09:48 (U2 / R9):
* The flag-OFF early return is deleted — its gate (`isWorkflowColumnsEnabled`)
* returned a literal `true`, so the empty payload was unreachable. `flagEnabled`
* stays on the WIRE as a constant `true` because shipped dashboard clients still
* branch on it (Board, ListView, TaskDetailModal, useBoardWorkflows); removing
* the field would change the response shape, which this delete-only unit must
* not do. U10 retires the field once no client reads it.
*/
export async function buildBoardWorkflowsPayload(
store: Pick<TaskStore, "getWorkflowDefinition" | "getTaskWorkflowSelection" | "getSettings" | "listWorkflowDefinitions"> &
@@ -167,16 +172,16 @@ export async function buildBoardWorkflowsPayload(
taskIds: string[],
settingsOverride?: Pick<Settings, "experimentalFeatures">,
): Promise<BoardWorkflowsPayload> {
const settings = settingsOverride ?? (await store.getSettings());
const flagEnabled = isWorkflowColumnsEnabled(settings);
const empty: BoardWorkflowsPayload = {
flagEnabled,
defaultWorkflowId: DEFAULT_WORKFLOW_LANE_ID,
workflows: [],
taskWorkflowIds: {},
};
if (!flagEnabled) return empty;
/*
FNXC:WorkflowColumns 2026-07-27-09:50 (U2 / R9):
`settingsOverride` and the `store.getSettings()` read it defaulted to existed
ONLY to feed the deleted flag check — no other field of this payload depends on
settings. The parameter stays in the signature (callers pass it positionally
and it costs nothing) but is no longer read, so the settings round-trip is gone
from the board-load path.
*/
void settingsOverride;
const flagEnabled = true;
const taskWorkflowIds: Record<string, string> = {};
const referenced = new Set<string>();

21
packages/dashboard/src/routes/register-task-workflow-routes.ts
View File

@@ -53,7 +53,6 @@ import {
findNearDuplicates,
isEphemeralAgent,
parseExplicitDuplicateMarker,
isWorkflowColumnsEnabled,
resolveWorkflowIrForTask,
workflowHasColumn,
columnHasFlag,
@@ -1002,8 +1001,10 @@ export function registerTaskWorkflowRoutes(ctx: ApiRoutesContext, deps: TaskWork
// any of these tasks. One batched query (cheap; short-circuits when the
// table is empty). The payload is otherwise byte-identical.
try {
const settings = await scopedStore.getSettingsFast();
if (isWorkflowColumnsEnabled(settings) && tasks.length > 0) {
// FNXC:WorkflowColumns 2026-07-27-09:52 (U2 / R9): the
// `isWorkflowColumnsEnabled` conjunct is deleted (literal `true`), so
// branch-progress enrichment is gated only on there being tasks.
if (tasks.length > 0) {
const byTask = await scopedStore.getBranchProgressByTask(tasks.map((t) => t.id));
if (byTask.size > 0) {
tasks = tasks.map((task) => {
@@ -1108,11 +1109,11 @@ export function registerTaskWorkflowRoutes(ctx: ApiRoutesContext, deps: TaskWork
router.get("/tasks/board-workflows", async (req, res) => {
try {
const { store: scopedStore } = await getProjectContext(req);
// FNXC:WorkflowColumns 2026-07-27-09:53 (U2 / R9): the flag-OFF
// `{ flagEnabled: false }` short-circuit is deleted — unreachable behind a
// literal `true`. `buildBoardWorkflowsPayload` still emits `flagEnabled: true`
// for shipped clients that branch on it.
const settings = await scopedStore.getSettingsFast();
if (!isWorkflowColumnsEnabled(settings)) {
res.json({ flagEnabled: false, defaultWorkflowId: "builtin:coding", workflows: [], taskWorkflowIds: {} });
return;
}
// Resolve over the same (non-archived) board list the client renders.
const tasks = await scopedStore.listTasks({ slim: true, includeArchived: false });
const taskIds = tasks.map((t) => t.id);
@@ -1841,10 +1842,10 @@ export function registerTaskWorkflowRoutes(ctx: ApiRoutesContext, deps: TaskWork
router.post("/tasks/:id/promote", async (req, res) => {
try {
const { store: scopedStore } = await getProjectContext(req);
// FNXC:WorkflowColumns 2026-07-27-09:54 (U2 / R9): the
// "Workflow columns are not enabled" rejection is deleted — its gate was a
// literal `true`, so promote never took it.
const settings = await scopedStore.getSettingsFast();
if (!isWorkflowColumnsEnabled(settings)) {
throw badRequest("Workflow columns are not enabled");
}
const existing = await scopedStore.getTask(req.params.id);
const rootDir = scopedStore.getRootDir();
const allocateWorktree = existing

29
packages/engine/src/__tests__/legacy-tombstones.test.ts
View File

@@ -32,6 +32,19 @@ const DELETED_FILES = [
"packages/core/src/workflow-cutover.ts",
"packages/engine/src/workflow-authoritative-driver.ts",
"packages/engine/src/workflow-parity-observer.ts",
/*
FNXC:WorkflowColumns 2026-07-27-10:25 (U2 / R9 — workflow-owned lifecycle):
Two more pre-cutover modules. `workflow-columns-settings.ts` held
`isWorkflowColumnsEnabled`, a function whose body was `return true` — eight live
call sites still branched on it, so every flag-OFF arm was unreachable code that
read as a supported configuration. `workflow-parity.ts` asserted the default
workflow's adjacency EQUALS the legacy `VALID_TRANSITIONS`; U11 deliberately
breaks that equality by merging Todo into Planning, so re-introducing it would
re-encode the exact shape this program chose to break. Neither may come back:
the first as a fake kill switch, the second as a contract against the target state.
*/
"packages/core/src/workflow-columns-settings.ts",
"packages/core/src/workflow-parity.ts",
];
/**
@@ -71,6 +84,22 @@ const DELETED_SYMBOLS: Array<{ symbol: string; why: string }> = [
{ symbol: "maybeExecuteWorkflowGraph", why: "renamed executeWorkflowGraph and returns void — the graph cannot decline a task, so there is no 'maybe'" },
{ symbol: "transferPreHeldToLegacy", why: "re-registered the pre-held global concurrency slot for a legacy execute path that no longer exists" },
{ symbol: "workflow-selection-api-unavailable: store lacks a workflow-selection reader so the workflow graph cannot run ", why: "the OLD fail-closed reason, emitted only when the task had enabled steps; failing closed is now unconditional and carries a different reason" },
/*
FNXC:WorkflowColumns 2026-07-27-10:26 (U2 / R9):
The permanently-true workflow-columns flag and the legacy transition-parity
harness. `isWorkflowColumnsEnabled` is the dangerous one: it LOOKS like a
runtime toggle, so a future reader adding a new lifecycle branch would naturally
gate it on the flag and ship a dead arm. The parity symbols are dangerous the
other way — they assert an equality with the legacy six-column transition graph
that the target lifecycle shape must violate.
*/
{ symbol: "isWorkflowColumnsEnabled", why: "returned a literal `true`; it advertised a kill switch that did not exist and every flag-OFF arm behind it was dead" },
{ symbol: "checkTransitionParity", why: "asserted the default workflow's adjacency equals the legacy VALID_TRANSITIONS — an equality U11 deliberately breaks" },
{ symbol: "compareWorkflowRunObservations", why: "compared a graph run against a legacy run that no longer exists" },
{ symbol: "computeWorkflowColumnsGraduationReport", why: "graduation criteria for a flag flip that already happened" },
{ symbol: "WORKFLOW_PARITY_OBSERVED_MUTATION", why: "run-audit mutation for the deleted dual-observe chain; nothing emits it" },
{ symbol: "WORKFLOW_PARITY_DRIFT_MUTATION", why: "run-audit mutation for the deleted dual-observe chain; nothing emits it" },
{ symbol: "getWorkflowParitySummary", why: "aggregated parity run-audit rows that no emitter writes" },
];
/** Strip block and line comments so an explanatory tombstone note is not read as a live reference. */

100
packages/engine/src/__tests__/workflow-column-boundary-capacity.test.ts
View File

@@ -12,8 +12,14 @@ the next graph run retries the move, while a NON-capacity rejection (an invarian
propagates as a real error. Regression direction matters — a change that makes every rejection park
would silently swallow invariant violations, so both halves are asserted.
*/
import { describe, expect, it, vi } from "vitest";
import { TransitionRejectionError, type WorkflowIr } from "@fusion/core";
import { describe, expect, it, vi, beforeEach, afterEach } from "vitest";
import {
TransitionRejectionError,
getWorkflowEventBus,
resetWorkflowEventBusForTesting,
type WorkflowIr,
type WorkflowLifecycleEvent,
} from "@fusion/core";
import { createWorkflowColumnBoundary } from "../workflow-column-boundary.js";
/** Minimal two-column IR: a wip column and a review column, plus the remediation target. */
@@ -115,3 +121,93 @@ describe("workflow column boundary — capacity rejection on the remediation cro
expect(boundary.currentColumn()).toBe("in-progress");
});
});
/*
FNXC:WorkflowEvents 2026-07-27-17:40 (U3, PR #2467 review):
ANTI-"BORN DEAD" GUARD. The bus REFUSES a payload that violates the ids-only or
required-key rules, and refusal is silent by design (the emitter is post-commit;
throwing there would turn a shape bug into a lifecycle fault). That combination
means an emitter regression — a dropped `nodeId`, a renamed field, a stray
`error` — would stop the event firing with NO test failure anywhere, and every
subscriber built on it would quietly never run.
So the real emitters are asserted end-to-end through the real bus: not "was emit
called" (a spy would pass on a refused payload) but "did a subscriber actually
receive it". These tests fail if a future edit makes the boundary's payloads
invalid.
*/
describe("column boundary emits SURVIVE the bus's shape validation (U3)", () => {
beforeEach(() => resetWorkflowEventBusForTesting());
afterEach(() => getWorkflowEventBus().clear());
it("NodeEntered is DELIVERED for a column-bearing node", async () => {
const received: WorkflowLifecycleEvent[] = [];
getWorkflowEventBus().subscribe((event) => { received.push(event); }, { name: "probe" });
const boundary = createWorkflowColumnBoundary({
taskId: "FN-EV-1",
workflowId: "builtin:coding",
ir: ir(),
initialColumn: "in-review",
moveTask: async () => {},
});
await boundary.onNodeEntry(remediationNode());
await getWorkflowEventBus().drain();
const entered = received.filter((e) => e.type === "NodeEntered");
expect(entered).toHaveLength(1);
expect(entered[0]).toMatchObject({
type: "NodeEntered",
taskId: "FN-EV-1",
nodeId: "code-review-remediation",
column: "in-progress",
});
});
it("NodeEntered is DELIVERED for a COLUMNLESS node, with column omitted", async () => {
const received: WorkflowLifecycleEvent[] = [];
getWorkflowEventBus().subscribe((event) => { received.push(event); }, { name: "probe" });
const boundary = createWorkflowColumnBoundary({
taskId: "FN-EV-2",
workflowId: "builtin:coding",
ir: ir(),
initialColumn: "in-review",
});
// `end` carries no column — the case the optional `column` field exists for.
await boundary.onNodeEntry({ id: "end", kind: "end" } as never);
await getWorkflowEventBus().drain();
const entered = received.filter((e) => e.type === "NodeEntered");
expect(entered).toHaveLength(1);
expect(entered[0].taskId).toBe("FN-EV-2");
expect("column" in entered[0]).toBe(false);
});
it("RunSuspended is DELIVERED when a capacity rejection parks the crossing", async () => {
const received: WorkflowLifecycleEvent[] = [];
getWorkflowEventBus().subscribe((event) => { received.push(event); }, { name: "probe" });
const boundary = createWorkflowColumnBoundary({
taskId: "FN-EV-3",
workflowId: "builtin:coding",
ir: ir(),
initialColumn: "in-review",
moveTask: async () => { throw capacityError(); },
});
const result = await boundary.onNodeEntry(remediationNode());
await getWorkflowEventBus().drain();
expect(result).toMatchObject({ kind: "suspended", reason: "capacity" });
const suspended = received.filter((e) => e.type === "RunSuspended");
expect(suspended).toHaveLength(1);
expect(suspended[0]).toMatchObject({
type: "RunSuspended",
taskId: "FN-EV-3",
nodeId: "code-review-remediation",
reason: "capacity",
fromColumn: "in-review",
toColumn: "in-progress",
});
});
});

113
packages/engine/src/__tests__/workflow-event-subscribers.test.ts Normal file
View File

@@ -0,0 +1,113 @@
/*
FNXC:WorkflowEvents 2026-07-27-16:00 (U3 / R5, PR #2467 review):
The engine registration point's own invariants. Two of them are only
interesting because getting them wrong FAILS SILENTLY — a process ends up with
no reactions registered and nothing reports it:
IDEMPOTENT RE-REGISTRATION — an engine restart must replace its subscriber set,
not stack a second copy that double-delivers every event.
REGISTRATION-SCOPED CLEANUP — a cleanup handle captured before a restart must
not unsubscribe the set registered AFTER it.
Both are asserted through the real global bus, since that is the object the
production wiring shares.
*/
import { describe, expect, it, beforeEach, afterEach } from "vitest";
import {
getWorkflowEventBus,
resetWorkflowEventBusForTesting,
type WorkflowLifecycleEvent,
} from "@fusion/core";
import {
registerWorkflowEventSubscribers,
unregisterWorkflowEventSubscribers,
activeWorkflowEventSubscriberCount,
ENGINE_WORKFLOW_EVENT_SUBSCRIBERS,
type WorkflowEventSubscriberRegistration,
} from "../workflow-event-subscribers.js";
function transitioned(): WorkflowLifecycleEvent {
return {
type: "TaskTransitioned",
taskId: "FN-1",
at: "2026-07-27T00:00:00.000Z",
from: "todo",
to: "in-progress",
};
}
function recorder(name: string, types: WorkflowLifecycleEvent["type"][]): WorkflowEventSubscriberRegistration & { seen: string[] } {
const seen: string[] = [];
return { name, types, seen, handle: (event) => { seen.push(event.type); } };
}
beforeEach(() => {
resetWorkflowEventBusForTesting();
unregisterWorkflowEventSubscribers();
});
afterEach(() => {
unregisterWorkflowEventSubscribers();
});
describe("engine workflow-event subscriber registry (U3 / R5)", () => {
it("ships EMPTY — reactions arrive with the units that convert them", () => {
// Landing the seam pre-populated would convert reactions in the same commit
// that introduces the mechanism they rely on (U7/U8/U10 do that work).
expect(ENGINE_WORKFLOW_EVENT_SUBSCRIBERS).toEqual([]);
});
it("delivers only the event types a registration declares", async () => {
const transitions = recorder("transitions", ["TaskTransitioned"]);
const nodes = recorder("nodes", ["NodeEntered"]);
registerWorkflowEventSubscribers([transitions, nodes]);
getWorkflowEventBus().emit(transitioned());
await getWorkflowEventBus().drain();
expect(transitions.seen).toEqual(["TaskTransitioned"]);
expect(nodes.seen).toEqual([]);
});
it("RE-registering replaces the previous set rather than double-delivering", async () => {
const first = recorder("first", ["TaskTransitioned"]);
registerWorkflowEventSubscribers([first]);
const second = recorder("second", ["TaskTransitioned"]);
registerWorkflowEventSubscribers([second]);
expect(activeWorkflowEventSubscriberCount()).toBe(1);
getWorkflowEventBus().emit(transitioned());
await getWorkflowEventBus().drain();
expect(first.seen).toEqual([]);
expect(second.seen).toEqual(["TaskTransitioned"]);
});
it("a STALE cleanup handle cannot unsubscribe a later registration", async () => {
// The silent failure this guards: an engine restart re-registers, then a
// deferred cleanup captured before the restart fires and leaves the process
// with no reactions and no error.
const first = recorder("first", ["TaskTransitioned"]);
const staleCleanup = registerWorkflowEventSubscribers([first]);
const second = recorder("second", ["TaskTransitioned"]);
registerWorkflowEventSubscribers([second]);
staleCleanup();
expect(activeWorkflowEventSubscriberCount()).toBe(1);
getWorkflowEventBus().emit(transitioned());
await getWorkflowEventBus().drain();
expect(second.seen).toEqual(["TaskTransitioned"]);
});
it("the returned cleanup removes its OWN registration", async () => {
const only = recorder("only", ["TaskTransitioned"]);
const cleanup = registerWorkflowEventSubscribers([only]);
cleanup();
expect(activeWorkflowEventSubscriberCount()).toBe(0);
getWorkflowEventBus().emit(transitioned());
await getWorkflowEventBus().drain();
expect(only.seen).toEqual([]);
});
});

51
packages/engine/src/__tests__/workflow-graph-executor-parity.test.ts
View File

@@ -16,13 +16,8 @@
// suite `stepwise-workflow-parity.test.ts`. Keep the two concerns separate.
// ─────────────────────────────────────────────────────────────────────────────
import { describe, expect, it, vi } from "vitest";
import type { TaskDetail, WorkflowIrV2, WorkflowStage } from "@fusion/core";
import {
BUILTIN_CODING_WORKFLOW_IR,
buildWorkflowObservation,
buildWorkflowObservationFromTask,
compareWorkflowRunObservations,
} from "@fusion/core";
import type { TaskDetail, WorkflowIrV2 } from "@fusion/core";
import { BUILTIN_CODING_WORKFLOW_IR } from "@fusion/core";
import { WorkflowGraphExecutor } from "../workflow-graph-executor.js";
import type { WorkflowLegacySeams } from "../workflow-node-handlers.js";
@@ -167,9 +162,15 @@ describe("WorkflowGraphExecutor interpreter-parity", () => {
//
// The per-column agent feature must be invisible when no column carries a
// binding: the built-in default workflow synthesizes no `agent` field on any
// column, and a binding-free run produces observations identical to legacy via
// the same `compareWorkflowRunObservations` machinery the dual-observe gate uses.
// This is the byte-identity / parity oracle for the feature being unbound.
// column, and a binding-free run drives exactly the historical seam sequence.
//
// FNXC:WorkflowColumns 2026-07-27-10:15 (U2 / R9):
// The observation-comparison tail of the second case is DELETED with
// `workflow-parity.ts`. It built BOTH observations in this file and asserted they
// agreed, so it could only fail if `compareWorkflowRunObservations` itself were
// broken — it covered nothing about the executor. The load-bearing assertion,
// `stages` equalling the run-captured seam sequence, is unchanged and is what
// actually catches seam drift.
// ─────────────────────────────────────────────────────────────────────────────
describe("column-agent feature is invisible when unbound (U7 / R9)", () => {
it("the default built-in workflow synthesizes NO column agent field on any column", () => {
@@ -182,11 +183,10 @@ describe("column-agent feature is invisible when unbound (U7 / R9)", () => {
}
});
it("a binding-free run yields observations identical to legacy (compareWorkflowRunObservations agrees)", async () => {
// Drive the graph executor over the default execute→review→merge sequence and
// collect the stage transitions; with zero column bindings, the column-agent
// feature contributes nothing, so the interpreter observation must equal the
// legacy authoritative observation with no drift.
it("a binding-free run drives the historical seam sequence exactly", async () => {
// Drive the graph executor over the default planning→execute→review→merge
// sequence and collect the seam transitions; with zero column bindings the
// column-agent feature must contribute nothing to that sequence.
const stages: string[] = [];
const seams: WorkflowLegacySeams = {
planning: async () => ({ outcome: "success" }),
@@ -217,26 +217,5 @@ describe("column-agent feature is invisible when unbound (U7 / R9)", () => {
// observation below derives from the run-captured seam sequence, so seam
// drift fails here instead of being masked by a hard-coded literal.
expect(stages).toEqual(["planning", "execute", "review", "merge"]);
// Legacy authoritative observation: a clean run that lands in `done`/merged.
const legacyObs = buildWorkflowObservation({
stageTransitions: ["triage", "planning", "execute", "review", "merge"],
terminalColumn: "done",
terminalStatus: "done",
reviewVerdict: "approve",
mergeOutcome: "merged",
});
// Interpreter (binding-free) observation assembled from the same run.
const interpreterObs = buildWorkflowObservation({
stageTransitions: ["triage", ...stages] as WorkflowStage[],
terminalColumn: "done",
terminalStatus: "done",
reviewVerdict: "approve",
mergeOutcome: "merged",
});
const report = compareWorkflowRunObservations(legacyObs, interpreterObs);
expect(report.agree).toBe(true);
expect(report.diffs).toEqual([]);
});
});

20
packages/engine/src/merge-trait.ts
View File

@@ -18,7 +18,6 @@
*/
import {
isWorkflowColumnsEnabled,
resolveWorkflowIrForTask,
type DirectMergeCommitStrategy,
type Settings,
@@ -101,11 +100,16 @@ function settingsPolicy(settings: Pick<Settings, "directMergeCommitStrategy" | "
}
/**
* Resolve the effective merge policy for a task (R10). Flag ON: read the merge
* trait's config from the task's resolved workflow column; fall back to
* settings for any field the trait leaves unset (the built-in default
* workflow's merge trait carries no config, so it resolves entirely from
* settings — verbatim back-compat). Flag OFF: settings only.
* Resolve the effective merge policy for a task (R10). Read the merge trait's
* config from the task's resolved workflow column; fall back to settings for
* any field the trait leaves unset (the built-in default workflow's merge trait
* carries no config, so it resolves entirely from settings — verbatim
* back-compat).
*
* FNXC:WorkflowColumns 2026-07-27-09:44 (U2 / R9):
* The settings-only "flag OFF" arm is deleted. Its gate was
* `isWorkflowColumnsEnabled`, a literal `true`, so the arm was unreachable;
* the settings fallback below is the surviving path and always was.
*
* The lost-work guard trio is intentionally NOT represented here: no field this
* resolver returns can disable the sibling-branch rejection, line-anchored
@@ -119,10 +123,6 @@ export async function resolveMergePolicy(
const resolvedSettings = settings ?? (await store.getSettings());
const fallback = settingsPolicy(resolvedSettings);
if (!isWorkflowColumnsEnabled(resolvedSettings)) {
return fallback;
}
let config: Record<string, unknown> | undefined;
try {
const ir = await resolveWorkflowIrForTask(store, task.id);

21
packages/engine/src/scheduler.ts
View File

@@ -44,7 +44,7 @@ import { StaleTaskReporter } from "./stale-task-reporter.js";
import { BacklogPressureReporter } from "./backlog-pressure-reporter.js";
import { UnlinkedMissionsAdvisoryReporter } from "./unlinked-missions-advisory-reporter.js";
import { createRunAuditor, generateSyntheticRunId } from "./run-audit.js";
import { isWorkflowColumnsEnabled, DEFAULT_WORKFLOW_POOL_ID, resolveWorkflowIrForTask, resolveWorkflowIrById, resolveColumnFlags } from "@fusion/core";
import { DEFAULT_WORKFLOW_POOL_ID, resolveWorkflowIrForTask, resolveWorkflowIrById, resolveColumnFlags } from "@fusion/core";
import type { WorkflowIr, WorkflowIrV2 } from "@fusion/core";
import { runHoldReleaseSweep, isUnplannedForExecution, type SlotReservation } from "./hold-release.js";
import { moveTaskToReplanColumn } from "./replan-target.js";
@@ -1551,15 +1551,16 @@ export class Scheduler {
// U6 (KTD-10): report the default workflow's in-progress capacity as a
// per-column gate — the generalization of the legacy maxConcurrent gate
// (which reads through to the same value).
const perColumnGates = isWorkflowColumnsEnabled(settings)
? [{
workflowId: DEFAULT_WORKFLOW_POOL_ID,
columnId: "in-progress",
used: agentSlots + started,
limit: maxConcurrent,
slack: maxConcurrent - (agentSlots + started),
}]
: undefined;
// FNXC:WorkflowColumns 2026-07-27-09:42 (U2 / R9): the
// `isWorkflowColumnsEnabled` conditional is deleted — it returned a
// literal `true`, so the `undefined` arm never produced a diagnostic.
const perColumnGates = [{
workflowId: DEFAULT_WORKFLOW_POOL_ID,
columnId: "in-progress",
used: agentSlots + started,
limit: maxConcurrent,
slack: maxConcurrent - (agentSlots + started),
}];
return computeConcurrencyGateDiagnostic({
agentSlots,
maxConcurrent,

19
packages/engine/src/self-healing.ts
View File

@@ -31,7 +31,7 @@ import { existsSync, mkdirSync, readdirSync, readFileSync, realpathSync, rmSync,
import { readFile } from "node:fs/promises";
import { tmpdir } from "node:os";
import { isAbsolute, join, relative, resolve } from "node:path";
import { resolveColumnFlags, IN_REVIEW_STALL_DEADLOCK_LOG_PREFIX, IN_REVIEW_STALL_LOG_PREFIX, IN_REVIEW_STALL_TERMINAL_LOG_PREFIX, allowsAutoMergeProcessing, resolveEffectiveAutoMerge, countRecentIdenticalStallEntries, detectDependencyCycle, detectSelfDefeatingDependency, evaluateNoCommitsNoOpFinalize, evaluateCompletedPromotionFailureProvenance, evaluateSkipBypassTaint, getInReviewStalledSignal, getInReviewStallReason, getPrimaryPrInfo, getStalePausedReviewSignal, getStalePausedTodoSignal, getTaskHardMergeBlocker, getTaskMergeBlocker, isEphemeralAgent, isMergeRequestContractShadowEnabled, isWorkflowColumnsEnabled, isWorkspaceTask, isSharedBranchGroupMemberIntegration, isNearDuplicateCanonicalInactive, parseExplicitDuplicateMarker, flagTriageDuplicate, isTriageDuplicateKeepAcknowledged, resolveMaxAutoMergeRetries, resolveOptionalStepRevisionBudget, resolveOptionalReviewRevisionBudget, resolveWorkflowIrForTask, resolveReboundTarget, workflowHasColumn, planLegacyAdoption, resolveOrphanedPendingStepResults, classifyReviewLease, PLAN_REVIEW_LEASE_STALENESS_MS, DEFAULT_MAX_POST_REVIEW_FIXES, ACTIVE_WORKFLOW_WORK_ITEM_STATES, AWAITING_APPROVAL_PAUSE_REASON, type Agent, type AgentStore, type ChatStore, type MessageStore, type TaskStore, type Settings, type Task, type MergeDetails, type TaskPriority, type MergeResult, type WorkflowStepResult } from "@fusion/core";
import { resolveColumnFlags, IN_REVIEW_STALL_DEADLOCK_LOG_PREFIX, IN_REVIEW_STALL_LOG_PREFIX, IN_REVIEW_STALL_TERMINAL_LOG_PREFIX, allowsAutoMergeProcessing, resolveEffectiveAutoMerge, countRecentIdenticalStallEntries, detectDependencyCycle, detectSelfDefeatingDependency, evaluateNoCommitsNoOpFinalize, evaluateCompletedPromotionFailureProvenance, evaluateSkipBypassTaint, getInReviewStalledSignal, getInReviewStallReason, getPrimaryPrInfo, getStalePausedReviewSignal, getStalePausedTodoSignal, getTaskHardMergeBlocker, getTaskMergeBlocker, isEphemeralAgent, isMergeRequestContractShadowEnabled, isWorkspaceTask, isSharedBranchGroupMemberIntegration, isNearDuplicateCanonicalInactive, parseExplicitDuplicateMarker, flagTriageDuplicate, isTriageDuplicateKeepAcknowledged, resolveMaxAutoMergeRetries, resolveOptionalStepRevisionBudget, resolveOptionalReviewRevisionBudget, resolveWorkflowIrForTask, resolveReboundTarget, workflowHasColumn, planLegacyAdoption, resolveOrphanedPendingStepResults, classifyReviewLease, PLAN_REVIEW_LEASE_STALENESS_MS, DEFAULT_MAX_POST_REVIEW_FIXES, ACTIVE_WORKFLOW_WORK_ITEM_STATES, AWAITING_APPROVAL_PAUSE_REASON, type Agent, type AgentStore, type ChatStore, type MessageStore, type TaskStore, type Settings, type Task, type MergeDetails, type TaskPriority, type MergeResult, type WorkflowStepResult } from "@fusion/core";
import { finalizePlanningSegment } from "@fusion/core";
import type { MeshLeaseManager } from "./mesh-lease-manager.js";
import { createLogger, schedulerLog } from "./logger.js";
@@ -5423,15 +5423,18 @@ export class SelfHealingManager {
}
/**
* #1401: periodic transitionPending recovery sweep. Flag-ON only — when
* `workflowColumns` is OFF the legacy path never writes markers, so there is
* nothing to recover. Delegates to the store's idempotent recovery method
* (a no-op when no stale markers exist), keeping capacity counts honest after
* a crash between the in-txn marker write and the post-commit clear.
* #1401: periodic transitionPending recovery sweep. Delegates to the store's
* idempotent recovery method (a no-op when no stale markers exist), keeping
* capacity counts honest after a crash between the in-txn marker write and the
* post-commit clear.
*
* FNXC:WorkflowColumns 2026-07-27-09:40 (U2 / R9):
* The `isWorkflowColumnsEnabled` gate is GONE, not disabled. That helper
* returned a literal `true`, so the early return was unreachable and the
* settings read that fed it was pure cost. The sweep is unconditional now,
* which is what it already was at runtime.
*/
async runStaleTransitionPendingSweep(): Promise<void> {
const settings = await this.store.getSettings();
if (!isWorkflowColumnsEnabled(settings)) return;
await this.store.recoverStaleTransitionPending();
}

52
packages/engine/src/workflow-column-boundary.ts
View File

@@ -40,6 +40,7 @@ import {
isHoldToWipBoundary,
resolveColumnFlags,
TransitionRejectionError,
emitWorkflowLifecycleEvent,
} from "@fusion/core";
/** Run-audit event emitted by the boundary controller (KTD-12, ids/counts only). */
@@ -276,6 +277,29 @@ export function createWorkflowColumnBoundary(
async onNodeEntry(node: WorkflowIrNode): Promise<WorkflowColumnBoundaryEntryResult> {
const toColumn = node.column;
/*
FNXC:WorkflowEvents 2026-07-27-15:20 (U3 / R5, PR #2467 review):
Announce the NODE ENTRY, not the column crossing — so this fires BEFORE the
columnless short-circuit and before the same-column no-op below. Traversal
genuinely entered the node in all three cases, and a subscriber tracking
graph progress must see rework loops and terminal `end` arrivals, which a
crossing-only signal hides. `column` is omitted for a columnless node,
which is exactly why `NodeEnteredEvent.column` is optional.
The paired `TaskTransitioned` comes from the store's own post-commit point,
so a real crossing produces both and every other entry produces only this
one. Neither is authoritative for any lifecycle decision.
*/
emitWorkflowLifecycleEvent({
type: "NodeEntered",
taskId: deps.taskId,
at: new Date().toISOString(),
workflowId: deps.workflowId,
nodeId: node.id,
...(toColumn ? { column: toColumn } : {}),
});
// KTD-1: a columnless node (e.g. `end`) never moves the card.
if (!toColumn) return { kind: "entered" };
@@ -310,6 +334,23 @@ export function createWorkflowColumnBoundary(
irHash: computeWorkflowIrPin(deps.ir, node.id).irHash,
} as const;
await deps.onSuspend?.(suspension);
/*
FNXC:WorkflowEvents 2026-07-27-12:05 (U3 / R5):
Emitted AFTER `onSuspend` has persisted the durable continuation, so an
observed `RunSuspended` implies the continuation exists. The continuation
— not this event — is what the scheduler resumes from; dropping the event
costs a notification, never a stranded run.
*/
emitWorkflowLifecycleEvent({
type: "RunSuspended",
taskId: deps.taskId,
at: new Date().toISOString(),
workflowId: deps.workflowId,
nodeId: node.id,
reason: "capacity",
fromColumn,
toColumn,
});
return suspension;
}
@@ -348,6 +389,17 @@ export function createWorkflowColumnBoundary(
irHash: computeWorkflowIrPin(deps.ir, node.id).irHash,
} as const;
await deps.onSuspend?.(suspension);
// FNXC:WorkflowEvents 2026-07-27-12:06 (U3): see the hold→wip seam above.
emitWorkflowLifecycleEvent({
type: "RunSuspended",
taskId: deps.taskId,
at: new Date().toISOString(),
workflowId: deps.workflowId,
nodeId: node.id,
reason: "capacity",
fromColumn,
toColumn,
});
return suspension;
}
// A rejected move (invariant) leaves the card in its current column;

113
packages/engine/src/workflow-event-subscribers.ts Normal file
View File

@@ -0,0 +1,113 @@
/*
FNXC:WorkflowEvents 2026-07-27-12:20 (U3 / R5 — workflow-owned lifecycle):
The ENGINE side of the post-commit lifecycle bus: where lane services register
their reactions to a committed transition, and the boundary that later units move
imperative cross-service calls across.
The problem this exists to solve: today, when a task transitions, every service
that must react — notify, wake an agent, refresh a board, enqueue follow-on work —
is invoked DIRECTLY from the transition site. That is the main reason
`executor.ts` is 21k lines; it is the junction box every lane routes through.
Registering a reaction here instead means the transition site does not know its
reactors exist.
THE ADMISSION RULE for anything registered here — a handler that violates it is a
bug even if its tests pass:
1. It must not perform a lifecycle transition. The graph decides placement; a
subscriber that moves a card is a second authority, which is the failure
mode (FN-8504, the done-laundering incident) this program removes.
2. It must be safe to skip. Delivery is fire-and-forget and a process can die
between commit and emit. If skipping the handler loses a unit of work, the
work belongs in the TRANSACTIONAL OUTBOX — a `workflow_work_items` row
written inside the transition transaction — and this handler should at most
nudge the worker that drains it.
3. It must tolerate being called for a transition it does not care about, and
must not throw. Throws are caught and logged by the bus, but a handler that
routinely throws is a handler that routinely does nothing.
Registration is process-scoped and idempotent: `registerWorkflowEventSubscribers`
tears down anything it registered before re-registering, so an engine restart or
a test re-init cannot double-deliver.
STARTING EMPTY IS DELIBERATE. U3 lands the seam and its invariants; U7/U8/U10
move real reactions (triage's column wake, board refresh, agent wake) onto it,
each with the characterization tests that prove the reaction was non-authoritative
before it moved. Landing the seam pre-populated would mean converting reactions
in the same commit that introduces the mechanism they rely on.
*/
import {
getWorkflowEventBus,
type WorkflowEventSubscriber,
type WorkflowLifecycleEvent,
} from "@fusion/core";
/** A named engine-side reaction to a committed lifecycle seam. */
export interface WorkflowEventSubscriberRegistration {
/** Diagnostic name; appears in the bus's isolation warnings. */
name: string;
/** Event types this reaction handles. Others are not delivered to it. */
types: readonly WorkflowLifecycleEvent["type"][];
handle: (event: WorkflowLifecycleEvent) => void | Promise<void>;
}
/** Unsubscribe functions for the currently-registered set. */
let active: Array<() => void> = [];
/**
* Register the engine's post-commit reactions on the global bus. Idempotent —
* a second call replaces the first registration rather than adding to it.
*
* Returns an unsubscribe function for symmetry with the bus API; callers that
* hold the engine for a process lifetime can ignore it.
*
* FNXC:WorkflowEvents 2026-07-27-15:30 (U3, PR #2467 review):
* The returned cleanup is scoped to THIS registration, not to the module. It
* previously returned `unregisterWorkflowEventSubscribers`, so a stale handle
* from an earlier call would silently unsubscribe a LATER registration's set —
* an engine restart followed by a deferred cleanup would leave the process with
* no reactions and no error. A per-call closure makes a stale cleanup a no-op
* instead.
*/
export function registerWorkflowEventSubscribers(
registrations: readonly WorkflowEventSubscriberRegistration[] = ENGINE_WORKFLOW_EVENT_SUBSCRIBERS,
): () => void {
unregisterWorkflowEventSubscribers();
const bus = getWorkflowEventBus();
const mine = registrations.map((registration) => {
const subscriber: WorkflowEventSubscriber = (event) => {
// Type filtering lives here rather than in the bus so the bus stays a
// dumb fan-out and a handler's declared interest is visible at its
// registration site.
if (!registration.types.includes(event.type)) return;
return registration.handle(event);
};
return bus.subscribe(subscriber, { name: registration.name });
});
active = mine;
return () => {
for (const off of mine) off();
// Only clear the module handle when it still points at THIS registration,
// so a stale cleanup cannot blank a newer one's bookkeeping.
if (active === mine) active = [];
};
}
/** Drop every subscriber this module registered. Safe to call when none are. */
export function unregisterWorkflowEventSubscribers(): void {
for (const off of active) off();
active = [];
}
/** Count of currently-registered engine subscribers (diagnostics/tests). */
export function activeWorkflowEventSubscriberCount(): number {
return active.length;
}
/**
* The engine's default reaction set. Empty by design in U3 — see the module
* header. Each later unit appends the reaction it converts, with the
* characterization test proving it was non-authoritative beforehand.
*/
export const ENGINE_WORKFLOW_EVENT_SUBSCRIBERS: readonly WorkflowEventSubscriberRegistration[] = [];