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

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$/);
   });
 

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<typeof realFs>();
@@ -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.

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;

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<string, string | string[] | undefined>;
 }
 
-function asString(v: unknown): string | undefined {
-  return typeof v === "string" && v.length > 0 ? v : undefined;
-}
-
 async function resolveProjectRoot(ctx: PluginContext, projectId?: string): Promise<string> {
   if (projectId && ctx.resolveProjectTaskStore) {
     try {

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;
+}

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 [
     {

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 `<skillId>/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,

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<T extends { id: string }>(
+  taskStore: { createTask(input: unknown): Promise<T> },
+  pipelineStore: CePipelineStore,
+  spec: CreateCeTaskWithLinkSpec,
+): Promise<T> {
+  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;
+}

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<string>();
     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<boolean> {
+  private async advance(
+    state: CePipelineState,
+    nextStage: string,
+    links: CePipelineLink[],
+  ): Promise<boolean> {
     // 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,