Implement pipeline review flows

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-06-09 23:13:20 +00:00
1 parent e80ffc86cd
commit d16f1caf41
6 files changed
+1695 -206

No files matched your search

+170 -10
View File
@@ -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<typeof createDb>;
let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | 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));
+421 -20
View File
@@ -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<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | 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<number>`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);
});
});
+45
View File
@@ -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<ReturnType<typeof parseIssueExecutionState>>;
type NormalizedExecutionPolicy = NonNullable<ReturnType<typeof normalizeIssueExecutionPolicy>>;
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,
});
});
+136 -126
View File
@@ -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<typeof pipelineService>[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<string, unknown>
: 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<typeof pipelineService>,
companyId: string,
caseId: string,
body: z.infer<typeof reviewCaseSchema>,
actor: PipelineActor,
) {
const detail = await getCaseDetail(db, companyId, caseId);
if (detail.stage.kind !== "review") {
throw unprocessable("Pipeline case is not in a review stage", { code: "validation" });
}
const config = (detail.stage.config ?? {}) as {
approveToStageKey?: string;
rejectToStageKey?: string;
requireRejectReason?: boolean;
};
if (body.decision === "reject" && config.requireRejectReason !== false && !body.reason?.trim()) {
throw unprocessable("Reject reason is required", { code: "validation" });
}
const toStageKey = body.decision === "approve" ? config.approveToStageKey : config.rejectToStageKey;
if (!toStageKey) throw unprocessable("Review stage is missing decision target config", { code: "validation" });
let expectedVersion = body.expectedVersion;
if (body.edits && Object.keys(body.edits).length > 0) {
const updated = await svc.patchCaseContent({ companyId, caseId, ...body.edits, expectedVersion, actor });
expectedVersion = updated.version;
}
const transitioned = await svc.transitionCase({
companyId,
caseId,
toStageKey,
expectedVersion,
leaseToken: body.leaseToken,
reason: body.reason,
actor,
});
const event = await writeRouteEvent(db, {
companyId,
caseId,
type: "review_decided",
actor,
payload: { decision: body.decision, reason: body.reason ?? null },
});
return { ...transitioned, reviewEvent: event };
}
async function computeRollup(db: Db, companyId: string, caseId: string) {
const rows = await db.execute(sql<{
id: string;
terminal_kind: string | null;
}>`
with recursive subtree as (
select id, terminal_kind from pipeline_cases where company_id = ${companyId} and id = ${caseId}
union all
select child.id, child.terminal_kind
from pipeline_cases child
join subtree parent on child.parent_case_id = parent.id
where child.company_id = ${companyId}
)
select id, terminal_kind from subtree
`);
const items = Array.from(rows);
if (items.length === 0) throw notFound("Pipeline case not found");
const descendants = items.slice(1);
const done = descendants.filter((item) => item.terminal_kind === "done").length;
const cancelled = descendants.filter((item) => item.terminal_kind === "cancelled").length;
const open = descendants.filter((item) => item.terminal_kind !== "done" && item.terminal_kind !== "cancelled").length;
return { total: descendants.length, done, cancelled, open, complete: open === 0 };
function buildCaseContextMarkdown(detail: Awaited<ReturnType<typeof getCaseDetail>>) {
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");
}
File diff suppressed because it is too large. Load diff
+30 -1
View File
@@ -1151,6 +1151,7 @@ export function routineService(
executionWorkspaceId?: string | null;
executionWorkspacePreference?: string | null;
executionWorkspaceSettings?: Record<string, unknown> | 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<string, unknown> | null | undefined,
variables: input.variables as Record<string, unknown> | 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<string, unknown> | null | undefined) ?? null,
descriptionAppendix: input.descriptionAppendix ?? null,
actor,
});
},
firePublicTrigger: async (publicId: string, input: {
authorizationHeader?: string | null;
signatureHeader?: string | null;