fix(compound-engineering): address code review feedback

- session-store: rowToSession parses JSON columns via a safeParse helper, so a
  corrupted currentQuestion/conversationHistory column degrades to null/[] instead
  of throwing and crashing reads of an otherwise-valid row (+regression test)
- session-routes: validate the ?status= list filter against CE_SESSION_STATUSES
  (asCeSessionStatus) instead of casting an arbitrary query string with 'as never'
- _harness: makeScriptedSession throws on an empty script rather than yielding
  undefined, surfacing test mistakes loudly
- useCeSession tests: use fake timers (advanceTimersByTimeAsync) instead of real
  setTimeout waits for deterministic poll-interval assertions

Skipped: 3 doc nits in src/skills/ce-*/references/** — those are pinned upstream
ce-* skill copies (KTD5 vendored snapshot), not this repo's content.

plugins/fusion-plugin-compound-engineering/src/__tests__/_harness.ts

@@ -68,7 +68,14 @@ export function makeScriptedSession(script: InteractiveAiSessionEvent[]): Intera
     answer: vi.fn(async () => {
       cursor++;
     }),
-    nextEvent: vi.fn(async () => script[Math.min(cursor, script.length - 1)]),
+    nextEvent: vi.fn(async () => {
+      if (script.length === 0) {
+        // An empty script is a test bug — surface it loudly rather than
+        // silently returning undefined (which masks the mistake downstream).
+        throw new Error("makeScriptedSession: empty script has no events to yield");
+      }
+      return script[Math.min(Math.max(cursor, 0), script.length - 1)];
+    }),
     dispose: vi.fn(),
   };
 }

plugins/fusion-plugin-compound-engineering/src/__tests__/session-store.test.ts

@@ -103,3 +103,33 @@ describe("interval-relative staleness (FN-4172 rubric)", () => {
     expect(store.get(stuck.id)!.status).toBe("interrupted");
   });
 });
+
+describe("corrupt-JSON resilience + status validation", () => {
+  it("degrades gracefully when a JSON column is corrupted (no throw)", () => {
+    const store = new CeSessionStore(h.db);
+    const s = store.create({ stage: "brainstorm" });
+    // Corrupt both JSON columns directly in the DB.
+    h.db
+      .prepare("UPDATE ce_sessions SET currentQuestion = ?, conversationHistory = ? WHERE id = ?")
+      .run("{not valid json", "also not json", s.id);
+
+    // Reading the row must not throw; corrupt fields fall back to null / [].
+    const read = store.get(s.id)!;
+    expect(read.id).toBe(s.id);
+    expect(read.currentQuestion).toBeNull();
+    expect(read.conversationHistory).toEqual([]);
+    // The rest of the row still surfaces the session's real state.
+    expect(read.stage).toBe("brainstorm");
+  });
+});
+
+describe("asCeSessionStatus validation", () => {
+  it("accepts valid statuses and rejects anything else", async () => {
+    const { asCeSessionStatus } = await import("../session/session-store.js");
+    expect(asCeSessionStatus("active")).toBe("active");
+    expect(asCeSessionStatus("interrupted")).toBe("interrupted");
+    expect(asCeSessionStatus("bogus")).toBeUndefined();
+    expect(asCeSessionStatus("")).toBeUndefined();
+    expect(asCeSessionStatus(undefined)).toBeUndefined();
+  });
+});

plugins/fusion-plugin-compound-engineering/src/dashboard/hooks/__tests__/useCeSession.test.tsx

@@ -1,4 +1,4 @@
-import { describe, expect, it, vi } from "vitest";
+import { afterEach, describe, expect, it, vi } from "vitest";
 import { act, render, screen } from "@testing-library/react";
 import type { PlanningQuestion } from "@fusion/core";
 import { useCeSession, type CeSessionTransport, type CeSessionSubscribe } from "../useCeSession.js";
@@ -40,6 +40,11 @@ function Harness({ transport }: { transport: CeSessionTransport }) {
 }
 
 describe("useCeSession lifecycle", () => {
+  afterEach(() => {
+    // Ensure faked timers never leak into the next test.
+    vi.useRealTimers();
+  });
+
   it("start → awaiting_input → answer → completed", async () => {
     const transport: CeSessionTransport = {
       start: vi.fn(async () => mkSession({ status: "awaiting_input", currentQuestion: Q })),
@@ -64,6 +69,7 @@ describe("useCeSession lifecycle", () => {
   });
 
   it("threads the start projectId through resume and poll", async () => {
+    vi.useFakeTimers();
     const get = vi.fn(async () => mkSession({ status: "active" }));
     const transport: CeSessionTransport = {
       start: vi.fn(async () => mkSession({ status: "interrupted", currentQuestion: Q })),
@@ -79,14 +85,16 @@ describe("useCeSession lifecycle", () => {
       screen.getByText("resume").click();
     });
     expect(transport.resume).toHaveBeenCalledWith("s1", "p1");
-    // The poll (active status) must also carry the projectId.
+    // The poll (active status) must also carry the projectId. Harness uses a 5ms
+    // interval; advance fake time deterministically past one tick.
     await act(async () => {
-      await new Promise((r) => setTimeout(r, 20));
+      await vi.advanceTimersByTimeAsync(20);
     });
     expect(get).toHaveBeenCalledWith("s1", "p1");
   });
 
   it("polls while active and stops once settled", async () => {
+    vi.useFakeTimers();
     let calls = 0;
     const get = vi.fn(async () => {
       calls += 1;
@@ -104,9 +112,9 @@ describe("useCeSession lifecycle", () => {
     });
     expect(screen.getByTestId("status")).toHaveTextContent("active");
 
-    // Let the poll interval fire and converge to awaiting_input.
+    // Advance fake time so the poll interval fires and converges to awaiting_input.
     await act(async () => {
-      await new Promise((r) => setTimeout(r, 40));
+      await vi.advanceTimersByTimeAsync(40);
     });
     expect(get).toHaveBeenCalled();
     expect(screen.getByTestId("status")).toHaveTextContent("awaiting_input");

plugins/fusion-plugin-compound-engineering/src/routes/session-routes.ts

@@ -1,6 +1,6 @@
 import type { PluginContext, PluginRouteDefinition, PluginRouteResponse } from "@fusion/core";
 import { CeOrchestrator } from "../session/orchestrator.js";
-import { getCeSessionStore } from "../session/session-store.js";
+import { asCeSessionStatus, getCeSessionStore } from "../session/session-store.js";
 import { getCePipelineStore } from "../sync/pipeline-store.js";
 import { asString } from "./route-helpers.js";
 
@@ -122,9 +122,9 @@ export function createSessionRoutes(): PluginRouteDefinition[] {
       description: "List CE sessions (optionally filtered by status/stage).",
       handler: async (req: unknown, ctx: PluginContext): Promise<PluginRouteResponse> => {
         const query = (req as RouteRequest).query ?? {};
-        const status = typeof query.status === "string" ? query.status : undefined;
+        const status = asCeSessionStatus(typeof query.status === "string" ? query.status : undefined);
         const stage = typeof query.stage === "string" ? query.stage : undefined;
-        const sessions = getCeSessionStore(ctx).list({ status: status as never, stage });
+        const sessions = getCeSessionStore(ctx).list({ status, stage });
         return { status: 200, body: { sessions } };
       },
     },

plugins/fusion-plugin-compound-engineering/src/session/session-store.ts

@@ -7,13 +7,23 @@ import { ensureCeSchema } from "../schema.js";
  * launching → active → awaiting_input ↔ active → completed | error | interrupted;
  * interrupted/error → active on resume/retry.
  */
-export type CeSessionStatus =
-  | "launching"
-  | "active"
-  | "awaiting_input"
-  | "completed"
-  | "error"
-  | "interrupted";
+export const CE_SESSION_STATUSES = [
+  "launching",
+  "active",
+  "awaiting_input",
+  "completed",
+  "error",
+  "interrupted",
+] as const;
+
+export type CeSessionStatus = (typeof CE_SESSION_STATUSES)[number];
+
+/** Narrow an arbitrary string (e.g. a query param) to a valid status, else undefined. */
+export function asCeSessionStatus(value: string | undefined): CeSessionStatus | undefined {
+  return value && (CE_SESSION_STATUSES as readonly string[]).includes(value)
+    ? (value as CeSessionStatus)
+    : undefined;
+}
 
 /** A single recorded turn in the conversation history (for resume). */
 export interface CeConversationTurn {
@@ -72,13 +82,26 @@ export const STALE_INTERVAL_MULTIPLE = 3;
 
 const DEFAULT_TURN_INTERVAL_MS = 120000;
 
+/** Parse a JSON column, falling back to `fallback` on corruption rather than throwing. */
+function safeParse<T>(raw: string | null, fallback: T): T {
+  if (!raw) return fallback;
+  try {
+    return JSON.parse(raw) as T;
+  } catch {
+    // A corrupted JSON column must not crash reads of an otherwise-valid row
+    // (and must not destroy the rest of the session). Degrade to the fallback;
+    // the row's status/error still surface the session's real state.
+    return fallback;
+  }
+}
+
 function rowToSession(row: CeSessionRow): CeSession {
   return {
     id: row.id,
     stage: row.stage,
     status: row.status,
-    currentQuestion: row.currentQuestion ? (JSON.parse(row.currentQuestion) as PlanningQuestion) : null,
-    conversationHistory: JSON.parse(row.conversationHistory) as CeConversationTurn[],
+    currentQuestion: safeParse<PlanningQuestion | null>(row.currentQuestion, null),
+    conversationHistory: safeParse<CeConversationTurn[]>(row.conversationHistory, []),
     projectId: row.projectId,
     artifactPath: row.artifactPath,
     error: row.error,