From e80ffc86cd666b874c638bcbb6d0719b1e6c7f36 Mon Sep 17 00:00:00 2001 From: Dotta Date: Tue, 9 Jun 2026 22:57:31 +0000 Subject: [PATCH] Add pipeline REST routes Co-Authored-By: Paperclip --- server/src/__tests__/pipelines-routes.test.ts | 291 ++++++ server/src/app.ts | 2 + server/src/middleware/error-handler.ts | 4 + server/src/routes/pipelines.ts | 848 ++++++++++++++++++ 4 files changed, 1145 insertions(+) create mode 100644 server/src/__tests__/pipelines-routes.test.ts create mode 100644 server/src/routes/pipelines.ts diff --git a/server/src/__tests__/pipelines-routes.test.ts b/server/src/__tests__/pipelines-routes.test.ts new file mode 100644 index 00000000000..39225a15c43 --- /dev/null +++ b/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; + let tempDb: Awaited> | 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"); + }); +}); diff --git a/server/src/app.ts b/server/src/app.ts index 40061e117af..af738176f91 100644 --- a/server/src/app.ts +++ b/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)); diff --git a/server/src/middleware/error-handler.ts b/server/src/middleware/error-handler.ts index c455cbad950..c9376882f27 100644 --- a/server/src/middleware/error-handler.ts +++ b/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 + : 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; diff --git a/server/src/routes/pipelines.ts b/server/src/routes/pipelines.ts new file mode 100644 index 00000000000..0bf8cf991ef --- /dev/null +++ b/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, + input: { + companyId: string; + caseId: string; + type: string; + actor: PipelineActor; + payload?: Record; + }, +) { + 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`count(distinct ${pipelineStages.id})::int`, + openCaseCount: sql`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; + }) => ({ + ...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 = { 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["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`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, + companyId: string, + caseId: string, + body: z.infer, + 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 }; +}