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

FN-8658: cover task update producer delivery paths

Browse Source 
Expand task update lane coverage across every direct and failure-isolated producer.

- Inventory direct and safe task:updated producer routes.
- Exercise warm and cold lane metadata through public producer operations.
- Cover lifecycle, workflow integrity, and completion update paths.

Files changed:
 .../task-updated-lanes-emit-surfaces.test.ts       | 308 +++++++++++++++++----
 1 file changed, 256 insertions(+), 52 deletions(-)

Fusion-Task-Id: FN-8658

Fusion-Task-Lineage: 7d93593c-6f5a-4299-96f3-f4ee8f260e41

Co-authored-by: Fusion (runfusion.ai) <noreply@runfusion.ai>
This commit is contained in:
gsxdsm
2026-08-01 01:41:13 -07:00
parent 13f7266b67
commit bdcfcbbc52
1 changed files with 255 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files

306
packages/core/src/__tests__/task-updated-lanes-emit-surfaces.test.ts
View File

@@ -1,4 +1,5 @@
import { describe, expect, it } from "vitest";
import { describe, expect, it, vi } from "vitest";
import { createTaskStoreForTest, pgDescribe } from "../__test-utils__/pg-test-harness.js";
import { readdirSync, readFileSync, statSync } from "node:fs";
import { join, relative, resolve } from "node:path";
import { TaskStore } from "../store.js";
@@ -7,39 +8,33 @@ import type { Task } from "../types.js";
const REPO_ROOT = resolve(__dirname, "../../../..");
const CORE_ROOT = join(REPO_ROOT, "packages/core/src");
const EMIT_SURFACES = [
"packages/core/src/store.ts",
"packages/core/src/task-store/audit-ops.ts",
"packages/core/src/task-store/branch-group-ops.ts",
"packages/core/src/task-store/comments-ops.ts",
"packages/core/src/task-store/lifecycle-ops.ts",
"packages/core/src/task-store/merge-queue-ops.ts",
"packages/core/src/task-store/moves.ts",
"packages/core/src/task-store/project-store-ops.ts",
"packages/core/src/task-store/task-artifacts-ops.ts",
"packages/core/src/task-store/task-mutation-ops.ts",
"packages/core/src/task-store/workflow-task-create-ops.ts",
] as const;
type ProducerRoute = "emit" | "safe";
/*
FNXC:WorkflowEvents 2026-08-01-07:35:
These are producer paths, not a sampled list of listeners. The static scan makes a newly added
producer fail closed, while the receiver assertion below proves each existing producer invokes the
TaskStore instance seam that is exercised with warm and cold cache states in this suite.
FNXC:WorkflowEvents 2026-08-01-07:44:
The cache seam has two production entry points: ordinary EventEmitter `emit` and the failure-isolated
`emitTaskLifecycleEventSafely`. Inventory both forms, including safe-only producers, because a
source scan for only `emit("task:updated")` leaves a producer unexercised and can hide a bypass.
Each registered route below is driven warm and cold through its real TaskStore delivery method.
*/
const TASK_STORE_EMIT_RECEIVERS: Record<(typeof EMIT_SURFACES)[number], readonly string[]> = {
"packages/core/src/store.ts": ["this.emit"],
"packages/core/src/task-store/audit-ops.ts": ["store.emit", "store.emitTaskLifecycleEventSafely"],
"packages/core/src/task-store/branch-group-ops.ts": ["store.emit"],
"packages/core/src/task-store/comments-ops.ts": ["store.emit"],
"packages/core/src/task-store/lifecycle-ops.ts": ["store.emit"],
"packages/core/src/task-store/merge-queue-ops.ts": ["store.emit"],
"packages/core/src/task-store/moves.ts": ["store.emit"],
"packages/core/src/task-store/project-store-ops.ts": ["store.emit"],
"packages/core/src/task-store/task-artifacts-ops.ts": ["store.emit"],
"packages/core/src/task-store/task-mutation-ops.ts": ["store.emitTaskLifecycleEventSafely", "store.emit"],
"packages/core/src/task-store/workflow-task-create-ops.ts": ["store.emit", "store.emitTaskLifecycleEventSafely"],
};
const PRODUCERS = {
"packages/core/src/store.ts": ["emit"],
"packages/core/src/task-store/audit-ops.ts": ["emit", "safe"],
"packages/core/src/task-store/branch-group-ops.ts": ["emit"],
"packages/core/src/task-store/comments-ops.ts": ["emit"],
"packages/core/src/task-store/lifecycle-ops.ts": ["emit"],
"packages/core/src/task-store/merge-queue-ops.ts": ["emit"],
"packages/core/src/task-store/moves.ts": ["emit"],
"packages/core/src/task-store/project-store-ops.ts": ["emit"],
"packages/core/src/task-store/task-artifacts-ops.ts": ["emit"],
"packages/core/src/task-store/task-mutation-ops.ts": ["emit", "safe"],
"packages/core/src/task-store/task-update.ts": ["safe"],
"packages/core/src/task-store/update-task-deps.ts": ["safe"],
"packages/core/src/task-store/workflow-integrity.ts": ["safe"],
"packages/core/src/task-store/workflow-task-create-ops.ts": ["emit", "safe"],
} as const satisfies Record<string, readonly ProducerRoute[]>;
const EMIT_SURFACES = Object.keys(PRODUCERS) as Array<keyof typeof PRODUCERS>;
function* walk(dir: string): Generator<string> {
for (const entry of readdirSync(dir)) {
@@ -53,40 +48,249 @@ function* walk(dir: string): Generator<string> {
const task = { id: "FN-surface", column: "building" } as Task;
/*
FNXC:WorkflowEvents 2026-08-01-06:57:
Every production core update producer invokes the TaskStore instance emitter, not EventEmitter's
prototype or a private relay. That structural boundary is what lets one decorated TaskStore seam
cover every producer warm and cold without adding an IR read to hot mutation paths.
FNXC:WorkflowEvents 2026-08-01-07:44:
Every production core update producer invokes a TaskStore delivery method, not EventEmitter's
prototype or a private relay. That structural boundary is what lets one decorated seam cover every
producer warm and cold without adding an IR read to hot mutation paths.
*/
describe("core task:updated emit surface", () => {
it("registers every production producer and constrains each to TaskStore's decorated emitter", () => {
const emittingModules = [...walk(CORE_ROOT)]
.filter((file) => readFileSync(file, "utf8").includes('emit("task:updated"'))
.map((file) => relative(REPO_ROOT, file).split("\\").join("/"))
.sort();
it("registers every direct and safe production producer", () => {
const routesByModule = new Map<string, Set<ProducerRoute>>();
for (const file of walk(CORE_ROOT)) {
const source = readFileSync(file, "utf8");
const routes = new Set<ProducerRoute>();
if (source.includes('emit("task:updated"')) routes.add("emit");
if (source.includes('emitTaskLifecycleEventSafely("task:updated"')) routes.add("safe");
if (routes.size > 0) routesByModule.set(relative(REPO_ROOT, file).split("\\").join("/"), routes);
}
expect(emittingModules).toEqual([...EMIT_SURFACES].sort());
expect(Object.keys(TASK_STORE_EMIT_RECEIVERS).sort()).toEqual([...EMIT_SURFACES].sort());
for (const file of emittingModules) {
expect([...routesByModule.keys()].sort()).toEqual([...EMIT_SURFACES].sort());
for (const [file, routes] of routesByModule) {
const expectedRoutes = PRODUCERS[file as keyof typeof PRODUCERS];
expect([...routes].sort(), `${file} must register every delivery route`).toEqual([...expectedRoutes].sort());
const source = readFileSync(join(REPO_ROOT, file), "utf8");
const receivers = TASK_STORE_EMIT_RECEIVERS[file as keyof typeof TASK_STORE_EMIT_RECEIVERS];
expect(receivers.some((receiver) => source.includes(receiver)), `${file} must call the TaskStore emitter`).toBe(true);
expect(source).not.toMatch(/EventEmitter\.prototype\.emit\([^\n]*task:updated/);
expect(source).not.toMatch(/EventEmitter\.prototype\.emit[^\n]*task:updated/);
}
});
it("delivers the warm and cold result at the TaskStore seam used by every registered producer", () => {
});
/*
FNXC:WorkflowEvents 2026-08-01-08:08:
A source inventory alone only proves where an update could originate. Execute public operations from
its producer modules so warm and cold assertions cover the producer-to-listener path, rather than
calling the central seam directly and mistaking seam coverage for producer coverage.
*/
pgDescribe("task:updated producer integration", () => {
async function withWarmAndColdUpdate(
description: string,
operation: (store: TaskStore, taskId: string, pass: "warm" | "cold") => Promise<unknown>,
setup?: (store: TaskStore, taskId: string) => Promise<unknown>,
): Promise<void> {
const harness = await createTaskStoreForTest({ prefix: `fusion_${description.replaceAll(/[^a-z]/g, "_")}_lanes` });
try {
const store = harness.store;
const created = await store.createTask({
description: `${description} lane surface`,
steps: [{ name: "surface step", status: "pending" }],
});
await setup?.(store, created.id);
const received: Array<{ lanes?: { wip?: string } } | undefined> = [];
store.on("task:updated", (emitted, meta) => {
if (emitted.id === created.id) received.push(meta);
});
store.laneCache.set(created.id, { wip: "building" });
await operation(store, created.id, "warm");
store.laneCache.invalidate(created.id);
await operation(store, created.id, "cold");
expect(received[0], `${description} warm operation`).toEqual({ lanes: { wip: "building" } });
expect(received.at(-1), `${description} cold operation`).toBeUndefined();
} finally {
await harness.teardown();
}
}
it("delivers warm then cold metadata through TaskStore's direct emit producer", async () => {
const harness = await createTaskStoreForTest({ prefix: "fusion_store_emit_lanes_surface" });
try {
const store = harness.store;
const created = await store.createTask({ description: "store emit lane surface" });
const received: Array<{ lanes?: { wip?: string } } | undefined> = [];
store.on("task:updated", (_task, meta) => received.push(meta));
store.laneCache.set(created.id, { wip: "building" });
store.emit("task:updated", created);
store.laneCache.invalidate(created.id);
store.emit("task:updated", created);
expect(received).toEqual([{ lanes: { wip: "building" } }, undefined]);
} finally {
await harness.teardown();
}
});
it("executes audit-ops logEntry through its real update producer", async () => {
await withWarmAndColdUpdate("audit", (store, id, pass) => store.logEntry(id, `surface ${pass}`));
});
it("executes branch-group-ops pauseTask through its real update producer", async () => {
await withWarmAndColdUpdate("branch", (store, id, pass) => store.pauseTask(id, pass === "warm"));
});
it("executes comments-ops addComment through its real update producer", async () => {
await withWarmAndColdUpdate("comment", (store, id, pass) => store.addComment(id, `surface ${pass}`, "agent", { skipRefinement: true }));
});
it("executes merge-queue-ops updateStep through its real update producer", async () => {
await withWarmAndColdUpdate(
"merge_step",
(store, id, pass) => store.updateStep(id, 0, pass === "warm" ? "in-progress" : "done"),
(store, id) => store.updateTask(id, { steps: [{ name: "surface step", status: "pending" }] }),
);
});
it("executes workflow-task-create-ops branch assignment through its real update producer", async () => {
await withWarmAndColdUpdate("workflow_task_create", (store, id) => store.setTaskBranchGroup(id, null));
});
it("executes project-store-ops updateIssueInfo through its real update producer", async () => {
await withWarmAndColdUpdate("issue", (store, id, pass) => store.updateIssueInfo(id, {
number: pass === "warm" ? 1 : 2,
url: `https://example.test/issues/${pass}`,
state: "open",
title: `surface ${pass}`,
}));
});
it("executes task-artifacts-ops steering comments through its real update producer", async () => {
await withWarmAndColdUpdate("artifact", (store, id, pass) => store.addSteeringComment(id, `surface ${pass}`));
});
it("executes task-mutation-ops atomic updates through its real update producer", async () => {
await withWarmAndColdUpdate("mutation", (store, id, pass) => store.updateTaskAtomic(id, () => ({ title: `surface ${pass}` })));
});
it("executes moves' same-column completion update producer", async () => {
const harness = await createTaskStoreForTest({ prefix: "fusion_moves_lanes_surface" });
try {
const store = harness.store;
const created = await store.createTask({ description: "move lane surface" });
await store.moveTask(created.id, "in-progress");
await store.moveTask(created.id, "done");
await store.updateTask(created.id, { status: "failed" });
const received: Array<{ lanes?: { wip?: string } } | undefined> = [];
store.on("task:updated", (_task, meta) => received.push(meta));
store.laneCache.set(created.id, { wip: "building" });
await store.moveTask(created.id, "done");
store.laneCache.invalidate(created.id);
await store.updateTask(created.id, { status: "failed" });
await store.moveTask(created.id, "done");
expect(received[0]).toEqual({ lanes: { wip: "building" } });
expect(received.at(-1)).toBeUndefined();
} finally {
await harness.teardown();
}
});
it("executes lifecycle polling's real update producer", async () => {
const store = new TaskStore(process.cwd());
const polledTask = { ...task };
const received: Array<{ lanes?: { wip?: string } } | undefined> = [];
store.on("task:updated", (_task, meta) => received.push(meta));
Object.defineProperty(store, "db", {
value: {
getLastModified: vi.fn().mockReturnValueOnce(1).mockReturnValueOnce(2),
prepare: vi.fn((query: string) => query.includes("SELECT id FROM tasks")
? { all: () => [{ id: task.id }] }
: { all: () => [{}] }),
},
});
vi.spyOn(store, "rowToTask").mockReturnValue(polledTask);
store.taskCache.set(task.id, { ...task });
// The producers above all invoke this real TaskStore override; this verifies their shared delivery
// contract rather than allowing a module-specific helper to silently bypass cache decoration.
store.laneCache.set(task.id, { wip: "building" });
store.emit("task:updated", task);
await store.checkForChanges();
store.laneCache.invalidate(task.id);
store.emit("task:updated", task);
store.taskCache.set(task.id, { ...task });
await store.checkForChanges();
expect(received).toEqual([{ lanes: { wip: "building" } }, undefined]);
});
it("executes workflow-integrity's real safe update producer", async () => {
const store = new TaskStore(process.cwd());
const stampableTask = { ...task, column: "in-review", autoMerge: true };
const received: Array<{ lanes?: { wip?: string } } | undefined> = [];
let marked = false;
store.on("task:updated", (_task, meta) => received.push(meta));
Object.defineProperty(store, "db", {
value: {
prepare: vi.fn(() => ({
get: () => marked ? { value: "1" } : undefined,
run: () => { marked = true; },
})),
bumpLastModified: vi.fn(),
},
});
vi.spyOn(store, "listLegacyAutoMergeStampCandidates").mockResolvedValue([stampableTask]);
vi.spyOn(store, "getTask").mockResolvedValue(stampableTask);
vi.spyOn(store, "isLegacyAutoMergeStampCandidate").mockReturnValue(true);
vi.spyOn(store, "atomicWriteTaskJson").mockResolvedValue();
vi.spyOn(store, "recordRunAuditEvent").mockResolvedValue({} as never);
store.laneCache.set(task.id, { wip: "building" });
await store.markLegacyAutoMergeStampsOnce();
marked = false;
store.laneCache.invalidate(task.id);
await store.markLegacyAutoMergeStampsOnce();
expect(received).toEqual([{ lanes: { wip: "building" } }, undefined]);
});
it("delivers warm then cold metadata through task-update's real safe emission", async () => {
const harness = await createTaskStoreForTest({ prefix: "fusion_update_lanes_surface" });
try {
const store = harness.store;
const created = await store.createTask({ description: "update lane surface" });
const received: Array<{ lanes?: { wip?: string } } | undefined> = [];
store.on("task:updated", (_task, meta) => received.push(meta));
store.laneCache.set(created.id, { wip: "building" });
await store.updateTask(created.id, { title: "warm update" });
store.laneCache.invalidate(created.id);
await store.updateTask(created.id, { title: "cold update" });
expect(received).toEqual([{ lanes: { wip: "building" } }, undefined]);
} finally {
await harness.teardown();
}
});
it("delivers warm then cold metadata through dependency updates' real safe emission", async () => {
const harness = await createTaskStoreForTest({ prefix: "fusion_dependency_lanes_surface" });
try {
const store = harness.store;
const prerequisite = await store.createTask({ description: "dependency prerequisite" });
const dependent = await store.createTask({ description: "dependency lane surface" });
const received: Array<{ lanes?: { wip?: string } } | undefined> = [];
store.on("task:updated", (_task, meta) => {
if (_task.id === dependent.id) received.push(meta);
});
store.laneCache.set(dependent.id, { wip: "building" });
await store.updateTaskDependencies(dependent.id, { operation: "add", dependency: prerequisite.id });
store.laneCache.invalidate(dependent.id);
await store.updateTaskDependencies(dependent.id, { operation: "remove", dependency: prerequisite.id });
expect(received).toEqual([{ lanes: { wip: "building" } }, undefined]);
} finally {
await harness.teardown();
}
});
});