diff --git a/plugins/fusion-plugin-compound-engineering/src/__tests__/sync.test.ts b/plugins/fusion-plugin-compound-engineering/src/__tests__/sync.test.ts
new file mode 100644
index 0000000000..0fb0a0efa7
--- /dev/null
+++ b/plugins/fusion-plugin-compound-engineering/src/__tests__/sync.test.ts
@@ -0,0 +1,250 @@
+import { rm } from "node:fs/promises";
+import { mkdtempSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import type { InteractiveAiSessionEvent, PluginContext, Task } from "@fusion/core";
+import { TaskStore } from "@fusion/core";
+import plugin, {
+  CeOrchestrator,
+  CE_PLUGIN_ID,
+  WORK_STAGE_ID,
+} from "../index.js";
+import { getCePipelineStore } from "../sync/pipeline-store.js";
+import { CeReconciler, reconcileCePipelines } from "../sync/reconciler.js";
+import { makeScriptedSession } from "./_harness.js";
+
+/**
+ * U8 bidirectional-sync tests. REAL in-memory TaskStore (genuine board tasks) +
+ * the actual lifecycle-hook handlers and reconciler. We exercise the two
+ * separate state machines (board columns vs ce_pipeline_state) and prove the
+ * dropped-event convergence path independently of the hooks.
+ */
+
+let rootDir: string;
+let taskStore: TaskStore;
+let ctx: PluginContext;
+let emitted: Array<{ event: string; data: unknown }>;
+
+beforeEach(async () => {
+  rootDir = mkdtempSync(join(tmpdir(), "ce-sync-"));
+  taskStore = new TaskStore(rootDir, join(rootDir, ".fusion-global"), { inMemoryDb: true });
+  await taskStore.init();
+  emitted = [];
+  ctx = {
+    pluginId: CE_PLUGIN_ID,
+    taskStore,
+    settings: {},
+    logger: { info: vi.fn(), warn: vi.fn(), error: vi.fn(), debug: vi.fn() },
+    emitEvent: (event: string, data: unknown) => emitted.push({ event, data }),
+  } as unknown as PluginContext;
+});
+
+afterEach(async () => {
+  taskStore?.close();
+  await rm(rootDir, { recursive: true, force: true, maxRetries: 5, retryDelay: 50 });
+});
+
+/** The board enforces ordered transitions; walk a task forward to a target column. */
+const COLUMN_PATH = ["triage", "todo", "in-progress", "in-review", "done"];
+async function moveTo(taskId: string, target: string): Promise<void> {
+  const current = (await taskStore.getTask(taskId))!.column;
+  const from = COLUMN_PATH.indexOf(current);
+  const to = COLUMN_PATH.indexOf(target);
+  for (let i = from + 1; i <= to; i++) {
+    await taskStore.moveTask(taskId, COLUMN_PATH[i] as never);
+  }
+}
+
+/** Run the work stage so a CE pipeline + its first board task + state record exist. */
+async function landPipeline(stage = "plan"): Promise<{ cePipelineId: string; task: Task }> {
+  // Register-free: drive the WORK stage (which seeds state) but point the link at
+  // `stage` so we can advance through the real stage order. Simplest: use the
+  // work bridge directly via the orchestrator at the work stage, then rewrite the
+  // pipeline state's currentStage to `stage` for ordering tests.
+  const script: InteractiveAiSessionEvent[] = [
+    { type: "complete", data: { artifact: "# log\n", tasks: [{ description: "do stage work" }] } },
+  ];
+  const orch = new CeOrchestrator({
+    ctx,
+    createInteractiveAiSession: vi.fn(async () => ({ session: makeScriptedSession(script) })),
+    projectRoot: rootDir,
+    turnTimeoutMs: 5000,
+  });
+  const started = await orch.start(WORK_STAGE_ID, { openingMessage: "go" });
+  const cePipelineId = started.session.id;
+  const store = getCePipelineStore(ctx);
+
+  if (stage !== WORK_STAGE_ID) {
+    // Reposition both the link stage and the state stage to `stage` so the
+    // pipeline has a non-terminal stage to advance FROM.
+    const links = store.listByPipeline(cePipelineId);
+    const db = taskStore.getDatabase();
+    for (const l of links) {
+      db.prepare(`UPDATE ce_pipeline_links SET ceStageId = ? WHERE id = ?`).run(stage, l.id);
+    }
+    store.upsertState({ cePipelineId, currentStage: stage, status: "running" });
+  }
+  const tasks = await taskStore.listTasks();
+  return { cePipelineId, task: tasks[0] };
+}
+
+describe("U8 inbound hooks (board → pipeline)", () => {
+  it("onTaskMoved only enqueues when the task is CE-linked; ignores unrelated tasks fast", async () => {
+    const { task } = await landPipeline("plan");
+    const store = getCePipelineStore(ctx);
+
+    // Unrelated (non-CE) board task → hook is a no-op (no queue row).
+    const other = await taskStore.createTask({ description: "unrelated work" });
+    await plugin.hooks.onTaskMoved!(other, "triage", "todo", ctx);
+    expect(store.listPendingSync()).toHaveLength(0);
+
+    // CE-linked task move → a queue row is appended synchronously. We do NOT
+    // await the hook (its body is synchronous; awaiting would let the
+    // fired-and-forgotten reconcile drain the row), so we observe the pending
+    // entry the fast path wrote before any deferred work runs.
+    void plugin.hooks.onTaskMoved!(task, "todo", "in-progress", ctx);
+    const pending = store.listPendingSync();
+    expect(pending.length).toBeGreaterThanOrEqual(1);
+    expect(pending.some((p) => p.taskId === task.id && p.reason === "task_moved")).toBe(true);
+  });
+
+  it("the hook handler does NOT advance the pipeline inline (heavy work is deferred)", async () => {
+    const { cePipelineId, task } = await landPipeline("plan");
+    const store = getCePipelineStore(ctx);
+
+    // Move the task to a terminal column then fire ONLY the synchronous part of
+    // the hook. We assert that synchronously the pipeline stage is unchanged —
+    // advancement happens in the (deferred) reconcile, not inline.
+    await moveTo(task.id, "done");
+    const stageBefore = store.getState(cePipelineId)!.currentStage;
+    // Drive the hook but capture state immediately after the synchronous body.
+    const p = plugin.hooks.onTaskMoved!(task, "in-progress", "done", ctx);
+    // The synchronous body has already run (enqueue) but the fired-and-forgotten
+    // reconcile has not been awaited. Inline, the stage must be unchanged.
+    expect(store.getState(cePipelineId)!.currentStage).toBe(stageBefore);
+    // A queue row exists (the fast path did its job).
+    expect(store.listPendingSync().some((q) => q.taskId === task.id)).toBe(true);
+    await p; // let the fire-and-forget settle for clean teardown.
+  });
+
+  it("the hook handler completes well under the 5s budget even with a slow reconciler", async () => {
+    const { task } = await landPipeline("plan");
+    await moveTo(task.id, "done");
+    const start = Date.now();
+    await plugin.hooks.onTaskMoved!(task, "in-progress", "done", ctx);
+    // The hook awaits NOTHING heavy; it returns synchronously-ish.
+    expect(Date.now() - start).toBeLessThan(1000);
+  });
+});
+
+describe("U8 reconciler (convergence + outbound)", () => {
+  it("AE3: a CE task reaching a terminal column advances the pipeline to the next stage with NO manual step", async () => {
+    const { cePipelineId, task } = await landPipeline("plan");
+    const store = getCePipelineStore(ctx);
+    expect(store.getState(cePipelineId)!.currentStage).toBe("plan");
+
+    // Board moves the task to done (the only manual-equivalent action: a normal
+    // board transition). The hook enqueues; reconcile advances.
+    await moveTo(task.id, "done");
+    await plugin.hooks.onTaskCompleted!({ ...task, column: "done" }, ctx);
+    await reconcileCePipelines(ctx);
+
+    // Pipeline advanced plan → work (next in stage order) with no manual step.
+    const state = store.getState(cePipelineId)!;
+    expect(state.currentStage).toBe("work");
+  });
+
+  it("outbound: advancing the pipeline propagates a NEW next-stage board task", async () => {
+    const { cePipelineId, task } = await landPipeline("plan");
+    const before = (await taskStore.listTasks()).length;
+
+    await moveTo(task.id, "in-review");
+    await reconcileCePipelines(ctx); // no hook fired — pure re-derivation.
+
+    const after = await taskStore.listTasks();
+    expect(after.length).toBe(before + 1);
+    const newTask = after.find((t) => t.id !== task.id)!;
+    const meta = newTask.sourceMetadata as Record<string, unknown>;
+    expect(meta.pluginId).toBe(CE_PLUGIN_ID);
+    expect(meta.cePipelineId).toBe(cePipelineId);
+    expect(meta.ceStageId).toBe("work");
+    // The pipeline is now awaiting the new board task.
+    expect(getCePipelineStore(ctx).getState(cePipelineId)!.status).toBe("awaiting_board");
+  });
+
+  it("MISSED HOOK EVENT → the reconcile sweep still converges (no queue row needed)", async () => {
+    const { cePipelineId, task } = await landPipeline("plan");
+    const store = getCePipelineStore(ctx);
+
+    // Simulate a DROPPED hook: move the board task to a terminal column but do
+    // NOT call any hook and do NOT enqueue anything.
+    await moveTo(task.id, "done");
+    expect(store.listPendingSync()).toHaveLength(0); // nothing was enqueued.
+    expect(store.getState(cePipelineId)!.currentStage).toBe("plan"); // not advanced yet.
+
+    // The on-demand sweep re-derives the transition from board truth alone.
+    const result = await new CeReconciler(ctx).reconcile();
+    expect(result.advanced).toBe(1);
+    expect(store.getState(cePipelineId)!.currentStage).toBe("work");
+  });
+
+  it("reconcile is idempotent: a second sweep does not double-advance or duplicate tasks", async () => {
+    const { cePipelineId, task } = await landPipeline("plan");
+    await moveTo(task.id, "done");
+    await reconcileCePipelines(ctx);
+    const afterFirst = (await taskStore.listTasks()).length;
+    const stageFirst = getCePipelineStore(ctx).getState(cePipelineId)!.currentStage;
+
+    await reconcileCePipelines(ctx);
+    expect((await taskStore.listTasks()).length).toBe(afterFirst);
+    expect(getCePipelineStore(ctx).getState(cePipelineId)!.currentStage).toBe(stageFirst);
+  });
+
+  it("partial completion does not advance: pipeline stays running until ALL current-stage tasks are terminal", async () => {
+    const { cePipelineId, task } = await landPipeline("plan");
+    const store = getCePipelineStore(ctx);
+    // Add a second current-stage task to the SAME pipeline/stage.
+    const t2 = await taskStore.createTask({ description: "second plan task" });
+    store.createLink({ taskId: t2.id, cePipelineId, ceStageId: "plan", ceArtifactPath: null });
+
+    await moveTo(task.id, "done"); // only one terminal.
+    await reconcileCePipelines(ctx);
+    expect(store.getState(cePipelineId)!.currentStage).toBe("plan"); // not advanced.
+
+    await moveTo(t2.id, "done"); // now both terminal.
+    await reconcileCePipelines(ctx);
+    expect(store.getState(cePipelineId)!.currentStage).toBe("work"); // advanced.
+  });
+});
+
+describe("U8 conflict resolution (board vs CE authority)", () => {
+  it("simultaneous board move + CE advance: board keeps the task column, CE keeps the pipeline content", async () => {
+    const { cePipelineId, task } = await landPipeline("plan");
+    const store = getCePipelineStore(ctx);
+
+    // CE-flow side: the pipeline owns its content; record an artifact (CE-authoritative).
+    store.transitionState(cePipelineId, { lastArtifactPath: "/docs/plans/p.md" });
+
+    // Board side: move the task to done (board-authoritative for the column).
+    await moveTo(task.id, "done");
+
+    // Reconcile resolves the collision: it READS the board column (never rewrites
+    // the terminal task) and WRITES only CE-owned fields + a NEW task.
+    await reconcileCePipelines(ctx);
+
+    // Board authority: the original task's column is exactly what the board set.
+    const reread = await taskStore.getTask(task.id);
+    expect(reread!.column).toBe("done");
+
+    // CE authority: the pipeline content (stage + artifact) is what CE wrote.
+    const state = store.getState(cePipelineId)!;
+    expect(state.currentStage).toBe("work");
+    expect(state.lastArtifactPath).toBe("/docs/plans/p.md");
+
+    // The new outbound task is a fresh row — the writers never contended on one cell.
+    const tasks = await taskStore.listTasks();
+    const next = tasks.find((t) => t.id !== task.id)!;
+    expect((next.sourceMetadata as Record<string, unknown>).ceStageId).toBe("work");
+  });
+});
diff --git a/plugins/fusion-plugin-compound-engineering/src/index.ts b/plugins/fusion-plugin-compound-engineering/src/index.ts
index 87900c06b4..4788aa0d91 100644
--- a/plugins/fusion-plugin-compound-engineering/src/index.ts
+++ b/plugins/fusion-plugin-compound-engineering/src/index.ts
@@ -4,6 +4,8 @@ import { installBundledCeSkills } from "./skill-installation.js";
 import { ensureCeSchema } from "./schema.js";
 import { createSessionRoutes } from "./routes/session-routes.js";
 import { createArtifactRoutes } from "./routes/artifact-routes.js";
+import { getCePipelineStore } from "./sync/pipeline-store.js";
+import { reconcileCePipelines } from "./sync/reconciler.js";
 
 export { CompoundEngineeringDashboardView } from "./dashboard-view.js";
 export { COMPOUND_ENGINEERING_SKILLS } from "./skills.js";
@@ -16,7 +18,16 @@ export {
 export { ensureCeSchema } from "./schema.js";
 export { CeSessionStore, getCeSessionStore } from "./session/session-store.js";
 export { CePipelineStore, getCePipelineStore } from "./sync/pipeline-store.js";
-export type { CePipelineLink, CreateCePipelineLinkInput } from "./sync/pipeline-store.js";
+export type {
+  CePipelineLink,
+  CreateCePipelineLinkInput,
+  CePipelineState,
+  CePipelineStatus,
+  CeSyncQueueEntry,
+  CeSyncReason,
+} from "./sync/pipeline-store.js";
+export { CeReconciler, reconcileCePipelines } from "./sync/reconciler.js";
+export type { ReconcileResult } from "./sync/reconciler.js";
 export {
   CeOrchestrator,
   WORK_STAGE_ID,
@@ -41,6 +52,42 @@ const plugin = definePlugin({
     // Idempotent DDL for the plugin-local CE tables (ce_sessions). Runs against
     // the same DB route handlers reach via ctx.taskStore.getDatabase() (U5).
     onSchemaInit: ensureCeSchema,
+    // INBOUND board→pipeline sync (U8 / FN-5719). The 5s hook budget
+    // (plugin-runner invokeHookSafe) means these MUST be fast: resolve the link,
+    // ENQUEUE a sync signal, and return. Heavy advancement (board reads, outbound
+    // task creation) happens in the reconciler, NOT inline here. A reconcile
+    // drain is fired-and-forgotten (never awaited) so a slow sweep cannot blow
+    // the hook budget; correctness does not depend on it firing because the next
+    // reconcile() sweep re-derives the transition from board truth.
+    onTaskMoved: (task, fromColumn, toColumn, ctx) => {
+      const store = getCePipelineStore(ctx);
+      const link = store.findByTaskId(task.id);
+      if (!link) return; // not a CE-linked task → ignore fast.
+      store.enqueueSync({
+        cePipelineId: link.cePipelineId,
+        taskId: task.id,
+        reason: "task_moved",
+        fromColumn,
+        toColumn,
+      });
+      void Promise.resolve()
+        .then(() => reconcileCePipelines(ctx))
+        .catch((err) => ctx.logger.warn(`CE reconcile (onTaskMoved) failed: ${String(err)}`));
+    },
+    onTaskCompleted: (task, ctx) => {
+      const store = getCePipelineStore(ctx);
+      const link = store.findByTaskId(task.id);
+      if (!link) return;
+      store.enqueueSync({
+        cePipelineId: link.cePipelineId,
+        taskId: task.id,
+        reason: "task_completed",
+        toColumn: "done",
+      });
+      void Promise.resolve()
+        .then(() => reconcileCePipelines(ctx))
+        .catch((err) => ctx.logger.warn(`CE reconcile (onTaskCompleted) failed: ${String(err)}`));
+    },
     // Install the bundled, pinned ce-* SKILL.md files into a plugin-local,
     // discoverable directory on load. The engine ingests
     // PluginSkillContribution only as a name; physical discovery requires the
diff --git a/plugins/fusion-plugin-compound-engineering/src/schema.ts b/plugins/fusion-plugin-compound-engineering/src/schema.ts
index b2bf15102c..5d9d6f61b5 100644
--- a/plugins/fusion-plugin-compound-engineering/src/schema.ts
+++ b/plugins/fusion-plugin-compound-engineering/src/schema.ts
@@ -25,6 +25,20 @@ import type { Database } from "@fusion/core";
  * board-task ownership and CE-pipeline ownership stay separate state machines.
  * U7 keeps it minimal (link records only); U8 extends it with the bidirectional
  * pipeline-state machine.
+ *
+ * `ce_pipeline_state` (U8) is the CE-pipeline's OWN state machine — DISTINCT from
+ * board-task column state (KTD4 / FN-5719: two separate ownership state machines,
+ * never one shared column encoding two concerns). It tracks where the pipeline
+ * itself is: `currentStage` (the CE stage the pipeline has reached) and `status`
+ * (`running` | `advancing` | `awaiting_board` | `completed`). The board owns task
+ * columns; this table owns pipeline progress. They are reconciled — never merged.
+ *
+ * `ce_pipeline_sync_queue` (U8) is the event-enqueue seam (FN-5719). Lifecycle
+ * hooks write a row here FAST (5s hook budget) and return; the reconciler drains
+ * it. A dropped/never-enqueued event is still recovered because the reconciler
+ * ALSO re-derives transitions from board state — the queue is an optimization,
+ * board+state comparison is the convergence guarantee. `processedAt NULL` =
+ * pending; non-null = drained (kept for audit, swept idempotently).
  */
 export function ensureCeSchema(db: Database): void {
   db.exec(`
@@ -68,5 +82,36 @@ export function ensureCeSchema(db: Database): void {
 
     CREATE UNIQUE INDEX IF NOT EXISTS idxCePipelineLinksTask
       ON ce_pipeline_links(taskId);
+
+    CREATE TABLE IF NOT EXISTS ce_pipeline_state (
+      cePipelineId TEXT PRIMARY KEY,
+      currentStage TEXT NOT NULL,
+      status TEXT NOT NULL CHECK (status IN (
+        'running','advancing','awaiting_board','completed'
+      )),
+      lastArtifactPath TEXT,
+      createdAt TEXT NOT NULL,
+      updatedAt TEXT NOT NULL
+    );
+
+    CREATE INDEX IF NOT EXISTS idxCePipelineStateStatus
+      ON ce_pipeline_state(status, updatedAt DESC, cePipelineId);
+
+    CREATE TABLE IF NOT EXISTS ce_pipeline_sync_queue (
+      id TEXT PRIMARY KEY,
+      cePipelineId TEXT NOT NULL,
+      taskId TEXT NOT NULL,
+      reason TEXT NOT NULL,
+      fromColumn TEXT,
+      toColumn TEXT,
+      enqueuedAt TEXT NOT NULL,
+      processedAt TEXT
+    );
+
+    CREATE INDEX IF NOT EXISTS idxCePipelineSyncQueuePending
+      ON ce_pipeline_sync_queue(processedAt, enqueuedAt, id);
+
+    CREATE INDEX IF NOT EXISTS idxCePipelineSyncQueuePipeline
+      ON ce_pipeline_sync_queue(cePipelineId, enqueuedAt, id);
   `);
 }
diff --git a/plugins/fusion-plugin-compound-engineering/src/session/orchestrator.ts b/plugins/fusion-plugin-compound-engineering/src/session/orchestrator.ts
index c02eb2a7c1..efc7d4c7a6 100644
--- a/plugins/fusion-plugin-compound-engineering/src/session/orchestrator.ts
+++ b/plugins/fusion-plugin-compound-engineering/src/session/orchestrator.ts
@@ -361,6 +361,16 @@ export class CeOrchestrator {
     const ceStageId = session.stage;
     const ceArtifactPath = session.artifactPath ?? null;
 
+    // Seed the CE-pipeline STATE record (U8). This is the pipeline's OWN state
+    // machine, distinct from the board task columns it will spawn. The pipeline
+    // is "running" at this stage until a board signal advances it.
+    this.pipelineStore.upsertState({
+      cePipelineId,
+      currentStage: ceStageId,
+      status: "running",
+      lastArtifactPath: ceArtifactPath,
+    });
+
     for (const spec of specs) {
       const description = spec.description.trim();
       if (!description) continue; // createTask rejects blank descriptions.
diff --git a/plugins/fusion-plugin-compound-engineering/src/sync/pipeline-store.ts b/plugins/fusion-plugin-compound-engineering/src/sync/pipeline-store.ts
index 3e2f57dca3..3ba24437c1 100644
--- a/plugins/fusion-plugin-compound-engineering/src/sync/pipeline-store.ts
+++ b/plugins/fusion-plugin-compound-engineering/src/sync/pipeline-store.ts
@@ -49,6 +49,106 @@ export interface CreateCePipelineLinkInput {
   id?: string;
 }
 
+/**
+ * CE-pipeline STATUS — the pipeline's own lifecycle, DISTINCT from board-task
+ * columns (KTD4 / FN-5719). The board owns task columns; this owns pipeline
+ * progress. The two are never encoded in one shared column.
+ *
+ *   running        — pipeline is at `currentStage`, work in flight on the board.
+ *   advancing      — a board signal arrived; reconciler is moving it on / feeding
+ *                    the next stage (transient, set inside the reconciler sweep).
+ *   awaiting_board — pipeline advanced and is waiting on board task(s) again.
+ *   completed      — pipeline reached its terminal stage and finished.
+ */
+export type CePipelineStatus = "running" | "advancing" | "awaiting_board" | "completed";
+
+/** The CE-pipeline's own state record (separate state machine from the board). */
+export interface CePipelineState {
+  cePipelineId: string;
+  /** The CE stage the pipeline has reached (a stage id, e.g. "work"). */
+  currentStage: string;
+  status: CePipelineStatus;
+  /** Last artifact the pipeline produced/propagated (CE-authoritative content). */
+  lastArtifactPath: string | null;
+  createdAt: string;
+  updatedAt: string;
+}
+
+interface CePipelineStateRow {
+  cePipelineId: string;
+  currentStage: string;
+  status: CePipelineStatus;
+  lastArtifactPath: string | null;
+  createdAt: string;
+  updatedAt: string;
+}
+
+export interface UpsertCePipelineStateInput {
+  cePipelineId: string;
+  currentStage: string;
+  status?: CePipelineStatus;
+  lastArtifactPath?: string | null;
+}
+
+/** Why a board change was enqueued for the pipeline (audit + reconcile routing). */
+export type CeSyncReason = "task_moved" | "task_completed" | "reconcile";
+
+/** A pending (or drained) board→pipeline sync signal. */
+export interface CeSyncQueueEntry {
+  id: string;
+  cePipelineId: string;
+  taskId: string;
+  reason: CeSyncReason;
+  fromColumn: string | null;
+  toColumn: string | null;
+  enqueuedAt: string;
+  processedAt: string | null;
+}
+
+interface CeSyncQueueRow {
+  id: string;
+  cePipelineId: string;
+  taskId: string;
+  reason: CeSyncReason;
+  fromColumn: string | null;
+  toColumn: string | null;
+  enqueuedAt: string;
+  processedAt: string | null;
+}
+
+export interface EnqueueSyncInput {
+  cePipelineId: string;
+  taskId: string;
+  reason: CeSyncReason;
+  fromColumn?: string | null;
+  toColumn?: string | null;
+  id?: string;
+}
+
+function rowToState(row: CePipelineStateRow): CePipelineState {
+  return {
+    cePipelineId: row.cePipelineId,
+    currentStage: row.currentStage,
+    status: row.status,
+    lastArtifactPath: row.lastArtifactPath,
+    createdAt: row.createdAt,
+    updatedAt: row.updatedAt,
+  };
+}
+
+function rowToQueueEntry(row: CeSyncQueueRow): CeSyncQueueEntry {
+  return {
+    id: row.id,
+    cePipelineId: row.cePipelineId,
+    taskId: row.taskId,
+    reason: row.reason,
+    fromColumn: row.fromColumn,
+    toColumn: row.toColumn,
+    enqueuedAt: row.enqueuedAt,
+    processedAt: row.processedAt,
+  };
+}
+
 function rowToLink(row: CePipelineLinkRow): CePipelineLink {
   return {
     id: row.id,
@@ -109,6 +209,115 @@ export class CePipelineStore {
       .get(taskId) as CePipelineLinkRow | undefined;
     return row ? rowToLink(row) : undefined;
   }
+
+  // ── CE-pipeline STATE machine (U8) ───────────────────────────────────
+  // Separate from board-task columns: this table is the pipeline's OWN state.
+
+  /** Read a pipeline's own state record. */
+  getState(cePipelineId: string): CePipelineState | undefined {
+    const row = this.db
+      .prepare(`SELECT * FROM ce_pipeline_state WHERE cePipelineId = ?`)
+      .get(cePipelineId) as CePipelineStateRow | undefined;
+    return row ? rowToState(row) : undefined;
+  }
+
+  /** All pipeline state records (the reconciler sweeps every one). */
+  listAllState(): CePipelineState[] {
+    const rows = this.db
+      .prepare(`SELECT * FROM ce_pipeline_state ORDER BY updatedAt DESC, cePipelineId`)
+      .all() as CePipelineStateRow[];
+    return rows.map(rowToState);
+  }
+
+  /**
+   * Create or update a pipeline's state. Idempotent on `cePipelineId`. `status`
+   * defaults to `running` on first write and is preserved on update unless given.
+   */
+  upsertState(input: UpsertCePipelineStateInput): CePipelineState {
+    const now = new Date().toISOString();
+    const existing = this.getState(input.cePipelineId);
+    const status = input.status ?? existing?.status ?? "running";
+    const lastArtifactPath =
+      input.lastArtifactPath !== undefined ? input.lastArtifactPath : existing?.lastArtifactPath ?? null;
+    if (existing) {
+      this.db
+        .prepare(
+          `UPDATE ce_pipeline_state
+             SET currentStage = ?, status = ?, lastArtifactPath = ?, updatedAt = ?
+           WHERE cePipelineId = ?`,
+        )
+        .run(input.currentStage, status, lastArtifactPath, now, input.cePipelineId);
+    } else {
+      this.db
+        .prepare(
+          `INSERT INTO ce_pipeline_state
+             (cePipelineId, currentStage, status, lastArtifactPath, createdAt, updatedAt)
+           VALUES (?, ?, ?, ?, ?, ?)`,
+        )
+        .run(input.cePipelineId, input.currentStage, status, lastArtifactPath, now, now);
+    }
+    return this.getState(input.cePipelineId)!;
+  }
+
+  /**
+   * Transition a pipeline to a new stage/status. Returns the updated state, or
+   * `undefined` if the pipeline has no state row yet (caller should seed first).
+   */
+  transitionState(
+    cePipelineId: string,
+    next: { currentStage?: string; status?: CePipelineStatus; lastArtifactPath?: string | null },
+  ): CePipelineState | undefined {
+    const existing = this.getState(cePipelineId);
+    if (!existing) return undefined;
+    return this.upsertState({
+      cePipelineId,
+      currentStage: next.currentStage ?? existing.currentStage,
+      status: next.status ?? existing.status,
+      lastArtifactPath:
+        next.lastArtifactPath !== undefined ? next.lastArtifactPath : existing.lastArtifactPath,
+    });
+  }
+
+  // ── Event-enqueue seam (U8 / FN-5719) ────────────────────────────────
+  // Hooks write here FAST and return; the reconciler drains. A missed enqueue is
+  // still recovered because reconcile() re-derives from board state too.
+
+  /** Append a pending board→pipeline sync signal. Fast, append-only. */
+  enqueueSync(input: EnqueueSyncInput): CeSyncQueueEntry {
+    const entry: CeSyncQueueEntry = {
+      id: input.id ?? randomUUID(),
+      cePipelineId: input.cePipelineId,
+      taskId: input.taskId,
+      reason: input.reason,
+      fromColumn: input.fromColumn ?? null,
+      toColumn: input.toColumn ?? null,
+      enqueuedAt: new Date().toISOString(),
+      processedAt: null,
+    };
+    this.db
+      .prepare(
+        `INSERT INTO ce_pipeline_sync_queue
+           (id, cePipelineId, taskId, reason, fromColumn, toColumn, enqueuedAt, processedAt)
+         VALUES (?, ?, ?, ?, ?, ?, ?, NULL)`,
+      )
+      .run(entry.id, entry.cePipelineId, entry.taskId, entry.reason, entry.fromColumn, entry.toColumn, entry.enqueuedAt);
+    return entry;
+  }
+
+  /** All pending (un-drained) queue entries, oldest first. */
+  listPendingSync(): CeSyncQueueEntry[] {
+    const rows = this.db
+      .prepare(`SELECT * FROM ce_pipeline_sync_queue WHERE processedAt IS NULL ORDER BY enqueuedAt, id`)
+      .all() as CeSyncQueueRow[];
+    return rows.map(rowToQueueEntry);
+  }
+
+  /** Mark a queue entry drained (idempotent). */
+  markSyncProcessed(id: string): void {
+    this.db
+      .prepare(`UPDATE ce_pipeline_sync_queue SET processedAt = ? WHERE id = ? AND processedAt IS NULL`)
+      .run(new Date().toISOString(), id);
+  }
 }
 
 const storeCache = new WeakMap<object, CePipelineStore>();
diff --git a/plugins/fusion-plugin-compound-engineering/src/sync/reconciler.ts b/plugins/fusion-plugin-compound-engineering/src/sync/reconciler.ts
new file mode 100644
index 0000000000..58da94dcb8
--- /dev/null
+++ b/plugins/fusion-plugin-compound-engineering/src/sync/reconciler.ts
@@ -0,0 +1,246 @@
+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 {
+  getCePipelineStore,
+  type CePipelineLink,
+  type CePipelineState,
+  type CePipelineStore,
+} from "./pipeline-store.js";
+
+/**
+ * BIDIRECTIONAL SYNC RECONCILER (U8 / FN-5719 pattern).
+ *
+ * Two SEPARATE state machines are kept in sync, never merged (KTD4):
+ *   - Board-task ownership  → the task's `column` (board is authoritative).
+ *   - CE-pipeline ownership → `ce_pipeline_state.{currentStage,status}` (CE flow
+ *                             is authoritative for artifact/pipeline content).
+ *
+ * INBOUND (board → pipeline): the lifecycle hooks (`onTaskMoved`/`onTaskCompleted`
+ * in index.ts) do the MINIMUM under the 5s hook budget — resolve the link and
+ * `enqueueSync(...)`, then return. They do NOT advance the pipeline inline.
+ *
+ * RECONCILE (the convergence guarantee): `reconcile()` is a single on-demand
+ * sweep — NOT a tight interval poll (per docs/performance/dashboard-load.md).
+ * It (1) drains the queue and (2) INDEPENDENTLY re-derives transitions by
+ * comparing live board state (`ctx.taskStore`) against pipeline state. Step (2)
+ * is why a DROPPED or never-enqueued hook event still converges: the queue is an
+ * optimization; the board↔state comparison is the source of truth.
+ *
+ * OUTBOUND (pipeline → board): when a pipeline advances to a stage that produces
+ * board work, the reconciler creates the next-stage board task via
+ * `ctx.taskStore.createTask` and links it — propagating the CE-flow change onto
+ * the board.
+ *
+ * TRIGGER MODEL (honest about the host seam): there is NO host scheduler wired to
+ * call this on a timer. In production the sweep is invoked (a) right after the
+ * hooks enqueue (a cheap drain on the same board mutation that triggered the
+ * hook), and (b) on demand from a route (U9 settings/refresh surface) or on a
+ * dashboard session-change. Because step (2) re-derives from board truth, any
+ * single missed trigger is recovered on the NEXT sweep — no continuous poll loop
+ * is needed for correctness.
+ */
+
+/** Columns that mean "this stage's board work is finished" → advance the pipeline. */
+const TERMINAL_COLUMNS = new Set(["in-review", "done"]);
+
+export interface ReconcileResult {
+  /** Queue entries drained this sweep. */
+  drained: number;
+  /** Pipelines whose state advanced this sweep. */
+  advanced: number;
+  /** Board tasks created outbound this sweep (next-stage propagation). */
+  tasksCreated: number;
+  /** Pipelines inspected. */
+  inspected: number;
+}
+
+/** The linear CE stage order. The pipeline advances along this sequence. */
+function stageOrder(): string[] {
+  return listStages().map((s) => s.stageId);
+}
+
+/** The stage AFTER `stageId` in the pipeline, or `undefined` if it's terminal. */
+function nextStageAfter(stageId: string): string | undefined {
+  const order = stageOrder();
+  const idx = order.indexOf(stageId);
+  if (idx < 0 || idx >= order.length - 1) return undefined;
+  return order[idx + 1];
+}
+
+export class CeReconciler {
+  private readonly ctx: PluginContext;
+  private readonly store: CePipelineStore;
+
+  constructor(ctx: PluginContext) {
+    this.ctx = ctx;
+    this.store = getCePipelineStore(ctx);
+  }
+
+  /**
+   * Drain the queue AND re-derive missed transitions from live board state, then
+   * apply any pipeline advancement (with outbound board propagation). Idempotent:
+   * running it twice is a no-op once everything has converged.
+   */
+  async reconcile(): Promise<ReconcileResult> {
+    const result: ReconcileResult = { drained: 0, advanced: 0, tasksCreated: 0, inspected: 0 };
+
+    // (1) Drain the queue. Draining is just an audit/ack — the actual decision is
+    // 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++;
+    }
+
+    // (2) Convergence sweep: inspect EVERY pipeline that has state, not only the
+    // 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++;
+      const advanced = await this.reconcileOne(state);
+      if (advanced) {
+        result.advanced++;
+        if (advanced.created) result.tasksCreated++;
+      }
+    }
+    return result;
+  }
+
+  /**
+   * Re-derive whether ONE pipeline should advance by reading the live board
+   * column of its current-stage task(s). Board is authoritative for task state;
+   * we never write the task column from here for the current stage.
+   */
+  private async reconcileOne(
+    state: CePipelineState,
+  ): 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;
+
+    const tasks = await this.loadTasks(links);
+    if (tasks.length === 0) return undefined;
+
+    // Advancement rule: every current-stage board task has reached a terminal
+    // column (board-authoritative read). Partial completion keeps it running.
+    const allTerminal = tasks.every((t) => t && TERMINAL_COLUMNS.has(t.column));
+    if (!allTerminal) {
+      // Still running on the board — make sure our status reflects that and stop.
+      if (state.status !== "running") {
+        this.store.transitionState(state.cePipelineId, { status: "running" });
+      }
+      return undefined;
+    }
+
+    const next = nextStageAfter(state.currentStage);
+    if (!next) {
+      // Terminal stage finished → pipeline completed. No outbound task.
+      this.store.transitionState(state.cePipelineId, { status: "completed" });
+      this.ctx.emitEvent("compound-engineering:pipeline-completed", {
+        cePipelineId: state.cePipelineId,
+        stage: state.currentStage,
+      });
+      return { created: false };
+    }
+
+    // CONFLICT POLICY (explicit): board is authoritative for the task columns we
+    // just READ (we never rewrote them); CE flow is authoritative for the
+    // pipeline content we WRITE (currentStage, artifact, the next-stage task).
+    // 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);
+    return { created };
+  }
+
+  /**
+   * Advance the pipeline to `nextStage` (CE-owned write) and propagate OUTBOUND
+   * 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> {
+    // 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);
+
+    this.store.transitionState(state.cePipelineId, {
+      currentStage: nextStage,
+      status: already ? "running" : "awaiting_board",
+    });
+
+    if (already) return false;
+
+    const task = await this.ctx.taskStore.createTask({
+      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" });
+
+    this.ctx.emitEvent("compound-engineering:pipeline-advanced", {
+      cePipelineId: state.cePipelineId,
+      fromStage: state.currentStage,
+      toStage: nextStage,
+      taskId: task.id,
+    });
+    return true;
+  }
+
+  /** Load the live board tasks for a set of links (board-authoritative read). */
+  private async loadTasks(links: CePipelineLink[]): Promise<Array<Task | undefined>> {
+    const out: Array<Task | undefined> = [];
+    for (const link of links) {
+      try {
+        const task = await this.ctx.taskStore.getTask(link.taskId);
+        out.push(task ?? undefined);
+      } catch {
+        // A deleted/missing task is treated as absent, not terminal.
+        out.push(undefined);
+      }
+    }
+    return out;
+  }
+}
+
+/**
+ * Convenience: build a reconciler and run one sweep. This is the entry point a
+ * route handler or post-hook drain calls.
+ */
+export async function reconcileCePipelines(ctx: PluginContext): Promise<ReconcileResult> {
+  return new CeReconciler(ctx).reconcile();
+}