From 061297b43290f34c134554a54ded980b8c2036b0 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Tue, 2 Jun 2026 20:15:29 -0700 Subject: [PATCH] refactor(compound-engineering): consolidate duplication from simplify pass Quality cleanup across the 9-unit build (behavior-preserving, 89 tests green): - extract createCeTaskWithLink so the work bridge and reconciler share one provenance+link contract (prevents drift) - discovery list scan probes readability via accessSync instead of reading and discarding full file bytes - makeError helper replaces ~7 duplicated CeArtifactError literals - shared asString route helper; drop dead pipelineIds set; collapse a double pipeline-state write and a redundant link re-query in advance - resolveStageSkillCwd no longer takes params it ignores --- .../src/__tests__/skill-reachability.test.ts | 2 +- .../src/artifacts/__tests__/discovery.test.ts | 26 +++-- .../src/artifacts/discovery.ts | 95 +++++++------------ .../src/routes/artifact-routes.ts | 5 +- .../src/routes/route-helpers.ts | 11 +++ .../src/routes/session-routes.ts | 5 +- .../src/session/orchestrator.ts | 46 +++------ .../src/sync/ce-task.ts | 67 +++++++++++++ .../src/sync/reconciler.ts | 73 ++++++-------- 9 files changed, 179 insertions(+), 151 deletions(-) create mode 100644 plugins/fusion-plugin-compound-engineering/src/routes/route-helpers.ts create mode 100644 plugins/fusion-plugin-compound-engineering/src/sync/ce-task.ts diff --git a/plugins/fusion-plugin-compound-engineering/src/__tests__/skill-reachability.test.ts b/plugins/fusion-plugin-compound-engineering/src/__tests__/skill-reachability.test.ts index 7ed2116696..6c361b6fb7 100644 --- a/plugins/fusion-plugin-compound-engineering/src/__tests__/skill-reachability.test.ts +++ b/plugins/fusion-plugin-compound-engineering/src/__tests__/skill-reachability.test.ts @@ -51,7 +51,7 @@ describe("stage skill reachability (carry-forward, resolver-layer proof)", () => expect(existsSync(installedSkillMd)).toBe(true); // resolveStageSkillCwd returns a plugin-local directory (never a global one). - const cwd = resolveStageSkillCwd(stage); + const cwd = resolveStageSkillCwd(); expect(cwd).toMatch(/\.fusion-ce-skills$/); }); diff --git a/plugins/fusion-plugin-compound-engineering/src/artifacts/__tests__/discovery.test.ts b/plugins/fusion-plugin-compound-engineering/src/artifacts/__tests__/discovery.test.ts index 716e29f9df..b6fbd1c489 100644 --- a/plugins/fusion-plugin-compound-engineering/src/artifacts/__tests__/discovery.test.ts +++ b/plugins/fusion-plugin-compound-engineering/src/artifacts/__tests__/discovery.test.ts @@ -3,10 +3,14 @@ import * as realFs from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; -// Mock node:fs so we can observe/inject behaviour around readFileSync without -// relying on vi.spyOn (ESM namespace exports are not configurable). The hooks -// below default to passthrough and individual tests override them. +// Mock node:fs so we can observe/inject behaviour around readFileSync and +// accessSync without relying on vi.spyOn (ESM namespace exports are not +// configurable). The list scan probes readability with accessSync (no bytes +// read); readFileSync is only used when an artifact's content is actually +// fetched (readArtifactById). The hooks below default to passthrough and +// individual tests override them. let readFileHook: ((path: realFs.PathOrFileDescriptor, original: typeof realFs.readFileSync, args: unknown[]) => unknown) | undefined; +let accessHook: ((path: realFs.PathLike, original: typeof realFs.accessSync, args: unknown[]) => unknown) | undefined; vi.mock("node:fs", async (importOriginal) => { const actual = await importOriginal(); @@ -16,6 +20,10 @@ vi.mock("node:fs", async (importOriginal) => { if (readFileHook) return readFileHook(path, actual.readFileSync, args); return (actual.readFileSync as (...a: unknown[]) => unknown)(path, ...args); }, + accessSync: (path: realFs.PathLike, ...args: unknown[]) => { + if (accessHook) return accessHook(path, actual.accessSync, args); + return (actual.accessSync as (...a: unknown[]) => unknown)(path, ...args); + }, }; }); @@ -32,6 +40,7 @@ describe("discoverArtifacts", () => { afterEach(() => { if (root) rmSync(root, { recursive: true, force: true }); readFileHook = undefined; + accessHook = undefined; vi.restoreAllMocks(); }); @@ -113,8 +122,9 @@ describe("discoverArtifacts", () => { const readable = join(root, "docs/plans/good.md"); writeFileSync(readable, "good"); - // Simulate a malformed/unreadable artifact: the specific file throws on read. - readFileHook = (path, original, args) => { + // Simulate a malformed/unreadable artifact: the specific file throws when + // the list scan probes readability (accessSync). + accessHook = (path, original, args) => { if (typeof path === "string" && path.endsWith("good.md")) { throw new Error("EIO: simulated read failure"); } @@ -147,14 +157,16 @@ describe("discoverArtifacts", () => { writeFileSync(join(root, "docs/random/r.md"), "unrelated"); // docs subtree but not conventional const opened: string[] = []; - readFileHook = (path, original, args) => { + // The list scan probes readability with accessSync (no bytes read); track + // exactly which paths it touches. + accessHook = (path, original, args) => { if (typeof path === "string") opened.push(path); return (original as (...a: unknown[]) => unknown)(path, ...args); }; const result = discoverArtifacts(root); - // Only the two conventional artifacts were read. + // Only the two conventional artifacts were probed. expect(opened.some((p) => p.endsWith("STRATEGY.md"))).toBe(true); expect(opened.some((p) => p.endsWith(join("ideation", "keep.md")))).toBe(true); // Nothing outside the allowlist was opened. diff --git a/plugins/fusion-plugin-compound-engineering/src/artifacts/discovery.ts b/plugins/fusion-plugin-compound-engineering/src/artifacts/discovery.ts index d27a4a8b31..cb551546d1 100644 --- a/plugins/fusion-plugin-compound-engineering/src/artifacts/discovery.ts +++ b/plugins/fusion-plugin-compound-engineering/src/artifacts/discovery.ts @@ -1,4 +1,4 @@ -import { readdirSync, readFileSync, statSync } from "node:fs"; +import { accessSync, constants, readdirSync, readFileSync, statSync } from "node:fs"; import { isAbsolute, join, relative, sep } from "node:path"; /** @@ -120,6 +120,18 @@ function makeId(stage: CeArtifactStage, relPath: string): string { return `${stage}:${relPath}`; } +/** Build a uniform `error` entry, deriving `id`/`name` from `(stage, relPath)`. */ +function makeError(stage: CeArtifactStage, relPath: string, message: string): CeArtifactError { + return { + id: makeId(stage, relPath), + stage, + path: relPath, + name: relPath.split("/").pop() ?? relPath, + kind: "error", + error: message, + }; +} + function readArtifactEntry( stage: CeArtifactStage, root: string, @@ -130,30 +142,16 @@ function readArtifactEntry( const name = relPath.split("/").pop() ?? relPath; // Defense in depth: refuse anything that escaped the conventional location. if (!isWithin(root, locationAbs, abs)) { - return { - id: makeId(stage, relPath), - stage, - path: relPath, - name, - kind: "error", - error: "Path is outside its conventional location", - }; + return makeError(stage, relPath, "Path is outside its conventional location"); } try { const st = statSync(abs); if (st.size > MAX_ARTIFACT_BYTES) { - return { - id: makeId(stage, relPath), - stage, - path: relPath, - name, - kind: "error", - error: `Artifact too large to read (${st.size} bytes)`, - }; + return makeError(stage, relPath, `Artifact too large to read (${st.size} bytes)`); } - // Read eagerly so a malformed/unreadable file is surfaced now as an error - // entry rather than crashing later at render time. - readFileSync(abs, "utf8"); + // Probe readability (no bytes transferred) so a malformed/unreadable file is + // surfaced now as an error entry rather than crashing later at render time. + accessSync(abs, constants.R_OK); return { id: makeId(stage, relPath), stage, @@ -164,14 +162,7 @@ function readArtifactEntry( kind: "artifact", }; } catch (err) { - return { - id: makeId(stage, relPath), - stage, - path: relPath, - name, - kind: "error", - error: err instanceof Error ? err.message : String(err), - }; + return makeError(stage, relPath, err instanceof Error ? err.message : String(err)); } } @@ -205,14 +196,13 @@ function discoverLocation(root: string, loc: ConventionalLocation): CeArtifactGr entries.push(readArtifactEntry(loc.stage, root, locationAbs, locationAbs, toPosix(loc.path))); } else { // A conventional file path that is actually a directory is malformed. - entries.push({ - id: makeId(loc.stage, toPosix(loc.path)), - stage: loc.stage, - path: toPosix(loc.path), - name: loc.path.split("/").pop() ?? loc.path, - kind: "error", - error: "Expected a file at the conventional location but found a directory", - }); + entries.push( + makeError( + loc.stage, + toPosix(loc.path), + "Expected a file at the conventional location but found a directory", + ), + ); } return { stage: loc.stage, label: loc.label, present, entries: sortEntries(entries) }; } @@ -222,26 +212,18 @@ function discoverLocation(root: string, loc: ConventionalLocation): CeArtifactGr let names: string[] = []; try { if (!st.isDirectory()) { - entries.push({ - id: makeId(loc.stage, toPosix(loc.path)), - stage: loc.stage, - path: toPosix(loc.path), - name: loc.path.split("/").pop() ?? loc.path, - kind: "error", - error: "Expected a directory at the conventional location but found a file", - }); + entries.push( + makeError( + loc.stage, + toPosix(loc.path), + "Expected a directory at the conventional location but found a file", + ), + ); return { stage: loc.stage, label: loc.label, present, entries: sortEntries(entries) }; } names = readdirSync(locationAbs); } catch (err) { - entries.push({ - id: makeId(loc.stage, toPosix(loc.path)), - stage: loc.stage, - path: toPosix(loc.path), - name: loc.path.split("/").pop() ?? loc.path, - kind: "error", - error: err instanceof Error ? err.message : String(err), - }); + entries.push(makeError(loc.stage, toPosix(loc.path), err instanceof Error ? err.message : String(err))); return { stage: loc.stage, label: loc.label, present, entries: sortEntries(entries) }; } @@ -257,14 +239,7 @@ function discoverLocation(root: string, loc: ConventionalLocation): CeArtifactGr try { childStat = statSync(abs); } catch (err) { - entries.push({ - id: makeId(loc.stage, relPath), - stage: loc.stage, - path: relPath, - name: childName, - kind: "error", - error: err instanceof Error ? err.message : String(err), - }); + entries.push(makeError(loc.stage, relPath, err instanceof Error ? err.message : String(err))); continue; } if (!childStat.isFile()) continue; diff --git a/plugins/fusion-plugin-compound-engineering/src/routes/artifact-routes.ts b/plugins/fusion-plugin-compound-engineering/src/routes/artifact-routes.ts index d68a5f5836..21e3fbc264 100644 --- a/plugins/fusion-plugin-compound-engineering/src/routes/artifact-routes.ts +++ b/plugins/fusion-plugin-compound-engineering/src/routes/artifact-routes.ts @@ -1,5 +1,6 @@ import type { PluginContext, PluginRouteDefinition, PluginRouteResponse } from "@fusion/core"; import { discoverArtifacts, readArtifactById } from "../artifacts/discovery.js"; +import { asString } from "./route-helpers.js"; /** * Artifact routes (U3): list discovered CE artifacts grouped by stage, and read @@ -19,10 +20,6 @@ interface RouteRequest { query?: Record; } -function asString(v: unknown): string | undefined { - return typeof v === "string" && v.length > 0 ? v : undefined; -} - async function resolveProjectRoot(ctx: PluginContext, projectId?: string): Promise { if (projectId && ctx.resolveProjectTaskStore) { try { diff --git a/plugins/fusion-plugin-compound-engineering/src/routes/route-helpers.ts b/plugins/fusion-plugin-compound-engineering/src/routes/route-helpers.ts new file mode 100644 index 0000000000..284540ca91 --- /dev/null +++ b/plugins/fusion-plugin-compound-engineering/src/routes/route-helpers.ts @@ -0,0 +1,11 @@ +/** + * Shared route helpers. + * + * `asString` coerces an unknown request value to a non-empty string or + * `undefined` — the common "optional, must be a real string" guard used across + * the CE route handlers. (settings.ts has a different-signature `asString` that + * is intentionally NOT consolidated here.) + */ +export function asString(v: unknown): string | undefined { + return typeof v === "string" && v.length > 0 ? v : undefined; +} diff --git a/plugins/fusion-plugin-compound-engineering/src/routes/session-routes.ts b/plugins/fusion-plugin-compound-engineering/src/routes/session-routes.ts index c7123e6a27..015691a6aa 100644 --- a/plugins/fusion-plugin-compound-engineering/src/routes/session-routes.ts +++ b/plugins/fusion-plugin-compound-engineering/src/routes/session-routes.ts @@ -2,6 +2,7 @@ import type { PluginContext, PluginRouteDefinition, PluginRouteResponse } from " import { CeOrchestrator } from "../session/orchestrator.js"; import { getCeSessionStore } from "../session/session-store.js"; import { getCePipelineStore } from "../sync/pipeline-store.js"; +import { asString } from "./route-helpers.js"; /** * Session routes (U5): start / answer / resume / get-session-state. @@ -44,10 +45,6 @@ function badRequest(message: string): PluginRouteResponse { return { status: 400, body: { error: message } }; } -function asString(v: unknown): string | undefined { - return typeof v === "string" && v.length > 0 ? v : undefined; -} - export function createSessionRoutes(): PluginRouteDefinition[] { return [ { diff --git a/plugins/fusion-plugin-compound-engineering/src/session/orchestrator.ts b/plugins/fusion-plugin-compound-engineering/src/session/orchestrator.ts index 66547d2dd3..3a81db1728 100644 --- a/plugins/fusion-plugin-compound-engineering/src/session/orchestrator.ts +++ b/plugins/fusion-plugin-compound-engineering/src/session/orchestrator.ts @@ -9,6 +9,7 @@ import type { } from "@fusion/core"; import { resolveDefaultInstallTargetRoot } from "../skill-installation.js"; import { getCePipelineStore, type CePipelineStore } from "../sync/pipeline-store.js"; +import { createCeTaskWithLink } from "../sync/ce-task.js"; import { getDefaultModelId, getDefaultProvider, getEnabledStages } from "../settings.js"; import type { CeSession, CeSessionStore } from "./session-store.js"; import { getCeSessionStore } from "./session-store.js"; @@ -21,18 +22,11 @@ import { getStage, type CeStageDefinition } from "./stage-registry.js"; export const WORK_STAGE_ID = "work"; /** - * The CE marker plugin id recorded on every CE-originated board task and link. - * Kept as a constant so U8's sync code reuses the same identity. + * CE provenance identity constants. Defined in `../sync/ce-task.ts` (so the + * reconciler can reuse them without importing this module) and re-exported here + * for existing consumers (index.ts, tests) that import them from the orchestrator. */ -export const CE_PLUGIN_ID = "fusion-plugin-compound-engineering"; - -/** - * SourceType chosen for CE-originated automated work. The work stage is a step in - * the CE pipeline, so `workflow_step` is the closest existing provenance value - * (vs. inventing a new SourceType). The CE marker + back-reference convenience - * copy ride in `sourceMetadata`; the authoritative link is the pipeline-link row. - */ -export const CE_WORK_SOURCE_TYPE = "workflow_step" as const; +export { CE_PLUGIN_ID, CE_WORK_SOURCE_TYPE } from "../sync/ce-task.js"; /** * COMPLETION-PAYLOAD → TASKS CONTRACT (U7). @@ -115,13 +109,12 @@ export interface OrchestratorDeps { * resolver layer (like U2 did) in the tests — the U4 options surface cannot yet * carry an explicit skill-path, so a complete fix needs U4's options to gain a * `requestedSkillNames`/`additionalSkillPaths` field forwarded into - * `createFnAgent`. That gap is documented as a carry-forward for U6/follow-up. + * `createFnAgent`. That gap is documented as a carry-forward for the follow-up, + * which will re-expand this to derive a per-stage/per-project path. */ -export function resolveStageSkillCwd(stage: CeStageDefinition, projectRoot?: string): string { +export function resolveStageSkillCwd(): string { // The install target root holds `/SKILL.md` for each installed // skill; using it as the discovery root makes the stage's skill loadable. - void stage; - void projectRoot; return resolveDefaultInstallTargetRoot(); } @@ -203,7 +196,7 @@ export class CeOrchestrator { }); this.store.appendHistory(session.id, { role: "user", text: opts.openingMessage, at: new Date().toISOString() }); - const cwd = resolveStageSkillCwd(stage, this.projectRoot); + const cwd = resolveStageSkillCwd(); const systemPrompt = buildStageSystemPrompt(stage); // Setting-gated model selection (U9): pass the operator's default @@ -392,25 +385,12 @@ export class CeOrchestrator { const description = spec.description.trim(); if (!description) continue; // createTask rejects blank descriptions. - const task = await this.ctx.taskStore.createTask({ + // Shared contract: create the CE-tagged board task AND its authoritative + // pipeline-link row (FN-5719) in one place (see createCeTaskWithLink). + await createCeTaskWithLink(this.ctx.taskStore, this.pipelineStore, { title: spec.title, description, - column: spec.column as never, - source: { - sourceType: CE_WORK_SOURCE_TYPE, - sourceSessionId: cePipelineId, - sourceMetadata: { - pluginId: CE_PLUGIN_ID, - cePipelineId, - ceStageId, - ceArtifactPath, - }, - }, - }); - - // Authoritative back-reference (FN-5719): the link row, not task-row JSON. - this.pipelineStore.createLink({ - taskId: task.id, + column: spec.column, cePipelineId, ceStageId, ceArtifactPath, diff --git a/plugins/fusion-plugin-compound-engineering/src/sync/ce-task.ts b/plugins/fusion-plugin-compound-engineering/src/sync/ce-task.ts new file mode 100644 index 0000000000..6b351af834 --- /dev/null +++ b/plugins/fusion-plugin-compound-engineering/src/sync/ce-task.ts @@ -0,0 +1,67 @@ +import type { CePipelineStore } from "./pipeline-store.js"; + +/** + * The CE marker plugin id recorded on every CE-originated board task and link. + * Kept here (separate from the orchestrator) so both the work-bridge AND the + * reconciler can build the identical provenance payload without importing the + * orchestrator module (which would create a cycle through pipeline-store). + */ +export const CE_PLUGIN_ID = "fusion-plugin-compound-engineering"; + +/** + * SourceType chosen for CE-originated automated work. The work stage is a step in + * the CE pipeline, so `workflow_step` is the closest existing provenance value + * (vs. inventing a new SourceType). The CE marker + back-reference convenience + * copy ride in `sourceMetadata`; the authoritative link is the pipeline-link row. + */ +export const CE_WORK_SOURCE_TYPE = "workflow_step" as const; + +/** What a CE task+link pair needs: provenance ids plus the board task content. */ +export interface CreateCeTaskWithLinkSpec { + title?: string; + description: string; + column?: string; + cePipelineId: string; + ceStageId: string; + ceArtifactPath: string | null; +} + +/** + * Build the shared CE provenance+link contract: create a board task tagged + * CE-originated (source payload) and record the authoritative pipeline-link row + * (FN-5719) resolving task→pipeline/stage/artifact. Returns the created task. + * + * This is the single source of truth for that contract — both the orchestrator's + * work bridge and the reconciler's outbound advance use it so the provenance + * payload and link row can never drift apart. + */ +export async function createCeTaskWithLink( + taskStore: { createTask(input: unknown): Promise }, + pipelineStore: CePipelineStore, + spec: CreateCeTaskWithLinkSpec, +): Promise { + const task = await taskStore.createTask({ + title: spec.title, + description: spec.description, + column: spec.column as never, + source: { + sourceType: CE_WORK_SOURCE_TYPE, + sourceSessionId: spec.cePipelineId, + sourceMetadata: { + pluginId: CE_PLUGIN_ID, + cePipelineId: spec.cePipelineId, + ceStageId: spec.ceStageId, + ceArtifactPath: spec.ceArtifactPath, + }, + }, + }); + + pipelineStore.createLink({ + taskId: task.id, + cePipelineId: spec.cePipelineId, + ceStageId: spec.ceStageId, + ceArtifactPath: spec.ceArtifactPath, + }); + + return task; +} diff --git a/plugins/fusion-plugin-compound-engineering/src/sync/reconciler.ts b/plugins/fusion-plugin-compound-engineering/src/sync/reconciler.ts index 58da94dcb8..f6cdf4cef1 100644 --- a/plugins/fusion-plugin-compound-engineering/src/sync/reconciler.ts +++ b/plugins/fusion-plugin-compound-engineering/src/sync/reconciler.ts @@ -1,9 +1,6 @@ import type { PluginContext, Task } from "@fusion/core"; import { listStages } from "../session/stage-registry.js"; -import { - CE_PLUGIN_ID, - CE_WORK_SOURCE_TYPE, -} from "../session/orchestrator.js"; +import { createCeTaskWithLink } from "./ce-task.js"; import { getCePipelineStore, type CePipelineLink, @@ -92,9 +89,7 @@ export class CeReconciler { // re-derived from board truth below, so a queue entry for an already-handled // transition is harmless. const pending = this.store.listPendingSync(); - const pipelineIds = new Set(); for (const entry of pending) { - pipelineIds.add(entry.cePipelineId); this.store.markSyncProcessed(entry.id); result.drained++; } @@ -103,7 +98,6 @@ export class CeReconciler { // ones with queued entries. This is what recovers a dropped/never-enqueued // hook event — board truth is compared against pipeline state regardless of // whether a queue row exists. - void pipelineIds; const states = this.store.listAllState(); for (const state of states) { result.inspected++; @@ -126,14 +120,14 @@ export class CeReconciler { ): Promise<{ created: boolean } | undefined> { if (state.status === "completed") return undefined; - // Links produced by this pipeline AT its current stage are the board tasks - // whose completion gates advancement. - const links = this.store - .listByPipeline(state.cePipelineId) - .filter((l) => l.ceStageId === state.currentStage); - if (links.length === 0) return undefined; + // All links for this pipeline (fetched once, reused by advance's idempotency + // check). The board tasks whose completion gates advancement are this + // pipeline's links AT its current stage. + const links = this.store.listByPipeline(state.cePipelineId); + const currentStageLinks = links.filter((l) => l.ceStageId === state.currentStage); + if (currentStageLinks.length === 0) return undefined; - const tasks = await this.loadTasks(links); + const tasks = await this.loadTasks(currentStageLinks); if (tasks.length === 0) return undefined; // Advancement rule: every current-stage board task has reached a terminal @@ -164,7 +158,7 @@ export class CeReconciler { // Advancing only moves the CE-owned fields + creates a NEW board task; it // never mutates the already-terminal board tasks, so the two writers never // contend over the same cell. - const created = await this.advance(state, next); + const created = await this.advance(state, next, links); return { created }; } @@ -173,44 +167,39 @@ export class CeReconciler { * by creating the next-stage board task (board-owned write on a NEW row). * Idempotent: if a link for the next stage already exists, we don't duplicate. */ - private async advance(state: CePipelineState, nextStage: string): Promise { + private async advance( + state: CePipelineState, + nextStage: string, + links: CePipelineLink[], + ): Promise { // Idempotency guard: if we already advanced (a next-stage link exists), just // ensure state is consistent and skip the outbound create. - const already = this.store - .listByPipeline(state.cePipelineId) - .some((l) => l.ceStageId === nextStage); + const already = links.some((l) => l.ceStageId === nextStage); - this.store.transitionState(state.cePipelineId, { - currentStage: nextStage, - status: already ? "running" : "awaiting_board", - }); + if (already) { + this.store.transitionState(state.cePipelineId, { + currentStage: nextStage, + status: "running", + }); + return false; + } - if (already) return false; - - const task = await this.ctx.taskStore.createTask({ + // Shared contract: create the CE-tagged next-stage board task AND its + // authoritative pipeline-link row (FN-5719) in one place. + const task = await createCeTaskWithLink(this.ctx.taskStore, this.store, { title: `CE ${nextStage}: continue pipeline`, description: `Continue the compound-engineering pipeline at the "${nextStage}" stage.`, - source: { - sourceType: CE_WORK_SOURCE_TYPE, - sourceSessionId: state.cePipelineId, - sourceMetadata: { - pluginId: CE_PLUGIN_ID, - cePipelineId: state.cePipelineId, - ceStageId: nextStage, - ceArtifactPath: state.lastArtifactPath, - }, - }, - }); - - this.store.createLink({ - taskId: task.id, cePipelineId: state.cePipelineId, ceStageId: nextStage, ceArtifactPath: state.lastArtifactPath, }); - // Pipeline is now waiting on the freshly-created board task. - this.store.transitionState(state.cePipelineId, { status: "awaiting_board" }); + // Single state write on the create path: advance to the next stage and mark + // the pipeline as waiting on the freshly-created board task. + this.store.transitionState(state.cePipelineId, { + currentStage: nextStage, + status: "awaiting_board", + }); this.ctx.emitEvent("compound-engineering:pipeline-advanced", { cePipelineId: state.cePipelineId,