Add pipeline REST routes

Co-Authored-By: Paperclip <noreply@paperclip.ing>

server/src/__tests__/pipelines-routes.test.ts

@@ -0,0 +1,291 @@
+import { randomUUID } from "node:crypto";
+import express from "express";
+import request from "supertest";
+import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest";
+import {
+  agents,
+  companies,
+  createDb,
+  documents,
+  documentRevisions,
+  issues,
+  pipelineAutomationExecutions,
+  pipelineCaseBlockers,
+  pipelineCaseEvents,
+  pipelineCaseIssueLinks,
+  pipelineCases,
+  pipelineDocuments,
+  pipelineStages,
+  pipelineTransitions,
+  pipelines,
+  routines,
+} from "@paperclipai/db";
+import {
+  getEmbeddedPostgresTestSupport,
+  startEmbeddedPostgresTestDatabase,
+} from "./helpers/embedded-postgres.js";
+import { errorHandler } from "../middleware/error-handler.js";
+import { pipelineRoutes } from "../routes/pipelines.js";
+
+const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
+const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe.sequential : describe.skip;
+
+if (!embeddedPostgresSupport.supported) {
+  console.warn(
+    `Skipping embedded Postgres pipeline route tests on this host: ${embeddedPostgresSupport.reason ?? "unsupported environment"}`,
+  );
+}
+
+describeEmbeddedPostgres("pipeline routes", () => {
+  let db!: ReturnType<typeof createDb>;
+  let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
+
+  beforeAll(async () => {
+    tempDb = await startEmbeddedPostgresTestDatabase("paperclip-pipelines-routes-");
+    db = createDb(tempDb.connectionString);
+  }, 20_000);
+
+  afterEach(async () => {
+    await db.delete(pipelineAutomationExecutions);
+    await db.delete(pipelineCaseBlockers);
+    await db.delete(pipelineCaseIssueLinks);
+    await db.delete(pipelineCaseEvents);
+    await db.delete(pipelineCases);
+    await db.delete(pipelineTransitions);
+    await db.delete(pipelineStages);
+    await db.delete(pipelineDocuments);
+    await db.delete(documentRevisions);
+    await db.delete(documents);
+    await db.delete(issues);
+    await db.delete(pipelines);
+    await db.delete(routines);
+    await db.delete(agents);
+    await db.delete(companies);
+  });
+
+  afterAll(async () => {
+    await tempDb?.cleanup();
+  });
+
+  function app(actor: Express.Request["actor"]) {
+    const instance = express();
+    instance.use(express.json());
+    instance.use((req, _res, next) => {
+      req.actor = actor;
+      next();
+    });
+    instance.use("/api", pipelineRoutes(db));
+    instance.use(errorHandler);
+    return instance;
+  }
+
+  async function seedCompany(name = "Pipeline Co") {
+    const [company] = await db.insert(companies).values({
+      name,
+      issuePrefix: `P${randomUUID().replace(/-/g, "").slice(0, 6).toUpperCase()}`,
+    }).returning();
+    return company!;
+  }
+
+  const boardActor: Express.Request["actor"] = {
+    type: "board",
+    userId: "board-user",
+    source: "local_implicit",
+    isInstanceAdmin: true,
+  };
+
+  it("exposes the pipeline and case route surface", async () => {
+    const company = await seedCompany();
+    const http = request(app(boardActor));
+
+    const createdPipeline = await http
+      .post(`/api/companies/${company.id}/pipelines`)
+      .send({
+        key: "content",
+        name: "Content",
+        stages: [
+          { key: "intake", name: "Intake", kind: "open", position: 100 },
+          {
+            key: "review",
+            name: "Review",
+            kind: "review",
+            position: 200,
+            config: { approveToStageKey: "done", rejectToStageKey: "cancelled", requireRejectReason: true },
+          },
+          { key: "done", name: "Done", kind: "done", position: 900 },
+          { key: "cancelled", name: "Cancelled", kind: "cancelled", position: 1000 },
+        ],
+      })
+      .expect(201);
+    const pipelineId = createdPipeline.body.id;
+    const stageId = createdPipeline.body.stages[0].id;
+
+    await http.get(`/api/companies/${company.id}/pipelines`).expect(200);
+    await http.get(`/api/pipelines/${pipelineId}`).expect(200);
+    await http.patch(`/api/pipelines/${pipelineId}`).send({ name: "Content Ops", enforceTransitions: true }).expect(200);
+    const qaStage = await http
+      .post(`/api/pipelines/${pipelineId}/stages`)
+      .send({ key: "qa", name: "QA", kind: "working", position: 300 })
+      .expect(201);
+    await http.patch(`/api/pipelines/${pipelineId}/stages/${qaStage.body.id}`).send({ name: "QA pass" }).expect(200);
+    await http
+      .put(`/api/pipelines/${pipelineId}/transitions`)
+      .send({ enforceTransitions: false, transitions: [{ fromStageKey: "intake", toStageKey: "review" }] })
+      .expect(200);
+    await http.put(`/api/pipelines/${pipelineId}/documents/guidance`).send({ body: "Use the rubric." }).expect(200);
+    await http.get(`/api/pipelines/${pipelineId}/documents/guidance`).expect(200);
+
+    const ingested = await http
+      .post(`/api/pipelines/${pipelineId}/cases`)
+      .send({ caseKey: "case-1", title: "Case 1", fields: { channel: "blog" } })
+      .expect(201);
+    const caseId = ingested.body.case.id;
+    await http
+      .post(`/api/pipelines/${pipelineId}/cases/batch`)
+      .send({ items: [{ caseKey: "case-2", title: "Case 2" }, { caseKey: "case-3", title: "Case 3" }] })
+      .expect(200);
+    await http.get(`/api/pipelines/${pipelineId}/cases`).expect(200);
+    await http.get(`/api/cases/${caseId}`).expect(200);
+    await http.patch(`/api/cases/${caseId}`).send({ title: "Case 1 updated", expectedVersion: 1 }).expect(200);
+    const claimed = await http.post(`/api/cases/${caseId}/claim`).send({ leaseSeconds: 60 }).expect(200);
+    await http.post(`/api/cases/${caseId}/release`).send({ leaseToken: claimed.body.leaseToken }).expect(200);
+    const suggestion = await http
+      .post(`/api/cases/${caseId}/suggest-transition`)
+      .send({ toStageKey: "review", rationale: "Ready for review" })
+      .expect(200);
+    await http
+      .post(`/api/cases/${caseId}/resolve-suggestion`)
+      .send({ suggestionId: suggestion.body.suggestion.id, resolution: "accept", expectedVersion: 2 })
+      .expect(200);
+    await http.get(`/api/cases/${caseId}/events`).expect(200);
+    await http.get(`/api/companies/${company.id}/review-cases`).expect(200);
+    await http.post(`/api/cases/${caseId}/review`).send({ decision: "approve", expectedVersion: 3 }).expect(200);
+
+    const reviewCase = await http
+      .post(`/api/pipelines/${pipelineId}/cases`)
+      .send({ caseKey: "case-review", title: "Bulk review" })
+      .expect(201);
+    await http
+      .post(`/api/cases/${reviewCase.body.case.id}/transition`)
+      .send({ toStageKey: "review", expectedVersion: 1 })
+      .expect(200);
+    await http
+      .post(`/api/companies/${company.id}/review-cases/bulk`)
+      .send({ items: [{ caseId: reviewCase.body.case.id, decision: "reject", reason: "Not useful", expectedVersion: 2 }] })
+      .expect(200);
+
+    const blocker = await http.post(`/api/pipelines/${pipelineId}/cases`).send({ caseKey: "blocker", title: "Blocker" }).expect(201);
+    const blocked = await http.post(`/api/pipelines/${pipelineId}/cases`).send({ caseKey: "blocked", title: "Blocked" }).expect(201);
+    await http
+      .put(`/api/cases/${blocked.body.case.id}/blockers`)
+      .send({ blockedByCaseIds: [blocker.body.case.id] })
+      .expect(200);
+    await http.get(`/api/cases/${blocked.body.case.id}/rollup`).expect(200);
+    await http.get(`/api/cases/${blocked.body.case.id}/context-pack`).expect(200);
+    await http.post(`/api/cases/${blocked.body.case.id}/open-conversation`).expect(201);
+
+    const [agent] = await db.insert(agents).values({
+      companyId: company.id,
+      name: "Routine Agent",
+      role: "engineer",
+      adapterType: "codex_local",
+      adapterConfig: {},
+      runtimeConfig: {},
+      permissions: {},
+    }).returning();
+    const [routine] = await db.insert(routines).values({ companyId: company.id, title: "Routine", assigneeAgentId: agent!.id }).returning();
+    await db.insert(pipelineAutomationExecutions).values({
+      companyId: company.id,
+      caseId: blocked.body.case.id,
+      automationId: "retry-me",
+      triggeringEventId: randomUUID(),
+      routineId: routine!.id,
+      status: "failed",
+      error: "boom",
+    });
+    await http.post(`/api/cases/${blocked.body.case.id}/automations/retry-me/retry`).expect(200);
+
+    await http.delete(`/api/pipelines/${pipelineId}/stages/${stageId}?moveCasesToStageId=${qaStage.body.id}`).expect(200);
+  });
+
+  it("returns 404 for cross-company pipeline access", async () => {
+    const company = await seedCompany();
+    const [pipeline] = await db.insert(pipelines).values({ companyId: company.id, key: "x", name: "X" }).returning();
+    const otherAgent: Express.Request["actor"] = {
+      type: "agent",
+      agentId: randomUUID(),
+      companyId: randomUUID(),
+      runId: randomUUID(),
+      source: "agent_key",
+    };
+
+    const res = await request(app(otherAgent)).get(`/api/pipelines/${pipeline!.id}`);
+    expect(res.status).toBe(404);
+  });
+
+  it("rejects agent mutations without a run id", async () => {
+    const company = await seedCompany();
+    const agentActor: Express.Request["actor"] = {
+      type: "agent",
+      agentId: randomUUID(),
+      companyId: company.id,
+      source: "agent_key",
+    };
+
+    const res = await request(app(agentActor))
+      .post(`/api/companies/${company.id}/pipelines`)
+      .send({ key: "agent", name: "Agent pipeline" });
+
+    expect(res.status).toBe(422);
+    expect(res.body.code).toBe("run_id_required");
+  });
+
+  it("rejects agent exits from human review stages", async () => {
+    const company = await seedCompany();
+    const http = request(app(boardActor));
+    const pipelineRes = await http
+      .post(`/api/companies/${company.id}/pipelines`)
+      .send({
+        key: "review-authz",
+        name: "Review authz",
+        stages: [
+          { key: "intake", name: "Intake", kind: "open", position: 100 },
+          { key: "review", name: "Review", kind: "review", position: 200, config: { approveToStageKey: "done", rejectToStageKey: "cancelled" } },
+          { key: "done", name: "Done", kind: "done", position: 900 },
+          { key: "cancelled", name: "Cancelled", kind: "cancelled", position: 1000 },
+        ],
+      })
+      .expect(201);
+    const caseRes = await http.post(`/api/pipelines/${pipelineRes.body.id}/cases`).send({ caseKey: "review", title: "Review me" }).expect(201);
+    await http.post(`/api/cases/${caseRes.body.case.id}/transition`).send({ toStageKey: "review", expectedVersion: 1 }).expect(200);
+
+    const agentActor: Express.Request["actor"] = {
+      type: "agent",
+      agentId: randomUUID(),
+      companyId: company.id,
+      runId: randomUUID(),
+      source: "agent_key",
+    };
+    const res = await request(app(agentActor))
+      .post(`/api/cases/${caseRes.body.case.id}/transition`)
+      .send({ toStageKey: "done", expectedVersion: 2 });
+
+    expect(res.status).toBe(403);
+    expect(res.body.code).toBe("review_required");
+  });
+
+  it("returns conflict bodies with code, current version, and stage", async () => {
+    const company = await seedCompany();
+    const http = request(app(boardActor));
+    const pipelineRes = await http.post(`/api/companies/${company.id}/pipelines`).send({ key: "conflict", name: "Conflict" }).expect(201);
+    const caseRes = await http.post(`/api/pipelines/${pipelineRes.body.id}/cases`).send({ caseKey: "conflict", title: "Conflict" }).expect(201);
+    await http.patch(`/api/cases/${caseRes.body.case.id}`).send({ title: "Updated", expectedVersion: 1 }).expect(200);
+
+    const res = await http.patch(`/api/cases/${caseRes.body.case.id}`).send({ title: "Stale", expectedVersion: 1 });
+
+    expect(res.status).toBe(409);
+    expect(res.body.code).toBe("version_conflict");
+    expect(res.body.details.version).toBe(2);
+    expect(res.body.details.stage.key).toBe("intake");
+  });
+});

server/src/app.ts

@@ -18,6 +18,7 @@ import { projectRoutes } from "./routes/projects.js";
 import { issueRoutes } from "./routes/issues.js";
 import { issueTreeControlRoutes } from "./routes/issue-tree-control.js";
 import { routineRoutes } from "./routes/routines.js";
+import { pipelineRoutes } from "./routes/pipelines.js";
 import { environmentRoutes } from "./routes/environments.js";
 import { executionWorkspaceRoutes } from "./routes/execution-workspaces.js";
 import { goalRoutes } from "./routes/goals.js";
@@ -223,6 +224,7 @@ export async function createApp(
   }));
   api.use(issueTreeControlRoutes(db));
   api.use(routineRoutes(db, { pluginWorkerManager: workerManager }));
+  api.use(pipelineRoutes(db));
   api.use(environmentRoutes(db, { pluginWorkerManager: workerManager }));
   api.use(executionWorkspaceRoutes(db));
   api.use(goalRoutes(db));

server/src/middleware/error-handler.ts

@@ -40,6 +40,9 @@ export function errorHandler(
   _next: NextFunction,
 ) {
   if (err instanceof HttpError) {
+    const details = err.details && typeof err.details === "object" && !Array.isArray(err.details)
+      ? err.details as Record<string, unknown>
+      : null;
     if (err.status >= 500) {
       attachErrorContext(
         req,
@@ -52,6 +55,7 @@ export function errorHandler(
     }
     res.status(err.status).json({
       error: err.message,
+      ...(typeof details?.code === "string" ? { code: details.code } : {}),
       ...(err.details ? { details: err.details } : {}),
     });
     return;

server/src/routes/pipelines.ts

@@ -0,0 +1,848 @@
+import { Router, type Request } from "express";
+import { z } from "zod";
+import { and, asc, eq, ilike, inArray, isNotNull, isNull, or, sql } from "drizzle-orm";
+import type { Db } from "@paperclipai/db";
+import {
+  documents,
+  documentRevisions,
+  issues as issueRows,
+  pipelineAutomationExecutions,
+  pipelineCaseBlockers,
+  pipelineCaseEvents,
+  pipelineCaseIssueLinks,
+  pipelineCases,
+  pipelineDocuments,
+  pipelineStages,
+  pipelineTransitions,
+  pipelines,
+} from "@paperclipai/db";
+import { validate } from "../middleware/validate.js";
+import { conflict, forbidden, HttpError, notFound, unauthorized, unprocessable } from "../errors.js";
+import { pipelineService, type PipelineActor, type PipelineStageConfig, type PipelineStageKind } from "../services/pipelines.js";
+import { assertCompanyAccess } from "./authz.js";
+
+const stageKindSchema = z.enum(["open", "working", "review", "done", "cancelled"]);
+const jsonObjectSchema = z.record(z.string(), z.unknown());
+const stageConfigSchema = z.record(z.string(), z.unknown()).default({});
+const casePatchSchema = z.object({
+  title: z.string().trim().min(1).max(500).optional(),
+  summary: z.string().max(8_000).nullable().optional(),
+  fields: jsonObjectSchema.optional(),
+  workspaceRef: jsonObjectSchema.nullable().optional(),
+  expectedVersion: z.number().int().positive().optional(),
+  leaseToken: z.string().uuid().nullable().optional(),
+});
+const ingestCaseSchema = z.object({
+  caseKey: z.string().max(1_024).nullable().optional(),
+  title: z.string().trim().min(1).max(500),
+  summary: z.string().max(8_000).nullable().optional(),
+  fields: jsonObjectSchema.optional(),
+  stageKey: z.string().trim().min(1).max(120).optional(),
+  parentCaseId: z.string().uuid().nullable().optional(),
+  workspaceRef: jsonObjectSchema.nullable().optional(),
+  blockedByCaseIds: z.array(z.string().uuid()).max(100).optional(),
+});
+const createPipelineSchema = z.object({
+  key: z.string().trim().min(1).max(120),
+  name: z.string().trim().min(1).max(200),
+  description: z.string().max(8_000).nullable().optional(),
+  projectId: z.string().uuid().nullable().optional(),
+  enforceTransitions: z.boolean().optional(),
+  stages: z.array(z.object({
+    key: z.string().trim().min(1).max(120),
+    name: z.string().trim().min(1).max(200),
+    kind: stageKindSchema,
+    position: z.number().int().optional(),
+    config: stageConfigSchema.optional(),
+  })).optional(),
+});
+const updatePipelineSchema = z.object({
+  name: z.string().trim().min(1).max(200).optional(),
+  description: z.string().max(8_000).nullable().optional(),
+  enforceTransitions: z.boolean().optional(),
+  archived: z.boolean().optional(),
+});
+const createStageSchema = z.object({
+  key: z.string().trim().min(1).max(120),
+  name: z.string().trim().min(1).max(200),
+  kind: stageKindSchema,
+  position: z.number().int(),
+  config: stageConfigSchema.optional(),
+});
+const updateStageSchema = z.object({
+  key: z.string().trim().min(1).max(120).optional(),
+  name: z.string().trim().min(1).max(200).optional(),
+  kind: stageKindSchema.optional(),
+  position: z.number().int().optional(),
+  config: stageConfigSchema.optional(),
+});
+const replaceTransitionsSchema = z.object({
+  transitions: z.array(z.object({
+    fromStageKey: z.string().trim().min(1).max(120),
+    toStageKey: z.string().trim().min(1).max(120),
+    label: z.string().max(200).nullable().optional(),
+  })).max(500),
+  enforceTransitions: z.boolean().optional(),
+});
+const batchIngestSchema = z.object({ items: z.array(ingestCaseSchema).max(200) });
+const claimCaseSchema = z.object({ leaseSeconds: z.number().int().positive().max(86_400).optional() });
+const releaseCaseSchema = z.object({
+  leaseToken: z.string().uuid().nullable().optional(),
+  force: z.boolean().optional(),
+});
+const transitionCaseSchema = z.object({
+  toStageKey: z.string().trim().min(1).max(120),
+  expectedVersion: z.number().int().positive(),
+  leaseToken: z.string().uuid().nullable().optional(),
+  reason: z.string().max(4_000).nullable().optional(),
+  acceptSuggestionId: z.string().uuid().optional(),
+});
+const suggestTransitionSchema = z.object({
+  toStageKey: z.string().trim().min(1).max(120),
+  rationale: z.string().trim().min(1).max(8_000),
+  confidence: z.number().min(0).max(1).optional(),
+});
+const resolveSuggestionSchema = z.object({
+  suggestionId: z.string().uuid(),
+  resolution: z.enum(["accept", "dismiss"]),
+  expectedVersion: z.number().int().positive().optional(),
+  reason: z.string().max(4_000).nullable().optional(),
+  leaseToken: z.string().uuid().nullable().optional(),
+});
+const reviewCaseSchema = z.object({
+  decision: z.enum(["approve", "reject"]),
+  reason: z.string().max(4_000).nullable().optional(),
+  edits: casePatchSchema.omit({ expectedVersion: true, leaseToken: true }).optional(),
+  expectedVersion: z.number().int().positive(),
+  leaseToken: z.string().uuid().nullable().optional(),
+});
+const blockersSchema = z.object({ blockedByCaseIds: z.array(z.string().uuid()).max(100) });
+const bulkReviewSchema = z.object({
+  items: z.array(reviewCaseSchema.extend({ caseId: z.string().uuid() })).max(100),
+});
+const upsertPipelineDocumentSchema = z.object({
+  title: z.string().trim().min(1).max(200).optional(),
+  body: z.string().max(200_000),
+});
+
+function isPgUniqueViolation(error: unknown) {
+  return (error as { code?: unknown })?.code === "23505";
+}
+
+function codedConflictForUnique(error: unknown): never {
+  if (isPgUniqueViolation(error)) {
+    throw conflict("Duplicate pipeline resource key", { code: "duplicate_key" });
+  }
+  throw error;
+}
+
+function assertPipelineCompanyAccess(req: Request, companyId: string) {
+  try {
+    assertCompanyAccess(req, companyId);
+  } catch (error) {
+    if (
+      error instanceof HttpError &&
+      error.status === 403 &&
+      (error.message.includes("another company") || error.message.includes("does not have access"))
+    ) {
+      throw notFound("Pipeline resource not found");
+    }
+    throw error;
+  }
+}
+
+function actorForMutation(req: Request): PipelineActor {
+  if (req.actor.type === "agent") {
+    if (!req.actor.agentId) throw unauthorized();
+    if (!req.actor.runId) throw unprocessable("Agent pipeline mutations require a run id", { code: "run_id_required" });
+    return { type: "agent", agentId: req.actor.agentId, runId: req.actor.runId };
+  }
+  if (req.actor.type === "board") {
+    return { type: "user", userId: req.actor.userId ?? "board" };
+  }
+  throw unauthorized();
+}
+
+async function resolvePipelineCompanyId(db: Db, pipelineId: string) {
+  const row = await db
+    .select({ companyId: pipelines.companyId })
+    .from(pipelines)
+    .where(eq(pipelines.id, pipelineId))
+    .limit(1)
+    .then((rows) => rows[0] ?? null);
+  if (!row) throw notFound("Pipeline not found");
+  return row.companyId;
+}
+
+async function resolveCaseCompanyId(db: Db, caseId: string) {
+  const row = await db
+    .select({ companyId: pipelineCases.companyId })
+    .from(pipelineCases)
+    .where(eq(pipelineCases.id, caseId))
+    .limit(1)
+    .then((rows) => rows[0] ?? null);
+  if (!row) throw notFound("Pipeline case not found");
+  return row.companyId;
+}
+
+async function assertPipelineAccess(db: Db, req: Request, pipelineId: string) {
+  const companyId = await resolvePipelineCompanyId(db, pipelineId);
+  assertPipelineCompanyAccess(req, companyId);
+  return companyId;
+}
+
+async function assertCaseAccess(db: Db, req: Request, caseId: string) {
+  const companyId = await resolveCaseCompanyId(db, caseId);
+  assertPipelineCompanyAccess(req, companyId);
+  return companyId;
+}
+
+async function getStagesByKey(db: Db, pipelineId: string) {
+  const rows = await db.select().from(pipelineStages).where(eq(pipelineStages.pipelineId, pipelineId));
+  return new Map(rows.map((stage) => [stage.key, stage]));
+}
+
+async function writeRouteEvent(
+  db: Pick<Db, "insert">,
+  input: {
+    companyId: string;
+    caseId: string;
+    type: string;
+    actor: PipelineActor;
+    payload?: Record<string, unknown>;
+  },
+) {
+  const actorPatch = input.actor.type === "agent"
+    ? { actorType: "agent", actorAgentId: input.actor.agentId, runId: input.actor.runId }
+    : input.actor.type === "user"
+      ? { actorType: "user", actorUserId: input.actor.userId }
+      : { actorType: "system" };
+  const [event] = await db.insert(pipelineCaseEvents).values({
+    companyId: input.companyId,
+    caseId: input.caseId,
+    type: input.type,
+    ...actorPatch,
+    payload: input.payload ?? {},
+  }).returning();
+  return event!;
+}
+
+export function pipelineRoutes(db: Db) {
+  const router = Router();
+  const svc = pipelineService(db);
+
+  router.get("/companies/:companyId/pipelines", async (req, res) => {
+    const companyId = req.params.companyId as string;
+    assertPipelineCompanyAccess(req, companyId);
+    const rows = await db
+      .select({
+        pipeline: pipelines,
+        stageCount: sql<number>`count(distinct ${pipelineStages.id})::int`,
+        openCaseCount: sql<number>`count(distinct ${pipelineCases.id}) filter (where ${pipelineCases.terminalKind} is null)::int`,
+      })
+      .from(pipelines)
+      .leftJoin(pipelineStages, eq(pipelineStages.pipelineId, pipelines.id))
+      .leftJoin(pipelineCases, eq(pipelineCases.pipelineId, pipelines.id))
+      .where(eq(pipelines.companyId, companyId))
+      .groupBy(pipelines.id)
+      .orderBy(asc(pipelines.createdAt));
+    res.json(rows.map((row) => ({ ...row.pipeline, stageCount: row.stageCount, openCaseCount: row.openCaseCount })));
+  });
+
+  router.post("/companies/:companyId/pipelines", validate(createPipelineSchema), async (req, res) => {
+    const companyId = req.params.companyId as string;
+    assertPipelineCompanyAccess(req, companyId);
+    const actor = actorForMutation(req);
+    try {
+      const created = await svc.createPipeline({
+        companyId,
+        key: req.body.key,
+        name: req.body.name,
+        description: req.body.description,
+        projectId: req.body.projectId,
+        enforceTransitions: req.body.enforceTransitions,
+        stages: req.body.stages?.map((stage: {
+          key: string;
+          name: string;
+          kind: PipelineStageKind;
+          position?: number;
+          config?: Record<string, unknown>;
+        }) => ({
+          ...stage,
+          kind: stage.kind as PipelineStageKind,
+          config: stage.config as PipelineStageConfig | undefined,
+        })),
+        actor,
+      });
+      res.status(201).json(created);
+    } catch (error) {
+      codedConflictForUnique(error);
+    }
+  });
+
+  router.get("/companies/:companyId/review-cases", async (req, res) => {
+    const companyId = req.params.companyId as string;
+    assertPipelineCompanyAccess(req, companyId);
+    const pipelineId = typeof req.query.pipelineId === "string" ? req.query.pipelineId : undefined;
+    const parentCaseId = typeof req.query.parentCaseId === "string" ? req.query.parentCaseId : undefined;
+    const rows = await db
+      .select({ case: pipelineCases, pipeline: pipelines, stage: pipelineStages })
+      .from(pipelineCases)
+      .innerJoin(pipelines, eq(pipelineCases.pipelineId, pipelines.id))
+      .innerJoin(pipelineStages, eq(pipelineCases.stageId, pipelineStages.id))
+      .where(and(
+        eq(pipelineCases.companyId, companyId),
+        eq(pipelineStages.kind, "review"),
+        isNull(pipelineCases.terminalKind),
+        pipelineId ? eq(pipelineCases.pipelineId, pipelineId) : undefined,
+        parentCaseId ? eq(pipelineCases.parentCaseId, parentCaseId) : undefined,
+      ))
+      .orderBy(asc(pipelineCases.createdAt));
+    res.json(rows);
+  });
+
+  router.post("/companies/:companyId/review-cases/bulk", validate(bulkReviewSchema), async (req, res) => {
+    const companyId = req.params.companyId as string;
+    assertPipelineCompanyAccess(req, companyId);
+    const actor = actorForMutation(req);
+    const results = [];
+    for (const item of req.body.items) {
+      try {
+        results.push({ caseId: item.caseId, ok: true, result: await reviewCase(db, svc, companyId, item.caseId, item, actor) });
+      } catch (error) {
+        const httpError = error as { status?: number; message?: string; details?: unknown };
+        results.push({
+          caseId: item.caseId,
+          ok: false,
+          error: { status: httpError.status ?? 500, message: httpError.message ?? "Unknown error", details: httpError.details },
+        });
+      }
+    }
+    res.json({ results });
+  });
+
+  router.get("/pipelines/:pipelineId", async (req, res) => {
+    const pipelineId = req.params.pipelineId as string;
+    const companyId = await assertPipelineAccess(db, req, pipelineId);
+    const [pipeline, stages, transitions, documentKeys] = await Promise.all([
+      db.select().from(pipelines).where(and(eq(pipelines.id, pipelineId), eq(pipelines.companyId, companyId))).then((rows) => rows[0] ?? null),
+      db.select().from(pipelineStages).where(eq(pipelineStages.pipelineId, pipelineId)).orderBy(asc(pipelineStages.position)),
+      db.select().from(pipelineTransitions).where(eq(pipelineTransitions.pipelineId, pipelineId)),
+      db.select({ key: pipelineDocuments.key, documentId: pipelineDocuments.documentId })
+        .from(pipelineDocuments)
+        .where(and(eq(pipelineDocuments.companyId, companyId), eq(pipelineDocuments.pipelineId, pipelineId))),
+    ]);
+    if (!pipeline) throw notFound("Pipeline not found");
+    res.json({ ...pipeline, stages, transitions, documentKeys });
+  });
+
+  router.patch("/pipelines/:pipelineId", validate(updatePipelineSchema), async (req, res) => {
+    const pipelineId = req.params.pipelineId as string;
+    const companyId = await assertPipelineAccess(db, req, pipelineId);
+    actorForMutation(req);
+    const patch: Partial<typeof pipelines.$inferInsert> = { updatedAt: new Date() };
+    if (req.body.name !== undefined) patch.name = req.body.name;
+    if (req.body.description !== undefined) patch.description = req.body.description;
+    if (req.body.enforceTransitions !== undefined) patch.enforceTransitions = req.body.enforceTransitions;
+    if (req.body.archived !== undefined) patch.archivedAt = req.body.archived ? new Date() : null;
+    const [updated] = await db
+      .update(pipelines)
+      .set(patch)
+      .where(and(eq(pipelines.id, pipelineId), eq(pipelines.companyId, companyId)))
+      .returning();
+    res.json(updated);
+  });
+
+  router.post("/pipelines/:pipelineId/stages", validate(createStageSchema), async (req, res) => {
+    const pipelineId = req.params.pipelineId as string;
+    const companyId = await assertPipelineAccess(db, req, pipelineId);
+    actorForMutation(req);
+    try {
+      const stage = await svc.createStage({
+        companyId,
+        pipelineId,
+        key: req.body.key,
+        name: req.body.name,
+        kind: req.body.kind,
+        position: req.body.position,
+        config: req.body.config,
+      });
+      res.status(201).json(stage);
+    } catch (error) {
+      codedConflictForUnique(error);
+    }
+  });
+
+  router.patch("/pipelines/:pipelineId/stages/:stageId", validate(updateStageSchema), async (req, res) => {
+    const pipelineId = req.params.pipelineId as string;
+    const stageId = req.params.stageId as string;
+    await assertPipelineAccess(db, req, pipelineId);
+    actorForMutation(req);
+    try {
+      const [updated] = await db
+        .update(pipelineStages)
+        .set({ ...req.body, updatedAt: new Date() })
+        .where(and(eq(pipelineStages.id, stageId), eq(pipelineStages.pipelineId, pipelineId)))
+        .returning();
+      if (!updated) throw notFound("Pipeline stage not found");
+      res.json(updated);
+    } catch (error) {
+      codedConflictForUnique(error);
+    }
+  });
+
+  router.delete("/pipelines/:pipelineId/stages/:stageId", async (req, res) => {
+    const pipelineId = req.params.pipelineId as string;
+    const stageId = req.params.stageId as string;
+    const companyId = await assertPipelineAccess(db, req, pipelineId);
+    const actor = actorForMutation(req);
+    const result = await svc.deleteStage({
+      companyId,
+      pipelineId,
+      stageId,
+      moveCasesToStageId: typeof req.query.moveCasesToStageId === "string" ? req.query.moveCasesToStageId : null,
+      actor,
+    });
+    res.json(result);
+  });
+
+  router.put("/pipelines/:pipelineId/transitions", validate(replaceTransitionsSchema), async (req, res) => {
+    const pipelineId = req.params.pipelineId as string;
+    const companyId = await assertPipelineAccess(db, req, pipelineId);
+    actorForMutation(req);
+    const byKey = await getStagesByKey(db, pipelineId);
+    const transitions = req.body.transitions.map((edge: z.infer<typeof replaceTransitionsSchema>["transitions"][number]) => {
+      const from = byKey.get(edge.fromStageKey);
+      const to = byKey.get(edge.toStageKey);
+      if (!from || !to) throw unprocessable("Transition references unknown stage", { code: "validation" });
+      return { pipelineId, fromStageId: from.id, toStageId: to.id, label: edge.label ?? null };
+    });
+    const result = await db.transaction(async (tx) => {
+      await tx.delete(pipelineTransitions).where(eq(pipelineTransitions.pipelineId, pipelineId));
+      if (req.body.enforceTransitions !== undefined) {
+        await tx.update(pipelines).set({ enforceTransitions: req.body.enforceTransitions, updatedAt: new Date() })
+          .where(and(eq(pipelines.id, pipelineId), eq(pipelines.companyId, companyId)));
+      }
+      return transitions.length ? tx.insert(pipelineTransitions).values(transitions).returning() : [];
+    });
+    res.json({ transitions: result });
+  });
+
+  router.get("/pipelines/:pipelineId/documents/:key", async (req, res) => {
+    const pipelineId = req.params.pipelineId as string;
+    const key = req.params.key as string;
+    const companyId = await assertPipelineAccess(db, req, pipelineId);
+    const row = await db
+      .select({ link: pipelineDocuments, document: documents, revision: documentRevisions })
+      .from(pipelineDocuments)
+      .innerJoin(documents, eq(pipelineDocuments.documentId, documents.id))
+      .leftJoin(documentRevisions, eq(documents.latestRevisionId, documentRevisions.id))
+      .where(and(eq(pipelineDocuments.companyId, companyId), eq(pipelineDocuments.pipelineId, pipelineId), eq(pipelineDocuments.key, key)))
+      .limit(1)
+      .then((rows) => rows[0] ?? null);
+    if (!row) throw notFound("Pipeline document not found");
+    res.json(row);
+  });
+
+  router.put("/pipelines/:pipelineId/documents/:key", validate(upsertPipelineDocumentSchema), async (req, res) => {
+    const pipelineId = req.params.pipelineId as string;
+    const key = req.params.key as string;
+    const companyId = await assertPipelineAccess(db, req, pipelineId);
+    const actor = actorForMutation(req);
+    const result = await db.transaction(async (tx) => {
+      const existing = await tx.select().from(pipelineDocuments)
+        .where(and(eq(pipelineDocuments.companyId, companyId), eq(pipelineDocuments.pipelineId, pipelineId), eq(pipelineDocuments.key, key)))
+        .limit(1)
+        .then((rows) => rows[0] ?? null);
+      const [document] = existing
+        ? await tx.update(documents).set({ title: req.body.title ?? key, updatedAt: new Date() }).where(eq(documents.id, existing.documentId)).returning()
+        : await tx.insert(documents).values({
+          companyId,
+          title: req.body.title ?? key,
+          latestBody: req.body.body,
+          latestRevisionNumber: 1,
+          createdByAgentId: actor.type === "agent" ? actor.agentId : null,
+          createdByUserId: actor.type === "user" ? actor.userId : null,
+        }).returning();
+      const [{ count }] = await tx.select({ count: sql<number>`count(*)::int` }).from(documentRevisions)
+        .where(eq(documentRevisions.documentId, document!.id));
+      const [revision] = await tx.insert(documentRevisions).values({
+        companyId,
+        documentId: document!.id,
+        revisionNumber: (count ?? 0) + 1,
+        title: req.body.title ?? document!.title,
+        body: req.body.body,
+        createdByAgentId: actor.type === "agent" ? actor.agentId : null,
+        createdByUserId: actor.type === "user" ? actor.userId : null,
+      }).returning();
+      await tx.update(documents).set({
+        latestBody: req.body.body,
+        latestRevisionId: revision!.id,
+        latestRevisionNumber: revision!.revisionNumber,
+        updatedAt: new Date(),
+        updatedByAgentId: actor.type === "agent" ? actor.agentId : null,
+        updatedByUserId: actor.type === "user" ? actor.userId : null,
+      }).where(eq(documents.id, document!.id));
+      if (!existing) {
+        await tx.insert(pipelineDocuments).values({ companyId, pipelineId, documentId: document!.id, key });
+      }
+      return { document: { ...document!, latestRevisionId: revision!.id }, revision };
+    });
+    res.json(result);
+  });
+
+  router.post("/pipelines/:pipelineId/cases", validate(ingestCaseSchema), async (req, res) => {
+    const pipelineId = req.params.pipelineId as string;
+    const companyId = await assertPipelineAccess(db, req, pipelineId);
+    const actor = actorForMutation(req);
+    const result = await svc.ingestCase({ companyId, pipelineId, ...req.body, actor });
+    res.status(result.created ? 201 : 200).json(result);
+  });
+
+  router.post("/pipelines/:pipelineId/cases/batch", validate(batchIngestSchema), async (req, res) => {
+    const pipelineId = req.params.pipelineId as string;
+    const companyId = await assertPipelineAccess(db, req, pipelineId);
+    const actor = actorForMutation(req);
+    res.json(await svc.ingestCases({ companyId, pipelineId, items: req.body.items, actor }));
+  });
+
+  router.get("/pipelines/:pipelineId/cases", async (req, res) => {
+    const pipelineId = req.params.pipelineId as string;
+    const companyId = await assertPipelineAccess(db, req, pipelineId);
+    const stageKey = typeof req.query.stageKey === "string" ? req.query.stageKey : undefined;
+    const q = typeof req.query.q === "string" ? req.query.q : undefined;
+    const terminal = req.query.terminal === "true" ? true : req.query.terminal === "false" ? false : undefined;
+    const parentCaseId = typeof req.query.parentCaseId === "string" ? req.query.parentCaseId : undefined;
+    const rows = await db
+      .select({ case: pipelineCases, stage: pipelineStages })
+      .from(pipelineCases)
+      .innerJoin(pipelineStages, eq(pipelineCases.stageId, pipelineStages.id))
+      .where(and(
+        eq(pipelineCases.companyId, companyId),
+        eq(pipelineCases.pipelineId, pipelineId),
+        stageKey ? eq(pipelineStages.key, stageKey) : undefined,
+        parentCaseId ? eq(pipelineCases.parentCaseId, parentCaseId) : undefined,
+        terminal === true ? isNotNull(pipelineCases.terminalKind) : terminal === false ? isNull(pipelineCases.terminalKind) : undefined,
+        q ? or(ilike(pipelineCases.title, `%${q}%`), ilike(pipelineCases.summary, `%${q}%`)) : undefined,
+      ))
+      .orderBy(asc(pipelineCases.createdAt));
+    res.json(rows);
+  });
+
+  router.get("/cases/:caseId", async (req, res) => {
+    const caseId = req.params.caseId as string;
+    const companyId = await assertCaseAccess(db, req, caseId);
+    const detail = await getCaseDetail(db, companyId, caseId);
+    res.json(detail);
+  });
+
+  router.patch("/cases/:caseId", validate(casePatchSchema), async (req, res) => {
+    const caseId = req.params.caseId as string;
+    const companyId = await assertCaseAccess(db, req, caseId);
+    const actor = actorForMutation(req);
+    const updated = await svc.patchCaseContent({ companyId, caseId, ...req.body, actor });
+    if (req.body.workspaceRef !== undefined) {
+      const [withWorkspace] = await db.update(pipelineCases)
+        .set({ workspaceRef: req.body.workspaceRef, updatedAt: new Date() })
+        .where(eq(pipelineCases.id, caseId))
+        .returning();
+      res.json(withWorkspace);
+      return;
+    }
+    res.json(updated);
+  });
+
+  router.post("/cases/:caseId/claim", validate(claimCaseSchema), async (req, res) => {
+    const caseId = req.params.caseId as string;
+    const companyId = await assertCaseAccess(db, req, caseId);
+    const actor = actorForMutation(req);
+    if (actor.type === "system") throw forbidden();
+    const claimed = await svc.claimCase({ companyId, caseId, actor, leaseMs: req.body.leaseSeconds ? req.body.leaseSeconds * 1000 : undefined });
+    res.json({ case: claimed, leaseToken: claimed.leaseToken, leaseExpiresAt: claimed.leaseExpiresAt });
+  });
+
+  router.post("/cases/:caseId/release", validate(releaseCaseSchema), async (req, res) => {
+    const caseId = req.params.caseId as string;
+    const companyId = await assertCaseAccess(db, req, caseId);
+    const actor = actorForMutation(req);
+    if (req.body.force && actor.type === "agent") throw new HttpError(403, "Agents cannot force-release pipeline leases", { code: "forbidden" });
+    res.json(await svc.releaseCase({ companyId, caseId, actor, leaseToken: req.body.leaseToken, force: req.body.force }));
+  });
+
+  router.post("/cases/:caseId/transition", validate(transitionCaseSchema), async (req, res) => {
+    const caseId = req.params.caseId as string;
+    const companyId = await assertCaseAccess(db, req, caseId);
+    const actor = actorForMutation(req);
+    res.json(await svc.transitionCase({
+      companyId,
+      caseId,
+      toStageKey: req.body.toStageKey,
+      expectedVersion: req.body.expectedVersion,
+      leaseToken: req.body.leaseToken,
+      reason: req.body.reason,
+      suggestionId: req.body.acceptSuggestionId,
+    actor,
+    }));
+  });
+
+  router.post("/cases/:caseId/suggest-transition", validate(suggestTransitionSchema), async (req, res) => {
+    const caseId = req.params.caseId as string;
+    const companyId = await assertCaseAccess(db, req, caseId);
+    const actor = actorForMutation(req);
+    res.json(await svc.suggestTransition({ companyId, caseId, ...req.body, actor }));
+  });
+
+  router.post("/cases/:caseId/resolve-suggestion", validate(resolveSuggestionSchema), async (req, res) => {
+    const caseId = req.params.caseId as string;
+    const companyId = await assertCaseAccess(db, req, caseId);
+    const actor = actorForMutation(req);
+    res.json(await svc.resolveSuggestion({
+      companyId,
+      caseId,
+      suggestionId: req.body.suggestionId,
+      decision: req.body.resolution,
+      expectedVersion: req.body.expectedVersion,
+      reason: req.body.reason,
+      leaseToken: req.body.leaseToken,
+      actor,
+    }));
+  });
+
+  router.post("/cases/:caseId/review", validate(reviewCaseSchema), async (req, res) => {
+    const caseId = req.params.caseId as string;
+    const companyId = await assertCaseAccess(db, req, caseId);
+    const actor = actorForMutation(req);
+    res.json(await reviewCase(db, svc, companyId, caseId, req.body, actor));
+  });
+
+  router.put("/cases/:caseId/blockers", validate(blockersSchema), async (req, res) => {
+    const caseId = req.params.caseId as string;
+    const companyId = await assertCaseAccess(db, req, caseId);
+    const actor = actorForMutation(req);
+    const blockers = await db.transaction(async (tx) => {
+      await tx.delete(pipelineCaseBlockers).where(and(eq(pipelineCaseBlockers.companyId, companyId), eq(pipelineCaseBlockers.caseId, caseId)));
+      if (req.body.blockedByCaseIds.length) {
+        const rows = await tx.select({ id: pipelineCases.id }).from(pipelineCases)
+          .where(and(eq(pipelineCases.companyId, companyId), inArray(pipelineCases.id, req.body.blockedByCaseIds)));
+        if (rows.length !== req.body.blockedByCaseIds.length) throw notFound("Pipeline blocker case not found");
+        await tx.insert(pipelineCaseBlockers).values(req.body.blockedByCaseIds.map((blockedByCaseId: string) => ({
+          companyId,
+          caseId,
+          blockedByCaseId,
+        })));
+      }
+      await writeRouteEvent(tx, { companyId, caseId, type: "blockers_set", actor, payload: { blockedByCaseIds: req.body.blockedByCaseIds } });
+      return tx.select().from(pipelineCaseBlockers).where(and(eq(pipelineCaseBlockers.companyId, companyId), eq(pipelineCaseBlockers.caseId, caseId)));
+    });
+    res.json({ blockers });
+  });
+
+  router.post("/cases/:caseId/open-conversation", async (req, res) => {
+    const caseId = req.params.caseId as string;
+    const companyId = await assertCaseAccess(db, req, caseId);
+    const actor = actorForMutation(req);
+    const existing = await db
+      .select({ issue: issueRows, link: pipelineCaseIssueLinks })
+      .from(pipelineCaseIssueLinks)
+      .innerJoin(issueRows, eq(pipelineCaseIssueLinks.issueId, issueRows.id))
+      .where(and(
+        eq(pipelineCaseIssueLinks.companyId, companyId),
+        eq(pipelineCaseIssueLinks.caseId, caseId),
+        eq(pipelineCaseIssueLinks.role, "conversation"),
+        isNull(issueRows.completedAt),
+        isNull(issueRows.cancelledAt),
+      ))
+      .limit(1)
+      .then((rows) => rows[0] ?? null);
+    if (existing) {
+      res.json({ issue: existing.issue, created: false });
+      return;
+    }
+    const detail = await getCaseDetail(db, companyId, caseId);
+    const result = await db.transaction(async (tx) => {
+      const [issue] = await tx.insert(issueRows).values({
+        companyId,
+        title: `Discuss case: ${detail.case.title}`,
+        description: `Pipeline case ${detail.case.caseKey}\n\nCase id: ${detail.case.id}`,
+        status: "todo",
+        priority: "medium",
+        originKind: "pipeline_case_conversation",
+        originId: detail.case.id,
+        createdByAgentId: actor.type === "agent" ? actor.agentId : null,
+        createdByUserId: actor.type === "user" ? actor.userId : null,
+      }).returning();
+      await tx.insert(pipelineCaseIssueLinks).values({
+        companyId,
+        caseId,
+        issueId: issue!.id,
+        role: "conversation",
+        createdByRunId: actor.type === "agent" ? actor.runId : null,
+      });
+      await writeRouteEvent(tx, {
+        companyId,
+        caseId,
+        type: "conversation_opened",
+        actor,
+        payload: { issueId: issue!.id },
+      });
+      return issue!;
+    });
+    res.status(201).json({ issue: result, created: true });
+  });
+
+  router.get("/cases/:caseId/events", async (req, res) => {
+    const caseId = req.params.caseId as string;
+    const companyId = await assertCaseAccess(db, req, caseId);
+    res.json(await svc.listCaseEvents(companyId, caseId));
+  });
+
+  router.get("/cases/:caseId/rollup", async (req, res) => {
+    const caseId = req.params.caseId as string;
+    const companyId = await assertCaseAccess(db, req, caseId);
+    res.json(await computeRollup(db, companyId, caseId));
+  });
+
+  router.get("/cases/:caseId/context-pack", async (req, res) => {
+    const caseId = req.params.caseId as string;
+    const companyId = await assertCaseAccess(db, req, caseId);
+    const detail = await getCaseDetail(db, companyId, caseId);
+    const events = await svc.listCaseEvents(companyId, caseId);
+    res.json({
+      case: {
+        id: detail.case.id,
+        caseKey: detail.case.caseKey,
+        title: detail.case.title,
+        version: detail.case.version,
+        untrustedContent: {
+          summary: detail.case.summary,
+          fields: detail.case.fields,
+        },
+      },
+      stage: detail.stage,
+      allowedTransitions: detail.allowedNextStages,
+      linkedIssues: detail.links,
+      blockers: detail.blockers,
+      events: events.slice(-20),
+    });
+  });
+
+  router.post("/cases/:caseId/automations/:automationId/retry", async (req, res) => {
+    const caseId = req.params.caseId as string;
+    const automationId = req.params.automationId as string;
+    const companyId = await assertCaseAccess(db, req, caseId);
+    actorForMutation(req);
+    const [execution] = await db.select().from(pipelineAutomationExecutions)
+      .where(and(
+        eq(pipelineAutomationExecutions.companyId, companyId),
+        eq(pipelineAutomationExecutions.caseId, caseId),
+        eq(pipelineAutomationExecutions.automationId, automationId),
+      ))
+      .limit(1);
+    if (!execution) throw notFound("Pipeline automation execution not found");
+    res.json({ execution, retried: false });
+  });
+
+  return router;
+}
+
+async function getCaseDetail(db: Db, companyId: string, caseId: string) {
+  const row = await db
+    .select({ case: pipelineCases, stage: pipelineStages, pipeline: pipelines })
+    .from(pipelineCases)
+    .innerJoin(pipelineStages, eq(pipelineCases.stageId, pipelineStages.id))
+    .innerJoin(pipelines, eq(pipelineCases.pipelineId, pipelines.id))
+    .where(and(eq(pipelineCases.companyId, companyId), eq(pipelineCases.id, caseId)))
+    .limit(1)
+    .then((rows) => rows[0] ?? null);
+  if (!row) throw notFound("Pipeline case not found");
+  const [allowedNextStages, links, blockers, blocks, children] = await Promise.all([
+    db.select().from(pipelineStages).where(eq(pipelineStages.pipelineId, row.case.pipelineId)).orderBy(asc(pipelineStages.position)),
+    db.select().from(pipelineCaseIssueLinks).where(and(eq(pipelineCaseIssueLinks.companyId, companyId), eq(pipelineCaseIssueLinks.caseId, caseId))),
+    db.select().from(pipelineCaseBlockers).where(and(eq(pipelineCaseBlockers.companyId, companyId), eq(pipelineCaseBlockers.caseId, caseId))),
+    db.select().from(pipelineCaseBlockers).where(and(eq(pipelineCaseBlockers.companyId, companyId), eq(pipelineCaseBlockers.blockedByCaseId, caseId))),
+    db.select().from(pipelineCases).where(and(eq(pipelineCases.companyId, companyId), eq(pipelineCases.parentCaseId, caseId))),
+  ]);
+  return {
+    ...row,
+    allowedNextStages,
+    links,
+    blockers,
+    blocks,
+    childrenSummary: {
+      childCount: row.case.childCount,
+      terminalChildCount: row.case.terminalChildCount,
+      loadedChildren: children.length,
+    },
+    pendingSuggestion: row.case.pendingSuggestion,
+  };
+}
+
+async function reviewCase(
+  db: Db,
+  svc: ReturnType<typeof pipelineService>,
+  companyId: string,
+  caseId: string,
+  body: z.infer<typeof reviewCaseSchema>,
+  actor: PipelineActor,
+) {
+  const detail = await getCaseDetail(db, companyId, caseId);
+  if (detail.stage.kind !== "review") {
+    throw unprocessable("Pipeline case is not in a review stage", { code: "validation" });
+  }
+  const config = (detail.stage.config ?? {}) as {
+    approveToStageKey?: string;
+    rejectToStageKey?: string;
+    requireRejectReason?: boolean;
+  };
+  if (body.decision === "reject" && config.requireRejectReason !== false && !body.reason?.trim()) {
+    throw unprocessable("Reject reason is required", { code: "validation" });
+  }
+  const toStageKey = body.decision === "approve" ? config.approveToStageKey : config.rejectToStageKey;
+  if (!toStageKey) throw unprocessable("Review stage is missing decision target config", { code: "validation" });
+  let expectedVersion = body.expectedVersion;
+  if (body.edits && Object.keys(body.edits).length > 0) {
+    const updated = await svc.patchCaseContent({ companyId, caseId, ...body.edits, expectedVersion, actor });
+    expectedVersion = updated.version;
+  }
+  const transitioned = await svc.transitionCase({
+    companyId,
+    caseId,
+    toStageKey,
+    expectedVersion,
+    leaseToken: body.leaseToken,
+    reason: body.reason,
+    actor,
+  });
+  const event = await writeRouteEvent(db, {
+    companyId,
+    caseId,
+    type: "review_decided",
+    actor,
+    payload: { decision: body.decision, reason: body.reason ?? null },
+  });
+  return { ...transitioned, reviewEvent: event };
+}
+
+async function computeRollup(db: Db, companyId: string, caseId: string) {
+  const rows = await db.execute(sql<{
+    id: string;
+    terminal_kind: string | null;
+  }>`
+    with recursive subtree as (
+      select id, terminal_kind from pipeline_cases where company_id = ${companyId} and id = ${caseId}
+      union all
+      select child.id, child.terminal_kind
+      from pipeline_cases child
+      join subtree parent on child.parent_case_id = parent.id
+      where child.company_id = ${companyId}
+    )
+    select id, terminal_kind from subtree
+  `);
+  const items = Array.from(rows);
+  if (items.length === 0) throw notFound("Pipeline case not found");
+  const descendants = items.slice(1);
+  const done = descendants.filter((item) => item.terminal_kind === "done").length;
+  const cancelled = descendants.filter((item) => item.terminal_kind === "cancelled").length;
+  const open = descendants.filter((item) => item.terminal_kind !== "done" && item.terminal_kind !== "cancelled").length;
+  return { total: descendants.length, done, cancelled, open, complete: open === 0 };
+}