diff --git a/server/src/__tests__/pipelines-routes.test.ts b/server/src/__tests__/pipelines-routes.test.ts index 39225a15c43..b22ef129e3e 100644 --- a/server/src/__tests__/pipelines-routes.test.ts +++ b/server/src/__tests__/pipelines-routes.test.ts @@ -8,6 +8,8 @@ import { createDb, documents, documentRevisions, + heartbeatRuns, + issueComments, issues, pipelineAutomationExecutions, pipelineCaseBlockers, @@ -18,6 +20,7 @@ import { pipelineStages, pipelineTransitions, pipelines, + routineRuns, routines, } from "@paperclipai/db"; import { @@ -25,6 +28,7 @@ import { startEmbeddedPostgresTestDatabase, } from "./helpers/embedded-postgres.js"; import { errorHandler } from "../middleware/error-handler.js"; +import { issueRoutes } from "../routes/issues.js"; import { pipelineRoutes } from "../routes/pipelines.js"; const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); @@ -39,6 +43,7 @@ if (!embeddedPostgresSupport.supported) { describeEmbeddedPostgres("pipeline routes", () => { let db!: ReturnType; let tempDb: Awaited> | null = null; + const noopHeartbeat = { wakeup: async () => null }; beforeAll(async () => { tempDb = await startEmbeddedPostgresTestDatabase("paperclip-pipelines-routes-"); @@ -56,6 +61,9 @@ describeEmbeddedPostgres("pipeline routes", () => { await db.delete(pipelineDocuments); await db.delete(documentRevisions); await db.delete(documents); + await db.delete(issueComments); + await db.delete(routineRuns); + await db.delete(heartbeatRuns); await db.delete(issues); await db.delete(pipelines); await db.delete(routines); @@ -74,7 +82,8 @@ describeEmbeddedPostgres("pipeline routes", () => { req.actor = actor; next(); }); - instance.use("/api", pipelineRoutes(db)); + instance.use("/api", pipelineRoutes(db, { heartbeat: noopHeartbeat })); + instance.use("/api", issueRoutes(db, {} as any)); instance.use(errorHandler); return instance; } @@ -182,18 +191,32 @@ describeEmbeddedPostgres("pipeline routes", () => { .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 conversation = await http.post(`/api/cases/${blocked.body.case.id}/open-conversation`).expect(201); + expect(conversation.body.created).toBe(true); + expect(conversation.body.issue.description).toContain("Pipeline Case Context"); + const sameConversation = await http.post(`/api/cases/${blocked.body.case.id}/open-conversation`).expect(200); + expect(sameConversation.body.created).toBe(false); + expect(sameConversation.body.issue.id).toBe(conversation.body.issue.id); - const [agent] = await db.insert(agents).values({ + const linkedIssue = await http.post(`/api/cases/${blocked.body.case.id}/issue-links`) + .send({ issueId: ingested.body.case.id, role: "work" }); + expect(linkedIssue.status).toBe(404); + const manualIssue = await db.insert(issues).values({ companyId: company.id, - name: "Routine Agent", - role: "engineer", - adapterType: "codex_local", - adapterConfig: {}, - runtimeConfig: {}, - permissions: {}, + title: "Manual work issue", + status: "todo", + priority: "medium", }).returning(); - const [routine] = await db.insert(routines).values({ companyId: company.id, title: "Routine", assigneeAgentId: agent!.id }).returning(); + const workLink = await http.post(`/api/cases/${blocked.body.case.id}/issue-links`) + .send({ issueId: manualIssue[0]!.id, role: "work" }) + .expect(201); + await http.get(`/api/cases/${blocked.body.case.id}/issue-links`).expect(200); + const issueDetail = await http.get(`/api/issues/${manualIssue[0]!.id}`).expect(200); + expect(issueDetail.body.linkedCases).toHaveLength(1); + expect(issueDetail.body.linkedCases[0].id).toBe(blocked.body.case.id); + await http.delete(`/api/cases/${blocked.body.case.id}/issue-links/${workLink.body.id}`).expect(200); + + const [routine] = await db.insert(routines).values({ companyId: company.id, title: "Routine" }).returning(); await db.insert(pipelineAutomationExecutions).values({ companyId: company.id, caseId: blocked.body.case.id, @@ -274,6 +297,143 @@ describeEmbeddedPostgres("pipeline routes", () => { expect(res.body.code).toBe("review_required"); }); + it("validates review stage config on create and update", async () => { + const company = await seedCompany(); + const http = request(app(boardActor)); + + await http + .post(`/api/companies/${company.id}/pipelines`) + .send({ + key: "bad-review", + name: "Bad review", + stages: [ + { key: "intake", name: "Intake", kind: "open", position: 100 }, + { key: "review", name: "Review", kind: "review", position: 200 }, + { key: "done", name: "Done", kind: "done", position: 900 }, + { key: "cancelled", name: "Cancelled", kind: "cancelled", position: 1000 }, + ], + }) + .expect(422); + + const pipeline = await http.post(`/api/companies/${company.id}/pipelines`).send({ key: "valid-review", name: "Valid review" }).expect(201); + const intake = pipeline.body.stages.find((stage: { key: string }) => stage.key === "intake"); + await http.patch(`/api/pipelines/${pipeline.body.id}/stages/${intake.id}`).send({ kind: "review" }).expect(422); + }); + + it("applies review decisions atomically with edits and stores reject reasons verbatim", async () => { + const company = await seedCompany(); + const http = request(app(boardActor)); + const pipeline = await http.post(`/api/companies/${company.id}/pipelines`).send({ key: "review-decisions", name: "Review decisions" }).expect(201); + + const approved = await http + .post(`/api/pipelines/${pipeline.body.id}/cases`) + .send({ caseKey: "approve-edit", title: "Approve edit" }) + .expect(201); + await http.post(`/api/cases/${approved.body.case.id}/transition`).send({ toStageKey: "review", expectedVersion: 1 }).expect(200); + const approval = await http + .post(`/api/cases/${approved.body.case.id}/review`) + .send({ + decision: "approve", + expectedVersion: 2, + edits: { title: "Approved title", fields: { channel: "blog" } }, + }) + .expect(200); + expect(approval.body.case.version).toBe(4); + expect(approval.body.updateEvent.payload.version).toBe(3); + const approvedDetail = await http.get(`/api/cases/${approved.body.case.id}`).expect(200); + expect(approvedDetail.body.case.title).toBe("Approved title"); + expect(approvedDetail.body.case.fields).toEqual({ channel: "blog" }); + const approvedEvents = await http.get(`/api/cases/${approved.body.case.id}/events`).expect(200); + expect(approvedEvents.body.map((event: { type: string }) => event.type)).toEqual([ + "ingested", + "transitioned", + "updated", + "transitioned", + "review_decided", + ]); + + const rejected = await http + .post(`/api/pipelines/${pipeline.body.id}/cases`) + .send({ caseKey: "reject-reason", title: "Reject reason" }) + .expect(201); + await http.post(`/api/cases/${rejected.body.case.id}/transition`).send({ toStageKey: "review", expectedVersion: 1 }).expect(200); + await http.post(`/api/cases/${rejected.body.case.id}/review`).send({ decision: "reject", expectedVersion: 2 }).expect(422); + const reason = " Keep this exact reason. "; + await http.post(`/api/cases/${rejected.body.case.id}/review`).send({ decision: "reject", reason, expectedVersion: 2 }).expect(200); + const rejectedEvents = await http.get(`/api/cases/${rejected.body.case.id}/events`).expect(200); + const reviewEvent = rejectedEvents.body.find((event: { type: string }) => event.type === "review_decided"); + expect(reviewEvent.payload.reason).toBe(reason); + }); + + it("aggregates the review inbox across pipelines with parent and review config context", async () => { + const company = await seedCompany(); + const http = request(app(boardActor)); + const first = await http.post(`/api/companies/${company.id}/pipelines`).send({ key: "inbox-a", name: "Inbox A" }).expect(201); + const second = await http.post(`/api/companies/${company.id}/pipelines`).send({ key: "inbox-b", name: "Inbox B" }).expect(201); + + const parent = await http + .post(`/api/pipelines/${first.body.id}/cases`) + .send({ caseKey: "parent", title: "Parent" }) + .expect(201); + const child = await http + .post(`/api/pipelines/${first.body.id}/cases`) + .send({ caseKey: "child", title: "Child", parentCaseId: parent.body.case.id }) + .expect(201); + await http.post(`/api/cases/${child.body.case.id}/transition`).send({ toStageKey: "review", expectedVersion: 1 }).expect(200); + + const other = await http + .post(`/api/pipelines/${second.body.id}/cases`) + .send({ caseKey: "other", title: "Other" }) + .expect(201); + await http.post(`/api/cases/${other.body.case.id}/transition`).send({ toStageKey: "review", expectedVersion: 1 }).expect(200); + + const notReview = await http + .post(`/api/pipelines/${second.body.id}/cases`) + .send({ caseKey: "not-review", title: "Not review" }) + .expect(201); + await http.post(`/api/cases/${notReview.body.case.id}/transition`).send({ toStageKey: "done", expectedVersion: 1 }).expect(200); + + const inbox = await http.get(`/api/companies/${company.id}/review-cases`).expect(200); + expect(inbox.body).toHaveLength(2); + expect(inbox.body.map((row: { pipeline: { key: string } }) => row.pipeline.key).sort()).toEqual(["inbox-a", "inbox-b"]); + const childRow = inbox.body.find((row: { case: { id: string } }) => row.case.id === child.body.case.id); + expect(childRow.parentCase.id).toBe(parent.body.case.id); + expect(childRow.reviewConfig).toMatchObject({ + approveToStageKey: "done", + rejectToStageKey: "cancelled", + requireRejectReason: true, + reviewerKind: "human", + }); + }); + + it("bulk reviews partial successes without aborting stale items", async () => { + const company = await seedCompany(); + const http = request(app(boardActor)); + const pipeline = await http.post(`/api/companies/${company.id}/pipelines`).send({ key: "bulk-review", name: "Bulk review" }).expect(201); + const caseIds: string[] = []; + for (let index = 0; index < 50; index += 1) { + const created = await http + .post(`/api/pipelines/${pipeline.body.id}/cases`) + .send({ caseKey: `bulk-${index}`, title: `Bulk ${index}` }) + .expect(201); + await http.post(`/api/cases/${created.body.case.id}/transition`).send({ toStageKey: "review", expectedVersion: 1 }).expect(200); + caseIds.push(created.body.case.id); + } + for (const staleCaseId of caseIds.slice(0, 3)) { + await http.patch(`/api/cases/${staleCaseId}`).send({ title: "Stale before bulk", expectedVersion: 2 }).expect(200); + } + + const bulk = await http + .post(`/api/companies/${company.id}/review-cases/bulk`) + .send({ items: caseIds.map((caseId) => ({ caseId, decision: "approve", expectedVersion: 2 })) }) + .expect(200); + + expect(bulk.body.results.filter((item: { ok: boolean }) => item.ok)).toHaveLength(47); + const failed = bulk.body.results.filter((item: { ok: boolean }) => !item.ok); + expect(failed).toHaveLength(3); + expect(failed.every((item: { error: { code: string } }) => item.error.code === "version_conflict")).toBe(true); + }); + it("returns conflict bodies with code, current version, and stage", async () => { const company = await seedCompany(); const http = request(app(boardActor)); diff --git a/server/src/__tests__/pipelines-service.test.ts b/server/src/__tests__/pipelines-service.test.ts index 2dfb62844bf..91129671b6e 100644 --- a/server/src/__tests__/pipelines-service.test.ts +++ b/server/src/__tests__/pipelines-service.test.ts @@ -5,13 +5,18 @@ import { agents, companies, createDb, + heartbeatRuns, + issueComments, + issues, pipelineAutomationExecutions, pipelineCaseBlockers, + pipelineCaseIssueLinks, pipelineCaseEvents, pipelineCases, pipelineStages, pipelineTransitions, pipelines, + routineRuns, routines, } from "@paperclipai/db"; import { @@ -19,6 +24,7 @@ import { startEmbeddedPostgresTestDatabase, } from "./helpers/embedded-postgres.js"; import { pipelineService, type PipelineActor } from "../services/pipelines.ts"; +import { routineService } from "../services/routines.ts"; const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; @@ -35,21 +41,27 @@ describeEmbeddedPostgres("pipelineService", () => { let tempDb: Awaited> | null = null; const userActor: PipelineActor = { type: "user", userId: "board-user" }; + const noopHeartbeat = { wakeup: async () => null }; beforeAll(async () => { tempDb = await startEmbeddedPostgresTestDatabase("paperclip-pipelines-service-"); db = createDb(tempDb.connectionString); - svc = pipelineService(db); + svc = pipelineService(db, { heartbeat: noopHeartbeat }); }, 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(pipelines); + await db.delete(issueComments); + await db.delete(routineRuns); + await db.delete(heartbeatRuns); + await db.delete(issues); await db.delete(routines); await db.delete(agents); await db.delete(companies); @@ -80,6 +92,30 @@ describeEmbeddedPostgres("pipelineService", () => { return { company, pipeline, stages, byKey: new Map(stages.map((stage) => [stage.key, stage])) }; } + async function seedRoutine(companyId: string, title = "Routine") { + const [agent] = await db.insert(agents).values({ + companyId, + name: `${title} Agent`, + role: "engineer", + adapterType: "codex_local", + adapterConfig: {}, + runtimeConfig: {}, + permissions: {}, + }).returning(); + return routineService(db, { heartbeat: noopHeartbeat }).create(companyId, { + projectId: null, + goalId: null, + parentIssueId: null, + title, + description: null, + assigneeAgentId: agent!.id, + priority: "medium", + status: "active", + concurrencyPolicy: "always_enqueue", + catchUpPolicy: "skip_missed", + }, {}); + } + async function eventCount(caseId: string) { const [{ count }] = await db .select({ count: sql`count(*)::int` }) @@ -336,10 +372,11 @@ describeEmbeddedPostgres("pipelineService", () => { title: "Blocking case", actor: userActor, }); - await db.insert(pipelineCaseBlockers).values({ + await svc.replaceBlockers({ companyId: company.id, caseId: blocked.case.id, - blockedByCaseId: blocker.case.id, + blockedByCaseIds: [blocker.case.id], + actor: userActor, }); await expect( @@ -352,6 +389,25 @@ describeEmbeddedPostgres("pipelineService", () => { }), ).rejects.toMatchObject({ status: 409, details: { code: "blocked" } }); + const reviewMove = await svc.transitionCase({ + companyId: company.id, + caseId: blocked.case.id, + toStageKey: "review", + expectedVersion: 1, + actor: userActor, + }); + expect(reviewMove.case.version).toBe(2); + + await expect( + svc.transitionCase({ + companyId: company.id, + caseId: blocked.case.id, + toStageKey: "done", + expectedVersion: 2, + actor: userActor, + }), + ).rejects.toMatchObject({ status: 409, details: { code: "blocked" } }); + await svc.transitionCase({ companyId: company.id, caseId: blocker.case.id, @@ -359,6 +415,59 @@ describeEmbeddedPostgres("pipelineService", () => { expectedVersion: 1, actor: userActor, }); + const moved = await svc.transitionCase({ + companyId: company.id, + caseId: blocked.case.id, + toStageKey: "in_progress", + expectedVersion: 2, + actor: userActor, + }); + expect(moved.case.version).toBe(3); + const events = await svc.listCaseEvents(company.id, blocked.case.id); + expect(events.map((event) => event.type)).toContain("blockers_resolved"); + }); + + it("keeps cancelled blockers unsatisfied until replaced", async () => { + const { company, pipeline } = await seedPipeline(); + const blocked = await svc.ingestCase({ + companyId: company.id, + pipelineId: pipeline.id, + caseKey: "blocked-cancelled", + title: "Blocked case", + actor: userActor, + }); + const blocker = await svc.ingestCase({ + companyId: company.id, + pipelineId: pipeline.id, + caseKey: "blocker-cancelled", + title: "Cancelled blocker", + actor: userActor, + }); + await svc.replaceBlockers({ + companyId: company.id, + caseId: blocked.case.id, + blockedByCaseIds: [blocker.case.id], + actor: userActor, + }); + await svc.transitionCase({ + companyId: company.id, + caseId: blocker.case.id, + toStageKey: "cancelled", + expectedVersion: 1, + actor: userActor, + }); + + await expect( + svc.transitionCase({ + companyId: company.id, + caseId: blocked.case.id, + toStageKey: "in_progress", + expectedVersion: 1, + actor: userActor, + }), + ).rejects.toMatchObject({ status: 409, details: { code: "blocked" } }); + + await svc.replaceBlockers({ companyId: company.id, caseId: blocked.case.id, blockedByCaseIds: [], actor: userActor }); const moved = await svc.transitionCase({ companyId: company.id, caseId: blocked.case.id, @@ -369,6 +478,172 @@ describeEmbeddedPostgres("pipelineService", () => { expect(moved.case.version).toBe(2); }); + it("rejects parent and blocker cycles and enforces parent depth", async () => { + const { company, pipeline } = await seedPipeline(); + const a = await svc.ingestCase({ companyId: company.id, pipelineId: pipeline.id, caseKey: "a", title: "A", actor: userActor }); + const b = await svc.ingestCase({ + companyId: company.id, + pipelineId: pipeline.id, + caseKey: "b", + title: "B", + parentCaseId: a.case.id, + actor: userActor, + }); + + await expect( + svc.patchCaseContent({ + companyId: company.id, + caseId: a.case.id, + parentCaseId: b.case.id, + expectedVersion: 1, + actor: userActor, + }), + ).rejects.toMatchObject({ status: 409, details: { code: "parent_cycle" } }); + + await svc.replaceBlockers({ companyId: company.id, caseId: a.case.id, blockedByCaseIds: [b.case.id], actor: userActor }); + await expect( + svc.replaceBlockers({ companyId: company.id, caseId: b.case.id, blockedByCaseIds: [a.case.id], actor: userActor }), + ).rejects.toMatchObject({ status: 409, details: { code: "blocker_cycle" } }); + + let parentCaseId: string | null = null; + for (let index = 0; index < 32; index += 1) { + const created = await svc.ingestCase({ + companyId: company.id, + pipelineId: pipeline.id, + caseKey: `chain-${index}`, + title: `Chain ${index}`, + parentCaseId, + actor: userActor, + }); + parentCaseId = created.case.id; + } + await expect( + svc.ingestCase({ + companyId: company.id, + pipelineId: pipeline.id, + caseKey: "too-deep", + title: "Too deep", + parentCaseId, + actor: userActor, + }), + ).rejects.toMatchObject({ status: 422, details: { code: "parent_depth_exceeded" } }); + }); + + it("rolls up a three-level tree, updates counters, and emits children_terminal once", async () => { + const { company, pipeline } = await seedPipeline(); + const root = await svc.ingestCase({ companyId: company.id, pipelineId: pipeline.id, caseKey: "root", title: "Root", actor: userActor }); + const [linkedIssue] = await db.insert(issues).values({ + companyId: company.id, + title: "Root conversation", + status: "todo", + priority: "medium", + }).returning(); + await db.insert(pipelineCaseIssueLinks).values({ + companyId: company.id, + caseId: root.case.id, + issueId: linkedIssue!.id, + role: "conversation", + }); + const childA = await svc.ingestCase({ + companyId: company.id, + pipelineId: pipeline.id, + caseKey: "child-a", + title: "Child A", + parentCaseId: root.case.id, + actor: userActor, + }); + const childB = await svc.ingestCase({ + companyId: company.id, + pipelineId: pipeline.id, + caseKey: "child-b", + title: "Child B", + parentCaseId: root.case.id, + actor: userActor, + }); + const childC = await svc.ingestCase({ + companyId: company.id, + pipelineId: pipeline.id, + caseKey: "child-c", + title: "Child C", + parentCaseId: root.case.id, + actor: userActor, + }); + const grandA = await svc.ingestCase({ + companyId: company.id, + pipelineId: pipeline.id, + caseKey: "grand-a", + title: "Grand A", + parentCaseId: childA.case.id, + actor: userActor, + }); + const grandB = await svc.ingestCase({ + companyId: company.id, + pipelineId: pipeline.id, + caseKey: "grand-b", + title: "Grand B", + parentCaseId: childA.case.id, + actor: userActor, + }); + + await svc.transitionCase({ companyId: company.id, caseId: childB.case.id, toStageKey: "done", expectedVersion: 1, actor: userActor }); + await svc.transitionCase({ companyId: company.id, caseId: childC.case.id, toStageKey: "done", expectedVersion: 1, actor: userActor }); + await svc.transitionCase({ companyId: company.id, caseId: grandA.case.id, toStageKey: "done", expectedVersion: 1, actor: userActor }); + await svc.transitionCase({ companyId: company.id, caseId: grandB.case.id, toStageKey: "cancelled", expectedVersion: 1, actor: userActor }); + await svc.transitionCase({ companyId: company.id, caseId: childA.case.id, toStageKey: "done", expectedVersion: 1, actor: userActor }); + + expect(await svc.getCaseRollup(company.id, root.case.id)).toEqual({ + total: 5, + done: 4, + cancelled: 1, + open: 0, + complete: true, + }); + const [freshRoot] = await db.select().from(pipelineCases).where(eq(pipelineCases.id, root.case.id)); + const [freshChildA] = await db.select().from(pipelineCases).where(eq(pipelineCases.id, childA.case.id)); + expect(freshRoot!.childCount).toBe(3); + expect(freshRoot!.terminalChildCount).toBe(3); + expect(freshChildA!.childCount).toBe(2); + expect(freshChildA!.terminalChildCount).toBe(2); + const rootEvents = await svc.listCaseEvents(company.id, root.case.id); + expect(rootEvents.filter((event) => event.type === "children_terminal")).toHaveLength(1); + const comments = await db.select().from(issueComments).where(eq(issueComments.issueId, linkedIssue!.id)); + expect(comments).toHaveLength(1); + expect(comments[0]!.authorType).toBe("system"); + expect(comments[0]!.body).toContain("All child cases"); + }); + + it("auto-advances a parent when all descendants are terminal", async () => { + const company = await seedCompany(); + const pipeline = await svc.createPipeline({ + companyId: company.id, + key: "auto-children", + name: "Auto children", + actor: userActor, + stages: [ + { key: "intake", name: "Intake", kind: "open", config: { autoAdvanceOnChildrenTerminal: "done" } }, + { key: "done", name: "Done", kind: "done" }, + { key: "cancelled", name: "Cancelled", kind: "cancelled" }, + ], + }); + const root = await svc.ingestCase({ companyId: company.id, pipelineId: pipeline.id, caseKey: "auto-root", title: "Root", actor: userActor }); + const child = await svc.ingestCase({ + companyId: company.id, + pipelineId: pipeline.id, + caseKey: "auto-child", + title: "Child", + parentCaseId: root.case.id, + actor: userActor, + }); + + await svc.transitionCase({ companyId: company.id, caseId: child.case.id, toStageKey: "done", expectedVersion: 1, actor: userActor }); + + const [freshRoot] = await db.select().from(pipelineCases).where(eq(pipelineCases.id, root.case.id)); + expect(freshRoot!.terminalKind).toBe("done"); + expect(freshRoot!.version).toBe(2); + const rootEvents = await svc.listCaseEvents(company.id, root.case.id); + expect(rootEvents.map((event) => event.type)).toEqual(["ingested", "children_terminal", "transitioned"]); + }); + it("records suggestion supersede, accept, and dismiss lifecycles", async () => { const { company, pipeline } = await seedPipeline(); const created = await svc.ingestCase({ @@ -476,22 +751,9 @@ describeEmbeddedPostgres("pipelineService", () => { expect(await eventCount(created.case.id)).toBe(5); }); - it("creates a stage-entry automation ledger idempotently with the transition event", async () => { + it("fires a stage-entry automation routine once and keeps crash-retry idempotent", async () => { const company = await seedCompany(); - 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: "Draft on enter", - assigneeAgentId: agent!.id, - }).returning(); + const routine = await seedRoutine(company.id, "Draft on enter"); const pipeline = await svc.createPipeline({ companyId: company.id, key: "automation", @@ -499,7 +761,7 @@ describeEmbeddedPostgres("pipelineService", () => { actor: userActor, stages: [ { key: "intake", name: "Intake", kind: "open" }, - { key: "drafting", name: "Drafting", kind: "working", config: { onEnter: { type: "run_routine", routineId: routine!.id } } }, + { key: "drafting", name: "Drafting", kind: "working", config: { onEnter: { type: "run_routine", routineId: routine.id } } }, { key: "done", name: "Done", kind: "done" }, { key: "cancelled", name: "Cancelled", kind: "cancelled" }, ], @@ -519,9 +781,148 @@ describeEmbeddedPostgres("pipelineService", () => { expectedVersion: 1, actor: userActor, }); - expect(moved.automationLedger?.routineId).toBe(routine!.id); + expect(moved.automationLedger?.routineId).toBe(routine.id); + expect(moved.automationExecution.status).toBe("succeeded"); const ledgers = await db.select().from(pipelineAutomationExecutions); expect(ledgers).toHaveLength(1); expect(ledgers[0]!.triggeringEventId).toBe(moved.event.id); + expect(ledgers[0]!.executionIssueId).toBeTruthy(); + const runsAfterTransition = await db.select().from(routineRuns); + expect(runsAfterTransition).toHaveLength(1); + const linksAfterTransition = await db.select().from(pipelineCaseIssueLinks); + expect(linksAfterTransition).toHaveLength(1); + expect(linksAfterTransition[0]!.role).toBe("automation"); + + const [issue] = await db.select().from(issues).where(eq(issues.id, ledgers[0]!.executionIssueId!)); + expect(issue!.description).toContain("Pipeline Case Context"); + expect(issue!.description).toContain("untrustedContent"); + + const triggerEvent = await db.insert(pipelineCaseEvents).values({ + companyId: company.id, + caseId: created.case.id, + type: "transitioned", + actorType: "system", + toStageId: moved.case.stageId, + payload: { simulatedCrash: true }, + }).returning(); + const automationId = ledgers[0]!.automationId; + await db.insert(pipelineAutomationExecutions).values({ + companyId: company.id, + caseId: created.case.id, + automationId, + triggeringEventId: triggerEvent[0]!.id, + routineId: routine.id, + status: "failed", + error: "pending_dispatch", + }); + + const firstRetry = await svc.retryAutomation({ + companyId: company.id, + caseId: created.case.id, + automationId, + actor: userActor, + }); + const secondRetry = await svc.retryAutomation({ + companyId: company.id, + caseId: created.case.id, + automationId, + actor: userActor, + }); + expect(firstRetry.status).toBe("succeeded"); + expect(secondRetry.status).toBe("succeeded"); + const runsAfterRetries = await db.select().from(routineRuns); + expect(runsAfterRetries).toHaveLength(2); + const crashExecutions = await db + .select() + .from(pipelineAutomationExecutions) + .where(eq(pipelineAutomationExecutions.triggeringEventId, triggerEvent[0]!.id)); + expect(crashExecutions).toHaveLength(1); + expect(crashExecutions[0]!.executionIssueId).toBeTruthy(); + const crashLinks = await db + .select() + .from(pipelineCaseIssueLinks) + .where(eq(pipelineCaseIssueLinks.issueId, crashExecutions[0]!.executionIssueId!)); + expect(crashLinks).toHaveLength(1); + }); + + it("rejects cross-company stage automation routines at save and execution", async () => { + const company = await seedCompany(); + const otherCompany = await seedCompany(); + const routine = await seedRoutine(company.id, "Own routine"); + const otherRoutine = await seedRoutine(otherCompany.id, "Other routine"); + + await expect(svc.createPipeline({ + companyId: company.id, + key: "bad-automation", + name: "Bad automation", + actor: userActor, + stages: [ + { key: "intake", name: "Intake", kind: "open" }, + { key: "drafting", name: "Drafting", kind: "working", config: { onEnter: { type: "run_routine", routineId: otherRoutine.id } } }, + { key: "done", name: "Done", kind: "done" }, + { key: "cancelled", name: "Cancelled", kind: "cancelled" }, + ], + })).rejects.toMatchObject({ status: 422, details: { code: "validation" } }); + + const pipeline = await svc.createPipeline({ + companyId: company.id, + key: "execution-automation", + name: "Execution automation", + actor: userActor, + stages: [ + { key: "intake", name: "Intake", kind: "open" }, + { key: "drafting", name: "Drafting", kind: "working", config: { onEnter: { type: "run_routine", routineId: routine.id } } }, + { key: "done", name: "Done", kind: "done" }, + { key: "cancelled", name: "Cancelled", kind: "cancelled" }, + ], + }); + const created = await svc.ingestCase({ + companyId: company.id, + pipelineId: pipeline.id, + caseKey: "cross-company-execution", + title: "Cross-company execution", + actor: userActor, + }); + const moved = await svc.transitionCase({ + companyId: company.id, + caseId: created.case.id, + toStageKey: "drafting", + expectedVersion: 1, + actor: userActor, + }); + expect(moved.automationExecution.status).toBe("succeeded"); + + const [triggerEvent] = await db.insert(pipelineCaseEvents).values({ + companyId: company.id, + caseId: created.case.id, + type: "transitioned", + actorType: "system", + toStageId: moved.case.stageId, + payload: { crossCompanyRoutine: true }, + }).returning(); + const [badExecution] = await db.insert(pipelineAutomationExecutions).values({ + companyId: company.id, + caseId: created.case.id, + automationId: moved.automationLedger!.automationId, + triggeringEventId: triggerEvent!.id, + routineId: otherRoutine.id, + status: "failed", + error: "pending_dispatch", + }).returning(); + + const retried = await svc.retryAutomation({ + companyId: company.id, + caseId: created.case.id, + automationId: moved.automationLedger!.automationId, + actor: userActor, + }); + expect(retried.status).toBe("failed"); + const [execution] = await db + .select() + .from(pipelineAutomationExecutions) + .where(eq(pipelineAutomationExecutions.id, badExecution!.id)); + expect(execution!.error).toContain("same company"); + const events = await svc.listCaseEvents(company.id, created.case.id); + expect(events.filter((event) => event.type === "automation_failed")).toHaveLength(1); }); }); diff --git a/server/src/routes/issues.ts b/server/src/routes/issues.ts index 6241b79f968..559665c2072 100644 --- a/server/src/routes/issues.ts +++ b/server/src/routes/issues.ts @@ -16,6 +16,10 @@ import { issueRelations, issues as issueRows, issueWorkProducts, + pipelineCaseIssueLinks, + pipelineCases, + pipelineStages, + pipelines, projectWorkspaces, } from "@paperclipai/db"; import { @@ -149,6 +153,44 @@ const promoteLowTrustOutputSchema = z.object({ summary: z.string().trim().min(1).max(8_000), }); +async function listIssueLinkedCases(db: Db, companyId: string, issueId: string) { + const rows = await db + .select({ + link: pipelineCaseIssueLinks, + case: pipelineCases, + pipeline: pipelines, + stage: pipelineStages, + }) + .from(pipelineCaseIssueLinks) + .innerJoin(pipelineCases, eq(pipelineCaseIssueLinks.caseId, pipelineCases.id)) + .innerJoin(pipelines, eq(pipelineCases.pipelineId, pipelines.id)) + .innerJoin(pipelineStages, eq(pipelineCases.stageId, pipelineStages.id)) + .where(and( + eq(pipelineCaseIssueLinks.companyId, companyId), + eq(pipelineCaseIssueLinks.issueId, issueId), + eq(pipelineCases.companyId, companyId), + eq(pipelines.companyId, companyId), + )); + return rows.map((row) => ({ + id: row.case.id, + caseKey: row.case.caseKey, + title: row.case.title, + status: row.case.terminalKind ?? "open", + role: row.link.role, + pipeline: { + id: row.pipeline.id, + key: row.pipeline.key, + name: row.pipeline.name, + }, + stage: { + id: row.stage.id, + key: row.stage.key, + name: row.stage.name, + kind: row.stage.kind, + }, + })); +} + type ParsedExecutionState = NonNullable>; type NormalizedExecutionPolicy = NonNullable>; type IssueRouteSnapshot = typeof issueRows.$inferSelect; @@ -2721,6 +2763,7 @@ export function issueRoutes( successfulRunHandoffStates, scheduledRetry, activeRecoveryAction, + linkedCases, ] = await Promise.all([ resolveIssueProjectAndGoal(issue), svc.getAncestors(issue.id), @@ -2733,6 +2776,7 @@ export function issueRoutes( listSuccessfulRunHandoffStates(db, issue.companyId, [issue.id]), svc.getCurrentScheduledRetry(issue.id), recoveryActionsSvc.getActiveForIssue(issue.companyId, issue.id), + listIssueLinkedCases(db, issue.companyId, issue.id), ]); const recoveryActionsByRelationIssue = await relationRecoveryActionMap( recoveryActionsSvc, @@ -2775,6 +2819,7 @@ export function issueRoutes( mentionedProjects, currentExecutionWorkspace, workProducts, + linkedCases, }); }); diff --git a/server/src/routes/pipelines.ts b/server/src/routes/pipelines.ts index 0bf8cf991ef..b933584ea55 100644 --- a/server/src/routes/pipelines.ts +++ b/server/src/routes/pipelines.ts @@ -1,6 +1,6 @@ import { Router, type Request } from "express"; import { z } from "zod"; -import { and, asc, eq, ilike, inArray, isNotNull, isNull, or, sql } from "drizzle-orm"; +import { and, asc, eq, ilike, isNotNull, isNull, or, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { documents, @@ -29,6 +29,7 @@ const casePatchSchema = z.object({ summary: z.string().max(8_000).nullable().optional(), fields: jsonObjectSchema.optional(), workspaceRef: jsonObjectSchema.nullable().optional(), + parentCaseId: z.string().uuid().nullable().optional(), expectedVersion: z.number().int().positive().optional(), leaseToken: z.string().uuid().nullable().optional(), }); @@ -109,14 +110,25 @@ const resolveSuggestionSchema = z.object({ reason: z.string().max(4_000).nullable().optional(), leaseToken: z.string().uuid().nullable().optional(), }); +const reviewEditsSchema = z.object({ + title: z.string().trim().min(1).max(500).optional(), + summary: z.string().max(8_000).nullable().optional(), + fields: jsonObjectSchema.optional(), + parentCaseId: 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(), + edits: reviewEditsSchema.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 issueLinkRoleSchema = z.enum(["origin", "conversation", "work", "automation"]); +const createIssueLinkSchema = z.object({ + issueId: z.string().uuid(), + role: issueLinkRoleSchema, +}); const bulkReviewSchema = z.object({ items: z.array(reviewCaseSchema.extend({ caseId: z.string().uuid() })).max(100), }); @@ -227,9 +239,9 @@ async function writeRouteEvent( return event!; } -export function pipelineRoutes(db: Db) { +export function pipelineRoutes(db: Db, options: Parameters[1] = {}) { const router = Router(); - const svc = pipelineService(db); + const svc = pipelineService(db, options); router.get("/companies/:companyId/pipelines", async (req, res) => { const companyId = req.params.companyId as string; @@ -285,20 +297,7 @@ export function pipelineRoutes(db: Db) { 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); + res.json(await svc.listReviewCases({ companyId, pipelineId, parentCaseId })); }); router.post("/companies/:companyId/review-cases/bulk", validate(bulkReviewSchema), async (req, res) => { @@ -308,13 +307,21 @@ export function pipelineRoutes(db: Db) { 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) }); + results.push({ caseId: item.caseId, ok: true, result: await svc.reviewCase({ companyId, ...item, actor }) }); } catch (error) { const httpError = error as { status?: number; message?: string; details?: unknown }; + const details = httpError.details && typeof httpError.details === "object" && !Array.isArray(httpError.details) + ? httpError.details as Record + : null; results.push({ caseId: item.caseId, ok: false, - error: { status: httpError.status ?? 500, message: httpError.message ?? "Unknown error", details: httpError.details }, + error: { + status: httpError.status ?? 500, + message: httpError.message ?? "Unknown error", + code: typeof details?.code === "string" ? details.code : undefined, + details: httpError.details, + }, }); } } @@ -376,16 +383,10 @@ export function pipelineRoutes(db: Db) { 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); + const companyId = 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); + res.json(await svc.updateStage({ companyId, pipelineId, stageId, patch: req.body })); } catch (error) { codedConflictForUnique(error); } @@ -581,7 +582,7 @@ export function pipelineRoutes(db: Db) { leaseToken: req.body.leaseToken, reason: req.body.reason, suggestionId: req.body.acceptSuggestionId, - actor, + actor, })); }); @@ -612,29 +613,14 @@ export function pipelineRoutes(db: Db) { 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)); + res.json(await svc.reviewCase({ 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 }); + res.json(await svc.replaceBlockers({ companyId, caseId, blockedByCaseIds: req.body.blockedByCaseIds, actor })); }); router.post("/cases/:caseId/open-conversation", async (req, res) => { @@ -663,7 +649,7 @@ export function pipelineRoutes(db: Db) { 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}`, + description: buildCaseContextMarkdown(detail), status: "todo", priority: "medium", originKind: "pipeline_case_conversation", @@ -690,6 +676,71 @@ export function pipelineRoutes(db: Db) { res.status(201).json({ issue: result, created: true }); }); + router.get("/cases/:caseId/issue-links", async (req, res) => { + const caseId = req.params.caseId as string; + const companyId = await assertCaseAccess(db, req, caseId); + const links = await db + .select({ link: pipelineCaseIssueLinks, issue: issueRows }) + .from(pipelineCaseIssueLinks) + .innerJoin(issueRows, eq(pipelineCaseIssueLinks.issueId, issueRows.id)) + .where(and( + eq(pipelineCaseIssueLinks.companyId, companyId), + eq(pipelineCaseIssueLinks.caseId, caseId), + eq(issueRows.companyId, companyId), + )) + .orderBy(asc(pipelineCaseIssueLinks.createdAt)); + res.json(links); + }); + + router.post("/cases/:caseId/issue-links", validate(createIssueLinkSchema), async (req, res) => { + const caseId = req.params.caseId as string; + const companyId = await assertCaseAccess(db, req, caseId); + const actor = actorForMutation(req); + const targetIssue = await db + .select({ id: issueRows.id, companyId: issueRows.companyId }) + .from(issueRows) + .where(eq(issueRows.id, req.body.issueId)) + .limit(1) + .then((rows) => rows[0] ?? null); + if (!targetIssue || targetIssue.companyId !== companyId) throw notFound("Issue not found"); + try { + const [link] = await db.insert(pipelineCaseIssueLinks).values({ + companyId, + caseId, + issueId: req.body.issueId, + role: req.body.role, + createdByRunId: actor.type === "agent" ? actor.runId : null, + }).returning(); + await writeRouteEvent(db, { + companyId, + caseId, + type: "issue_linked", + actor, + payload: { issueId: req.body.issueId, role: req.body.role }, + }); + res.status(201).json(link); + } catch (error) { + codedConflictForUnique(error); + } + }); + + router.delete("/cases/:caseId/issue-links/:linkId", async (req, res) => { + const caseId = req.params.caseId as string; + const linkId = req.params.linkId as string; + const companyId = await assertCaseAccess(db, req, caseId); + actorForMutation(req); + const [deleted] = await db + .delete(pipelineCaseIssueLinks) + .where(and( + eq(pipelineCaseIssueLinks.id, linkId), + eq(pipelineCaseIssueLinks.companyId, companyId), + eq(pipelineCaseIssueLinks.caseId, caseId), + )) + .returning(); + if (!deleted) throw notFound("Pipeline case issue link not found"); + res.json({ deleted: true }); + }); + router.get("/cases/:caseId/events", async (req, res) => { const caseId = req.params.caseId as string; const companyId = await assertCaseAccess(db, req, caseId); @@ -699,7 +750,7 @@ export function pipelineRoutes(db: Db) { 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)); + res.json(await svc.getCaseRollup(companyId, caseId)); }); router.get("/cases/:caseId/context-pack", async (req, res) => { @@ -730,16 +781,8 @@ export function pipelineRoutes(db: Db) { 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 }); + const actor = actorForMutation(req); + res.json(await svc.retryAutomation({ companyId, caseId, automationId, actor })); }); return router; @@ -777,72 +820,39 @@ async function getCaseDetail(db: Db, companyId: string, caseId: string) { }; } -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 }; +function buildCaseContextMarkdown(detail: Awaited>) { + return [ + "## Pipeline Case Context", + "", + `Case: ${detail.case.title}`, + `Pipeline: ${detail.pipeline.name} (${detail.pipeline.key})`, + `Stage: ${detail.stage.name} (${detail.stage.key}, ${detail.stage.kind})`, + `Case link: /PAP/pipelines/${detail.pipeline.id}/cases/${detail.case.id}`, + "", + "```json", + JSON.stringify({ + pipeline: { + id: detail.pipeline.id, + key: detail.pipeline.key, + name: detail.pipeline.name, + }, + 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: { + id: detail.stage.id, + key: detail.stage.key, + name: detail.stage.name, + kind: detail.stage.kind, + }, + }, null, 2), + "```", + ].join("\n"); } diff --git a/server/src/services/pipelines.ts b/server/src/services/pipelines.ts index 7789e63473c..0c0bb0cf64d 100644 --- a/server/src/services/pipelines.ts +++ b/server/src/services/pipelines.ts @@ -1,16 +1,23 @@ import { randomUUID } from "node:crypto"; -import { and, asc, eq, isNull, ne, or, sql } from "drizzle-orm"; +import { and, asc, eq, inArray, isNull, ne, or, sql } from "drizzle-orm"; +import { alias } from "drizzle-orm/pg-core"; import type { Db } from "@paperclipai/db"; import { + issueComments, + issues, pipelineAutomationExecutions, pipelineCaseBlockers, pipelineCaseEvents, + pipelineCaseIssueLinks, pipelineCases, pipelineStages, pipelineTransitions, pipelines, + routines, } from "@paperclipai/db"; import { conflict, HttpError, notFound, unprocessable } from "../errors.js"; +import { routineService } from "./routines.js"; +import type { IssueAssignmentWakeupDeps } from "./issue-assignment-wakeup.js"; const DEFAULT_LEASE_MS = 15 * 60 * 1000; const MAX_LEASE_MS = 24 * 60 * 60 * 1000; @@ -21,7 +28,13 @@ const MAX_FIELDS_BYTES = 64 * 1024; const DEFAULT_STAGES = [ { key: "intake", name: "Intake", kind: "open", position: 100 }, { key: "in_progress", name: "In progress", kind: "working", position: 200 }, - { key: "review", name: "Review", kind: "review", position: 300 }, + { + key: "review", + name: "Review", + kind: "review", + position: 300, + config: { approveToStageKey: "done", rejectToStageKey: "cancelled", requireRejectReason: true, reviewerKind: "human" }, + }, { key: "done", name: "Done", kind: "done", position: 900 }, { key: "cancelled", name: "Cancelled", kind: "cancelled", position: 1000 }, ] as const; @@ -35,6 +48,11 @@ export type PipelineStageKind = "open" | "working" | "review" | "done" | "cancel export type PipelineStageConfig = Record & { autonomy?: "manual" | "suggest" | "auto"; + autoAdvanceOnChildrenTerminal?: string; + approveToStageKey?: string; + rejectToStageKey?: string; + requireRejectReason?: boolean; + reviewerKind?: "human" | "any"; onEnter?: { type?: "run_routine"; routineId?: string; @@ -42,6 +60,11 @@ export type PipelineStageConfig = Record & { }; }; +export type PipelineAutomationExecutionResult = + | { status: "none" } + | { status: "succeeded"; execution: typeof pipelineAutomationExecutions.$inferSelect } + | { status: "failed"; execution: typeof pipelineAutomationExecutions.$inferSelect }; + type PipelineDb = Db | Parameters[0]>[0]; function nowDate() { @@ -120,6 +143,116 @@ function stageConfig(stage: typeof pipelineStages.$inferSelect): PipelineStageCo return (stage.config ?? {}) as PipelineStageConfig; } +function normalizeStageConfig(kind: PipelineStageKind | string, config?: PipelineStageConfig | null): PipelineStageConfig { + const next = { ...(config ?? {}) }; + if (kind !== "review") return next; + + if (typeof next.approveToStageKey !== "string" || next.approveToStageKey.trim().length === 0) { + throw unprocessable("Review stages require approveToStageKey", { code: "validation" }); + } + if (typeof next.rejectToStageKey !== "string" || next.rejectToStageKey.trim().length === 0) { + throw unprocessable("Review stages require rejectToStageKey", { code: "validation" }); + } + if (next.requireRejectReason !== undefined && typeof next.requireRejectReason !== "boolean") { + throw unprocessable("Review stage requireRejectReason must be boolean", { code: "validation" }); + } + if (next.reviewerKind !== undefined && next.reviewerKind !== "human" && next.reviewerKind !== "any") { + throw unprocessable("Review stage reviewerKind must be human or any", { code: "validation" }); + } + + return { + ...next, + approveToStageKey: next.approveToStageKey.trim(), + rejectToStageKey: next.rejectToStageKey.trim(), + requireRejectReason: next.requireRejectReason ?? true, + reviewerKind: next.reviewerKind ?? "human", + }; +} + +function reviewConfigForStage(stage: typeof pipelineStages.$inferSelect) { + return normalizeStageConfig(stage.kind, stageConfig(stage)); +} + +function assertReviewTargetsInSet( + kind: PipelineStageKind | string, + config: PipelineStageConfig, + stageKeys: Set, +) { + if (kind !== "review") return; + if (!stageKeys.has(config.approveToStageKey!)) { + throw unprocessable("Review approveToStageKey references an unknown stage", { code: "validation" }); + } + if (!stageKeys.has(config.rejectToStageKey!)) { + throw unprocessable("Review rejectToStageKey references an unknown stage", { code: "validation" }); + } +} + +function stageAutomation(stage: typeof pipelineStages.$inferSelect) { + const onEnter = stageConfig(stage).onEnter; + if (!onEnter || onEnter.type !== "run_routine" || !onEnter.routineId) return null; + return { + id: onEnter.id ?? `${stage.id}:on_enter`, + routineId: onEnter.routineId, + }; +} + +function buildCaseDeepLink(input: { pipelineId: string; caseId: string }) { + return `/PAP/pipelines/${input.pipelineId}/cases/${input.caseId}`; +} + +function buildPipelineCaseContextPack(input: { + pipeline: typeof pipelines.$inferSelect; + case: typeof pipelineCases.$inferSelect; + stage: typeof pipelineStages.$inferSelect; +}) { + return { + pipeline: { + id: input.pipeline.id, + key: input.pipeline.key, + name: input.pipeline.name, + }, + case: { + id: input.case.id, + caseKey: input.case.caseKey, + title: input.case.title, + version: input.case.version, + deepLink: buildCaseDeepLink({ pipelineId: input.pipeline.id, caseId: input.case.id }), + untrustedContent: { + summary: input.case.summary, + fields: input.case.fields, + }, + }, + stage: { + id: input.stage.id, + key: input.stage.key, + name: input.stage.name, + kind: input.stage.kind, + }, + }; +} + +function buildPipelineCaseContextMarkdown(input: { + pipeline: typeof pipelines.$inferSelect; + case: typeof pipelineCases.$inferSelect; + stage: typeof pipelineStages.$inferSelect; + triggeringEventId?: string | null; +}) { + const contextPack = buildPipelineCaseContextPack(input); + return [ + "## Pipeline Case Context", + "", + `Case: ${input.case.title}`, + `Pipeline: ${input.pipeline.name} (${input.pipeline.key})`, + `Stage: ${input.stage.name} (${input.stage.key}, ${input.stage.kind})`, + `Case link: ${contextPack.case.deepLink}`, + input.triggeringEventId ? `Triggering event: ${input.triggeringEventId}` : null, + "", + "```json", + JSON.stringify(contextPack, null, 2), + "```", + ].filter((line): line is string => line !== null).join("\n"); +} + async function writeCaseEvent( db: PipelineDb, input: { @@ -237,7 +370,7 @@ async function assertLeaseAvailable( } async function assertNoOpenBlockers(db: PipelineDb, row: typeof pipelineCases.$inferSelect, toStage: typeof pipelineStages.$inferSelect) { - if (toStage.kind === "cancelled") return; + if (toStage.kind !== "working" && toStage.kind !== "done") return; const blockers = await db .select({ id: pipelineCases.id, @@ -259,6 +392,215 @@ async function assertNoOpenBlockers(db: PipelineDb, row: typeof pipelineCases.$i } } +async function getCaseOrThrow(db: PipelineDb, companyId: string, caseId: string) { + const row = await db + .select() + .from(pipelineCases) + .where(and(eq(pipelineCases.id, caseId), eq(pipelineCases.companyId, companyId))) + .limit(1) + .then((rows) => rows[0] ?? null); + if (!row) throw notFound("Pipeline case not found"); + return row; +} + +async function assertValidParentCase( + db: PipelineDb, + input: { companyId: string; caseId?: string | null; parentCaseId?: string | null }, +) { + if (!input.parentCaseId) return null; + if (input.caseId && input.parentCaseId === input.caseId) { + throw conflict("Pipeline case parent cycle detected", { code: "parent_cycle" }); + } + + let current = await getCaseOrThrow(db, input.companyId, input.parentCaseId); + let depth = 1; + while (current.parentCaseId) { + if (input.caseId && current.parentCaseId === input.caseId) { + throw conflict("Pipeline case parent cycle detected", { code: "parent_cycle" }); + } + if (depth >= 32) { + throw unprocessable("Pipeline case parent depth exceeds 32", { code: "parent_depth_exceeded" }); + } + current = await getCaseOrThrow(db, input.companyId, current.parentCaseId); + depth += 1; + } + if (depth >= 32) { + throw unprocessable("Pipeline case parent depth exceeds 32", { code: "parent_depth_exceeded" }); + } + return input.parentCaseId; +} + +async function adjustParentCounts( + db: PipelineDb, + input: { parentCaseId: string | null | undefined; childDelta?: number; terminalChildDelta?: number }, +) { + if (!input.parentCaseId) return; + const patch: Partial = { updatedAt: nowDate() }; + if (input.childDelta) { + patch.childCount = sql`${pipelineCases.childCount} + ${input.childDelta}` as unknown as number; + } + if (input.terminalChildDelta) { + patch.terminalChildCount = sql`${pipelineCases.terminalChildCount} + ${input.terminalChildDelta}` as unknown as number; + } + if (!input.childDelta && !input.terminalChildDelta) return; + await db.update(pipelineCases).set(patch).where(eq(pipelineCases.id, input.parentCaseId)); +} + +async function computeCaseRollup(db: PipelineDb, 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 }; +} + +async function hasCaseEvent(db: PipelineDb, caseId: string, type: string) { + const row = await db + .select({ id: pipelineCaseEvents.id }) + .from(pipelineCaseEvents) + .where(and(eq(pipelineCaseEvents.caseId, caseId), eq(pipelineCaseEvents.type, type))) + .limit(1) + .then((rows) => rows[0] ?? null); + return Boolean(row); +} + +async function postSystemCommentOnLinkedIssues( + db: PipelineDb, + input: { + companyId: string; + caseId: string; + roles: Array<"origin" | "conversation" | "work" | "automation">; + body: string; + }, +) { + const rows = await db + .select({ issueId: issues.id }) + .from(pipelineCaseIssueLinks) + .innerJoin(issues, eq(pipelineCaseIssueLinks.issueId, issues.id)) + .where(and( + eq(pipelineCaseIssueLinks.companyId, input.companyId), + eq(pipelineCaseIssueLinks.caseId, input.caseId), + inArray(pipelineCaseIssueLinks.role, input.roles), + ne(issues.status, "done"), + ne(issues.status, "cancelled"), + isNull(issues.hiddenAt), + )); + + for (const row of rows) { + await db.insert(issueComments).values({ + companyId: input.companyId, + issueId: row.issueId, + authorType: "system", + body: input.body, + }); + await db.update(issues).set({ updatedAt: nowDate() }).where(eq(issues.id, row.issueId)); + } +} + +async function getAncestorCases(db: PipelineDb, companyId: string, parentCaseId: string | null | undefined) { + const ancestors: Array<{ + case: typeof pipelineCases.$inferSelect; + stage: typeof pipelineStages.$inferSelect; + }> = []; + let nextId = parentCaseId ?? null; + let depth = 0; + while (nextId) { + if (depth >= 32) break; + const row = await getCaseWithStageOrThrow(db, companyId, nextId); + ancestors.push(row); + nextId = row.case.parentCaseId; + depth += 1; + } + return ancestors; +} + +async function handleBlockersResolved(db: PipelineDb, companyId: string, blockerCaseId: string) { + const blockedRows = await db + .select({ caseId: pipelineCaseBlockers.caseId }) + .from(pipelineCaseBlockers) + .where(and(eq(pipelineCaseBlockers.companyId, companyId), eq(pipelineCaseBlockers.blockedByCaseId, blockerCaseId))); + + for (const blocked of blockedRows) { + const [{ count }] = await db + .select({ count: sql`count(*)::int` }) + .from(pipelineCaseBlockers) + .innerJoin(pipelineCases, eq(pipelineCaseBlockers.blockedByCaseId, pipelineCases.id)) + .where(and( + eq(pipelineCaseBlockers.companyId, companyId), + eq(pipelineCaseBlockers.caseId, blocked.caseId), + or(isNull(pipelineCases.terminalKind), ne(pipelineCases.terminalKind, "done")), + )); + if ((count ?? 0) > 0 || await hasCaseEvent(db, blocked.caseId, "blockers_resolved")) continue; + await writeCaseEvent(db, { + companyId, + caseId: blocked.caseId, + type: "blockers_resolved", + actor: { type: "system" }, + payload: { resolvedByCaseId: blockerCaseId }, + }); + await postSystemCommentOnLinkedIssues(db, { + companyId, + caseId: blocked.caseId, + roles: ["work"], + body: `Pipeline blockers resolved for case ${blocked.caseId}. The case can be retried now that blocker ${blockerCaseId} is done.`, + }); + } +} + +async function validateBlockerSet( + db: PipelineDb, + input: { companyId: string; caseId: string; blockedByCaseIds: string[] }, +) { + const uniqueBlockerIds = [...new Set(input.blockedByCaseIds)]; + if (uniqueBlockerIds.length !== input.blockedByCaseIds.length) { + throw unprocessable("Pipeline blocker set contains duplicate cases", { code: "validation" }); + } + if (uniqueBlockerIds.includes(input.caseId)) { + throw conflict("Pipeline case cannot block itself", { code: "blocker_cycle" }); + } + if (uniqueBlockerIds.length === 0) return uniqueBlockerIds; + + const rows = await db + .select({ id: pipelineCases.id }) + .from(pipelineCases) + .where(and(eq(pipelineCases.companyId, input.companyId), inArray(pipelineCases.id, uniqueBlockerIds))); + if (rows.length !== uniqueBlockerIds.length) throw notFound("Pipeline blocker case not found"); + + const stack = [...uniqueBlockerIds]; + const seen = new Set(); + while (stack.length) { + const current = stack.pop()!; + if (current === input.caseId) { + throw conflict("Pipeline blocker cycle detected", { code: "blocker_cycle" }); + } + if (seen.has(current)) continue; + seen.add(current); + const next = await db + .select({ blockedByCaseId: pipelineCaseBlockers.blockedByCaseId }) + .from(pipelineCaseBlockers) + .where(and(eq(pipelineCaseBlockers.companyId, input.companyId), eq(pipelineCaseBlockers.caseId, current))); + stack.push(...next.map((row) => row.blockedByCaseId)); + } + + return uniqueBlockerIds; +} + async function enqueueStageAutomationLedger( db: PipelineDb, input: { @@ -268,19 +610,18 @@ async function enqueueStageAutomationLedger( eventId: string; }, ) { - const config = stageConfig(input.stage); - const onEnter = config.onEnter; - if (!onEnter || onEnter.type !== "run_routine" || !onEnter.routineId) return null; - const automationId = onEnter.id ?? `${input.stage.id}:on_enter`; + const automation = stageAutomation(input.stage); + if (!automation) return null; const [ledger] = await db .insert(pipelineAutomationExecutions) .values({ companyId: input.companyId, caseId: input.caseId, - automationId, + automationId: automation.id, triggeringEventId: input.eventId, - routineId: onEnter.routineId, - status: "succeeded", + routineId: automation.routineId, + status: "failed", + error: "pending_dispatch", }) .onConflictDoNothing({ target: [ @@ -293,7 +634,222 @@ async function enqueueStageAutomationLedger( return ledger ?? null; } -export function pipelineService(db: Db) { +export function pipelineService(db: Db, deps: { heartbeat?: IssueAssignmentWakeupDeps } = {}) { + const routinesSvc = routineService(db, { heartbeat: deps.heartbeat }); + + async function assertRoutineInCompany(companyId: string, routineId: string) { + const routine = await db + .select({ id: routines.id, companyId: routines.companyId, assigneeAgentId: routines.assigneeAgentId }) + .from(routines) + .where(eq(routines.id, routineId)) + .limit(1) + .then((rows) => rows[0] ?? null); + if (!routine) throw notFound("Routine not found"); + if (routine.companyId !== companyId) { + throw unprocessable("Pipeline automation routine must belong to the same company", { code: "validation" }); + } + return routine; + } + + async function validateStageAutomationConfig(companyId: string, config?: PipelineStageConfig | null) { + const onEnter = config?.onEnter; + if (!onEnter || onEnter.type !== "run_routine" || !onEnter.routineId) return; + await assertRoutineInCompany(companyId, onEnter.routineId); + } + + async function validateStageTargets(companyId: string, pipelineId: string, kind: PipelineStageKind | string, config: PipelineStageConfig) { + if (kind !== "review") return; + const rows = await db + .select({ key: pipelineStages.key }) + .from(pipelineStages) + .innerJoin(pipelines, eq(pipelineStages.pipelineId, pipelines.id)) + .where(and(eq(pipelineStages.pipelineId, pipelineId), eq(pipelines.companyId, companyId))); + assertReviewTargetsInSet(kind, config, new Set(rows.map((row) => row.key))); + } + + async function executeAutomationLedger( + executionId: string, + actor: PipelineActor = { type: "system" }, + ): Promise { + const execution = await db + .select() + .from(pipelineAutomationExecutions) + .where(eq(pipelineAutomationExecutions.id, executionId)) + .limit(1) + .then((rows) => rows[0] ?? null); + if (!execution) throw notFound("Pipeline automation execution not found"); + if (execution.status === "succeeded" && execution.executionIssueId) { + return { status: "succeeded", execution }; + } + + const detail = await getCaseWithStageOrThrow(db, execution.companyId, execution.caseId); + const automation = stageAutomation(detail.stage); + if (!automation || automation.id !== execution.automationId) { + const [failed] = await db + .update(pipelineAutomationExecutions) + .set({ status: "failed", error: "automation_not_configured", updatedAt: nowDate() }) + .where(eq(pipelineAutomationExecutions.id, execution.id)) + .returning(); + await writeCaseEvent(db, { + companyId: execution.companyId, + caseId: execution.caseId, + type: "automation_failed", + actor, + payload: { automationId: execution.automationId, error: "automation_not_configured" }, + }); + return { status: "failed", execution: failed! }; + } + + try { + const routine = await assertRoutineInCompany(execution.companyId, execution.routineId); + const contextPack = buildPipelineCaseContextPack(detail); + const run = await routinesSvc.runPipelineStageEntryRoutine(execution.routineId, { + source: "api", + assigneeAgentId: routine.assigneeAgentId, + idempotencyKey: `pipeline:${execution.caseId}:${execution.automationId}:${execution.triggeringEventId}`, + payload: { + pipeline: contextPack.pipeline, + case: contextPack.case, + stage: contextPack.stage, + triggeringEventId: execution.triggeringEventId, + contextPack, + }, + descriptionAppendix: buildPipelineCaseContextMarkdown({ + ...detail, + triggeringEventId: execution.triggeringEventId, + }), + }); + if (!run.linkedIssueId) { + throw new Error(`Routine run ${run.id} did not create or coalesce an execution issue`); + } + const [updated] = await db + .update(pipelineAutomationExecutions) + .set({ + status: "succeeded", + executionIssueId: run.linkedIssueId, + error: null, + updatedAt: nowDate(), + }) + .where(eq(pipelineAutomationExecutions.id, execution.id)) + .returning(); + await db + .insert(pipelineCaseIssueLinks) + .values({ + companyId: execution.companyId, + caseId: execution.caseId, + issueId: run.linkedIssueId, + role: "automation", + createdByRunId: null, + }) + .onConflictDoNothing({ target: [pipelineCaseIssueLinks.caseId, pipelineCaseIssueLinks.issueId] }); + await writeCaseEvent(db, { + companyId: execution.companyId, + caseId: execution.caseId, + type: "automation_executed", + actor, + payload: { + automationId: execution.automationId, + routineId: execution.routineId, + routineRunId: run.id, + issueId: run.linkedIssueId, + status: run.status, + }, + }); + return { status: "succeeded", execution: updated! }; + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + const [failed] = await db + .update(pipelineAutomationExecutions) + .set({ status: "failed", error: message, updatedAt: nowDate() }) + .where(eq(pipelineAutomationExecutions.id, execution.id)) + .returning(); + await writeCaseEvent(db, { + companyId: execution.companyId, + caseId: execution.caseId, + type: "automation_failed", + actor, + payload: { automationId: execution.automationId, routineId: execution.routineId, error: message }, + }); + return { status: "failed", execution: failed! }; + } + } + + async function patchCaseContentInTransaction( + tx: PipelineDb, + input: { + companyId: string; + caseId: string; + title?: string; + summary?: string | null; + fields?: Record; + parentCaseId?: string | null; + expectedVersion?: number; + leaseToken?: string | null; + actor: PipelineActor; + }, + ) { + if (input.fields !== undefined) assertJsonSize(input.fields, "fields"); + const { case: existing, stage } = await getCaseWithStageOrThrow(tx, input.companyId, input.caseId); + const current = await assertLeaseAvailable(tx, existing, input.actor, input.leaseToken); + if (input.expectedVersion !== undefined && current.version !== input.expectedVersion) { + throw conflict("Pipeline case version conflict", conflictDetailsForCase(current, stage)); + } + if (input.parentCaseId !== undefined) { + await assertValidParentCase(tx, { + companyId: input.companyId, + caseId: current.id, + parentCaseId: input.parentCaseId, + }); + } + const patch: Partial = { + version: current.version + 1, + updatedAt: nowDate(), + }; + if (input.title !== undefined) patch.title = input.title; + if (input.summary !== undefined) patch.summary = input.summary; + if (input.fields !== undefined) patch.fields = input.fields; + if (input.parentCaseId !== undefined) patch.parentCaseId = input.parentCaseId; + + const [updated] = await tx + .update(pipelineCases) + .set(patch) + .where(and(eq(pipelineCases.id, current.id), eq(pipelineCases.version, current.version))) + .returning(); + if (!updated) { + const latest = await getCaseWithStageOrThrow(tx, input.companyId, input.caseId); + throw conflict("Pipeline case version conflict", conflictDetailsForCase(latest.case, latest.stage)); + } + + const event = await writeCaseEvent(tx, { + companyId: input.companyId, + caseId: updated.id, + type: "updated", + actor: input.actor, + payload: { + previousVersion: current.version, + version: updated.version, + parentCaseChanged: input.parentCaseId !== undefined && input.parentCaseId !== current.parentCaseId, + }, + }); + if (input.parentCaseId !== undefined && input.parentCaseId !== current.parentCaseId) { + const terminalDelta = isTerminalKind(current.terminalKind) ? 1 : 0; + await adjustParentCounts(tx, { + parentCaseId: current.parentCaseId, + childDelta: -1, + terminalChildDelta: -terminalDelta, + }); + await adjustParentCounts(tx, { + parentCaseId: input.parentCaseId, + childDelta: 1, + terminalChildDelta: terminalDelta, + }); + if (isTerminalKind(current.terminalKind)) { + await handleChildrenTerminal(tx, input.companyId, input.parentCaseId); + } + } + return { case: updated, event }; + } + async function transitionCaseInTransaction( tx: PipelineDb, input: { @@ -395,9 +951,61 @@ export function pipelineService(db: Db) { stage: toStage, eventId: event.id, }); + const wasTerminal = isTerminalKind(current.terminalKind); + const isTerminal = isTerminalKind(updated.terminalKind); + if (current.parentCaseId && wasTerminal !== isTerminal) { + await adjustParentCounts(tx, { + parentCaseId: current.parentCaseId, + terminalChildDelta: isTerminal ? 1 : -1, + }); + } + if (!wasTerminal && updated.terminalKind === "done") { + await handleBlockersResolved(tx, input.companyId, current.id); + } + if (!wasTerminal && isTerminal) { + await handleChildrenTerminal(tx, input.companyId, current.parentCaseId); + } return { case: updated, event, automationLedger: ledger }; } + async function handleChildrenTerminal(tx: PipelineDb, companyId: string, parentCaseId: string | null | undefined) { + const ancestors = await getAncestorCases(tx, companyId, parentCaseId); + for (const ancestor of ancestors) { + const rollup = await computeCaseRollup(tx, companyId, ancestor.case.id); + if (!rollup.complete || rollup.total === 0 || await hasCaseEvent(tx, ancestor.case.id, "children_terminal")) { + continue; + } + await writeCaseEvent(tx, { + companyId, + caseId: ancestor.case.id, + type: "children_terminal", + actor: { type: "system" }, + payload: { rollup }, + }); + await postSystemCommentOnLinkedIssues(tx, { + companyId, + caseId: ancestor.case.id, + roles: ["origin", "conversation"], + body: `All child cases for pipeline case "${ancestor.case.title}" are terminal. Rollup: ${rollup.done} done, ${rollup.cancelled} cancelled, ${rollup.open} open.`, + }); + + const toStageKey = stageConfig(ancestor.stage).autoAdvanceOnChildrenTerminal; + if (typeof toStageKey !== "string" || !toStageKey || isTerminalKind(ancestor.case.terminalKind)) { + continue; + } + const toStage = await getStageByKeyOrThrow(tx, ancestor.case.pipelineId, toStageKey); + if (toStage.id === ancestor.stage.id) continue; + await transitionCaseInTransaction(tx, { + companyId, + caseId: ancestor.case.id, + toStageKey, + expectedVersion: ancestor.case.version, + actor: { type: "system" }, + reason: "children_terminal", + }); + } + } + const service = { async createPipeline(input: { companyId: string; @@ -410,6 +1018,21 @@ export function pipelineService(db: Db) { actor: PipelineActor; }) { return db.transaction(async (tx) => { + const stageInputs = input.stages?.length + ? input.stages.map((stage, index) => ({ + ...stage, + position: stage.position ?? (index + 1) * 100, + config: normalizeStageConfig(stage.kind, stage.config), + })) + : DEFAULT_STAGES.map((stage) => ({ + ...stage, + config: normalizeStageConfig(stage.kind, "config" in stage ? stage.config : {}), + })); + const stageKeys = new Set(stageInputs.map((stage) => stage.key)); + for (const stage of stageInputs) { + assertReviewTargetsInSet(stage.kind, stage.config, stageKeys); + await validateStageAutomationConfig(input.companyId, stage.config); + } const [pipeline] = await tx .insert(pipelines) .values({ @@ -423,9 +1046,6 @@ export function pipelineService(db: Db) { createdByAgentId: input.actor.type === "agent" ? input.actor.agentId : null, }) .returning(); - const stageInputs = input.stages?.length - ? input.stages.map((stage, index) => ({ ...stage, position: stage.position ?? (index + 1) * 100 })) - : DEFAULT_STAGES.map((stage) => ({ ...stage, config: {} })); const insertedStages = await tx .insert(pipelineStages) .values(stageInputs.map((stage) => ({ @@ -479,6 +1099,9 @@ export function pipelineService(db: Db) { config?: PipelineStageConfig; }) { await getPipelineOrThrow(db, input.companyId, input.pipelineId); + const config = normalizeStageConfig(input.kind, input.config); + await validateStageTargets(input.companyId, input.pipelineId, input.kind, config); + await validateStageAutomationConfig(input.companyId, config); const [stage] = await db .insert(pipelineStages) .values({ @@ -487,12 +1110,44 @@ export function pipelineService(db: Db) { name: input.name, kind: input.kind, position: input.position, - config: input.config ?? {}, + config, }) .returning(); return stage!; }, + async updateStage(input: { + companyId: string; + pipelineId: string; + stageId: string; + patch: { + key?: string; + name?: string; + kind?: PipelineStageKind; + position?: number; + config?: PipelineStageConfig; + }; + }) { + await getPipelineOrThrow(db, input.companyId, input.pipelineId); + const existing = await getStageOrThrow(db, input.pipelineId, input.stageId); + const kind = input.patch.kind ?? existing.kind; + const config = normalizeStageConfig(kind, input.patch.config !== undefined ? input.patch.config : stageConfig(existing)); + await validateStageTargets(input.companyId, input.pipelineId, kind, config); + await validateStageAutomationConfig(input.companyId, config); + const [updated] = await db + .update(pipelineStages) + .set({ + ...input.patch, + kind, + config, + updatedAt: nowDate(), + }) + .where(and(eq(pipelineStages.id, input.stageId), eq(pipelineStages.pipelineId, input.pipelineId))) + .returning(); + if (!updated) throw notFound("Pipeline stage not found"); + return updated; + }, + async deleteStage(input: { companyId: string; pipelineId: string; @@ -573,6 +1228,7 @@ export function pipelineService(db: Db) { fields?: Record; stageKey?: string | null; parentCaseId?: string | null; + blockedByCaseIds?: string[]; actor: PipelineActor; }) { assertJsonSize(input.fields ?? {}, "fields"); @@ -583,6 +1239,12 @@ export function pipelineService(db: Db) { return db.transaction(async (tx) => { const pipeline = await getPipelineOrThrow(tx, input.companyId, input.pipelineId); if (pipeline.archivedAt) throw unprocessable("Pipeline is archived", { code: "pipeline_archived" }); + await assertValidParentCase(tx, { companyId: input.companyId, parentCaseId: input.parentCaseId ?? null }); + const blockedByCaseIds = await validateBlockerSet(tx, { + companyId: input.companyId, + caseId: "__new_case__", + blockedByCaseIds: input.blockedByCaseIds ?? [], + }); const stage = input.stageKey ? await getStageByKeyOrThrow(tx, input.pipelineId, input.stageKey) : await tx @@ -633,6 +1295,25 @@ export function pipelineService(db: Db) { toStageId: stage.id, payload: { caseKey }, }); + await adjustParentCounts(tx, { + parentCaseId: inserted.parentCaseId, + childDelta: 1, + terminalChildDelta: isTerminalKind(inserted.terminalKind) ? 1 : 0, + }); + if (blockedByCaseIds.length > 0) { + await tx.insert(pipelineCaseBlockers).values(blockedByCaseIds.map((blockedByCaseId) => ({ + companyId: input.companyId, + caseId: inserted.id, + blockedByCaseId, + }))); + await writeCaseEvent(tx, { + companyId: input.companyId, + caseId: inserted.id, + type: "blockers_set", + actor: input.actor, + payload: { blockedByCaseIds }, + }); + } return { case: inserted, created: true }; }); }, @@ -647,6 +1328,7 @@ export function pipelineService(db: Db) { fields?: Record; stageKey?: string | null; parentCaseId?: string | null; + blockedByCaseIds?: string[]; }>; actor: PipelineActor; }) { @@ -687,43 +1369,14 @@ export function pipelineService(db: Db) { title?: string; summary?: string | null; fields?: Record; + parentCaseId?: string | null; expectedVersion?: number; leaseToken?: string | null; actor: PipelineActor; }) { - if (input.fields !== undefined) assertJsonSize(input.fields, "fields"); return db.transaction(async (tx) => { - const { case: existing, stage } = await getCaseWithStageOrThrow(tx, input.companyId, input.caseId); - const current = await assertLeaseAvailable(tx, existing, input.actor, input.leaseToken); - if (input.expectedVersion !== undefined && current.version !== input.expectedVersion) { - throw conflict("Pipeline case version conflict", conflictDetailsForCase(current, stage)); - } - const patch: Partial = { - version: current.version + 1, - updatedAt: nowDate(), - }; - if (input.title !== undefined) patch.title = input.title; - if (input.summary !== undefined) patch.summary = input.summary; - if (input.fields !== undefined) patch.fields = input.fields; - - const [updated] = await tx - .update(pipelineCases) - .set(patch) - .where(and(eq(pipelineCases.id, current.id), eq(pipelineCases.version, current.version))) - .returning(); - if (!updated) { - const latest = await getCaseWithStageOrThrow(tx, input.companyId, input.caseId); - throw conflict("Pipeline case version conflict", conflictDetailsForCase(latest.case, latest.stage)); - } - - await writeCaseEvent(tx, { - companyId: input.companyId, - caseId: updated.id, - type: "updated", - actor: input.actor, - payload: { previousVersion: current.version, version: updated.version }, - }); - return updated; + const result = await patchCaseContentInTransaction(tx, input); + return result.case; }); }, @@ -813,7 +1466,39 @@ export function pipelineService(db: Db) { suggestionId?: string; reason?: string | null; }) { - return db.transaction((tx) => transitionCaseInTransaction(tx, input)); + const result = await db.transaction((tx) => transitionCaseInTransaction(tx, input)); + if (result.automationLedger) { + return { + ...result, + automationExecution: await executeAutomationLedger(result.automationLedger.id, { type: "system" }), + }; + } + return { ...result, automationExecution: { status: "none" } satisfies PipelineAutomationExecutionResult }; + }, + + async retryAutomation(input: { + companyId: string; + caseId: string; + automationId: string; + actor: PipelineActor; + }) { + const execution = await db + .select() + .from(pipelineAutomationExecutions) + .where(and( + eq(pipelineAutomationExecutions.companyId, input.companyId), + eq(pipelineAutomationExecutions.caseId, input.caseId), + eq(pipelineAutomationExecutions.automationId, input.automationId), + )) + .orderBy(sql`case when ${pipelineAutomationExecutions.status} = 'failed' then 0 else 1 end`, asc(pipelineAutomationExecutions.createdAt)) + .limit(1) + .then((rows) => rows[0] ?? null); + if (!execution) throw notFound("Pipeline automation execution not found"); + return executeAutomationLedger(execution.id, input.actor); + }, + + async validateStageAutomationConfig(companyId: string, config?: PipelineStageConfig | null) { + return validateStageAutomationConfig(companyId, config); }, async suggestTransition(input: { @@ -863,7 +1548,7 @@ export function pipelineService(db: Db) { reason?: string | null; leaseToken?: string | null; }) { - return db.transaction(async (tx) => { + const result = await db.transaction(async (tx) => { const { case: existing } = await getCaseWithStageOrThrow(tx, input.companyId, input.caseId); const suggestion = existing.pendingSuggestion; if (!suggestion || suggestion.id !== input.suggestionId) { @@ -905,6 +1590,165 @@ export function pipelineService(db: Db) { }); return transition; }); + if ("automationLedger" in result && result.automationLedger) { + return { + ...result, + automationExecution: await executeAutomationLedger(result.automationLedger.id, { type: "system" }), + }; + } + return result; + }, + + async reviewCase(input: { + companyId: string; + caseId: string; + decision: "approve" | "reject"; + reason?: string | null; + edits?: { + title?: string; + summary?: string | null; + fields?: Record; + parentCaseId?: string | null; + }; + expectedVersion: number; + leaseToken?: string | null; + actor: PipelineActor; + }) { + const result = await db.transaction(async (tx) => { + const detail = await getCaseWithStageOrThrow(tx, input.companyId, input.caseId); + if (detail.stage.kind !== "review") { + throw unprocessable("Pipeline case is not in a review stage", { code: "validation" }); + } + const config = reviewConfigForStage(detail.stage); + if (input.actor.type === "agent" && config.reviewerKind !== "any") { + throw new HttpError(403, "Human review is required", { code: "review_required" }); + } + if (input.decision === "reject" && config.requireRejectReason !== false && !input.reason?.trim()) { + throw unprocessable("Reject reason is required", { code: "validation" }); + } + const toStageKey = input.decision === "approve" ? config.approveToStageKey! : config.rejectToStageKey!; + const suggestionId = detail.case.pendingSuggestion?.id ?? null; + let expectedVersion = input.expectedVersion; + let updateEvent: typeof pipelineCaseEvents.$inferSelect | null = null; + const hasEdits = input.edits && Object.keys(input.edits).length > 0; + + if (hasEdits) { + const updated = await patchCaseContentInTransaction(tx, { + companyId: input.companyId, + caseId: input.caseId, + ...input.edits, + expectedVersion, + leaseToken: input.leaseToken, + actor: input.actor, + }); + expectedVersion = updated.case.version; + updateEvent = updated.event; + } + + const transitioned = await transitionCaseInTransaction(tx, { + companyId: input.companyId, + caseId: input.caseId, + toStageKey, + expectedVersion, + leaseToken: input.leaseToken, + reason: input.reason, + actor: input.actor, + }); + const reviewEvent = await writeCaseEvent(tx, { + companyId: input.companyId, + caseId: input.caseId, + type: "review_decided", + actor: input.actor, + fromStageId: detail.stage.id, + toStageId: transitioned.case.stageId, + payload: { + decision: input.decision, + reason: input.reason ?? null, + suggestionId, + updateEventId: updateEvent?.id ?? null, + transitionEventId: transitioned.event.id, + }, + }); + return { ...transitioned, updateEvent, reviewEvent }; + }); + if (result.automationLedger) { + return { + ...result, + automationExecution: await executeAutomationLedger(result.automationLedger.id, { type: "system" }), + }; + } + return { ...result, automationExecution: { status: "none" } satisfies PipelineAutomationExecutionResult }; + }, + + async listReviewCases(input: { + companyId: string; + pipelineId?: string; + parentCaseId?: string; + }) { + const parentCase = alias(pipelineCases, "parent_pipeline_case"); + const rows = await db + .select({ case: pipelineCases, pipeline: pipelines, stage: pipelineStages, parentCase }) + .from(pipelineCases) + .innerJoin(pipelines, eq(pipelineCases.pipelineId, pipelines.id)) + .innerJoin(pipelineStages, eq(pipelineCases.stageId, pipelineStages.id)) + .leftJoin(parentCase, and(eq(pipelineCases.parentCaseId, parentCase.id), eq(parentCase.companyId, input.companyId))) + .where(and( + eq(pipelineCases.companyId, input.companyId), + eq(pipelines.companyId, input.companyId), + eq(pipelineStages.kind, "review"), + isNull(pipelineCases.terminalKind), + input.pipelineId ? eq(pipelineCases.pipelineId, input.pipelineId) : undefined, + input.parentCaseId ? eq(pipelineCases.parentCaseId, input.parentCaseId) : undefined, + )) + .orderBy(asc(pipelineCases.createdAt)); + return rows.map((row) => ({ + ...row, + pendingSuggestion: row.case.pendingSuggestion, + reviewConfig: reviewConfigForStage(row.stage), + })); + }, + + async replaceBlockers(input: { + companyId: string; + caseId: string; + blockedByCaseIds: string[]; + actor: PipelineActor; + }) { + return db.transaction(async (tx) => { + await getCaseWithStageOrThrow(tx, input.companyId, input.caseId); + const blockedByCaseIds = await validateBlockerSet(tx, { + companyId: input.companyId, + caseId: input.caseId, + blockedByCaseIds: input.blockedByCaseIds, + }); + await tx.delete(pipelineCaseBlockers).where(and( + eq(pipelineCaseBlockers.companyId, input.companyId), + eq(pipelineCaseBlockers.caseId, input.caseId), + )); + if (blockedByCaseIds.length > 0) { + await tx.insert(pipelineCaseBlockers).values(blockedByCaseIds.map((blockedByCaseId) => ({ + companyId: input.companyId, + caseId: input.caseId, + blockedByCaseId, + }))); + } + const event = await writeCaseEvent(tx, { + companyId: input.companyId, + caseId: input.caseId, + type: "blockers_set", + actor: input.actor, + payload: { blockedByCaseIds }, + }); + const blockers = await tx + .select() + .from(pipelineCaseBlockers) + .where(and(eq(pipelineCaseBlockers.companyId, input.companyId), eq(pipelineCaseBlockers.caseId, input.caseId))); + return { blockers, event }; + }); + }, + + async getCaseRollup(companyId: string, caseId: string) { + return computeCaseRollup(db, companyId, caseId); }, async listCaseEvents(companyId: string, caseId: string) { diff --git a/server/src/services/routines.ts b/server/src/services/routines.ts index 7081377c5e2..764043480f4 100644 --- a/server/src/services/routines.ts +++ b/server/src/services/routines.ts @@ -1151,6 +1151,7 @@ export function routineService( executionWorkspaceId?: string | null; executionWorkspacePreference?: string | null; executionWorkspaceSettings?: Record | null; + descriptionAppendix?: string | null; actor?: Actor; }) { const projectId = input.projectId ?? input.routine.projectId ?? null; @@ -1185,7 +1186,10 @@ export function routineService( }); const allVariables = { ...getBuiltinRoutineVariableValues(), ...automaticVariables, ...resolvedVariables }; const title = interpolateRoutineTemplate(input.routine.title, allVariables) ?? input.routine.title; - const description = interpolateRoutineTemplate(input.routine.description, allVariables); + const baseDescription = interpolateRoutineTemplate(input.routine.description, allVariables); + const description = [baseDescription, input.descriptionAppendix] + .filter((part): part is string => Boolean(part && part.trim())) + .join("\n\n"); const triggerPayload = mergeRoutineRunPayload(input.payload, { ...automaticVariables, ...resolvedVariables }); const managedRoutineBinding = await getManagedRoutineBinding(input.routine); const managedIssueTemplate = readManagedRoutineIssueTemplate(managedRoutineBinding?.defaultsJson); @@ -2200,6 +2204,31 @@ export function routineService( }); }, + runPipelineStageEntryRoutine: async (id: string, input: RunRoutine & { descriptionAppendix?: string | null }, actor?: Actor) => { + const routine = await getRoutineById(id); + if (!routine) throw notFound("Routine not found"); + if (routine.status === "archived") throw conflict("Routine is archived"); + await assertProject(routine.companyId, input.projectId ?? null); + const assigneeAgentId = input.assigneeAgentId ?? routine.assigneeAgentId ?? null; + await assertAssignableAgent(db, routine.companyId, assigneeAgentId, { kind: "routine" }); + return dispatchRoutineRun({ + routine, + trigger: null, + source: "api", + payload: input.payload as Record | null | undefined, + variables: input.variables as Record | null | undefined, + projectId: input.projectId ?? null, + assigneeAgentId: input.assigneeAgentId ?? null, + idempotencyKey: input.idempotencyKey, + executionWorkspaceId: input.executionWorkspaceId ?? null, + executionWorkspacePreference: input.executionWorkspacePreference ?? null, + executionWorkspaceSettings: + (input.executionWorkspaceSettings as Record | null | undefined) ?? null, + descriptionAppendix: input.descriptionAppendix ?? null, + actor, + }); + }, + firePublicTrigger: async (publicId: string, input: { authorizationHeader?: string | null; signatureHeader?: string | null;