From a236029aeb4c0dec6bb48de0596bd3b117816c88 Mon Sep 17 00:00:00 2001 From: Dotta Date: Wed, 10 Jun 2026 20:39:41 +0000 Subject: [PATCH] Implement pipeline learnings UI --- .../migrations/0099_pipeline_foundation.sql | 141 ++ packages/db/src/migrations/meta/_journal.json | 7 + packages/db/src/schema/index.ts | 8 + packages/db/src/schema/pipelines.ts | 165 +++ packages/shared/src/index.ts | 51 + packages/shared/src/validators/index.ts | 53 + .../shared/src/validators/pipeline.test.ts | 29 + packages/shared/src/validators/pipeline.ts | 235 +++- server/src/app.ts | 2 + server/src/routes/index.ts | 1 + server/src/routes/pipelines.ts | 886 ++++++++++++ server/src/routes/routines.ts | 16 +- server/src/services/index.ts | 1 + server/src/services/pipelines.test.ts | 127 ++ server/src/services/pipelines.ts | 1237 +++++++++++++++++ ui/src/App.tsx | 14 + ui/src/api/pipelines.ts | 1 + ui/src/components/Sidebar.tsx | 6 + ui/src/lib/pipeline-item-detail.ts | 211 +++ ui/src/lib/pipeline-learnings.test.ts | 66 + ui/src/lib/pipeline-learnings.ts | 88 ++ ui/src/lib/queryKeys.ts | 13 + ui/src/pages/Pipelines.test.tsx | 355 +++++ ui/src/pages/Pipelines.tsx | 700 +++++++++- ui/vite.config.ts | 23 +- 25 files changed, 4422 insertions(+), 14 deletions(-) create mode 100644 packages/db/src/migrations/0099_pipeline_foundation.sql create mode 100644 packages/db/src/schema/pipelines.ts create mode 100644 server/src/routes/pipelines.ts create mode 100644 server/src/services/pipelines.test.ts create mode 100644 server/src/services/pipelines.ts create mode 100644 ui/src/lib/pipeline-item-detail.ts create mode 100644 ui/src/lib/pipeline-learnings.test.ts create mode 100644 ui/src/lib/pipeline-learnings.ts create mode 100644 ui/src/pages/Pipelines.test.tsx diff --git a/packages/db/src/migrations/0099_pipeline_foundation.sql b/packages/db/src/migrations/0099_pipeline_foundation.sql new file mode 100644 index 00000000000..a99693916af --- /dev/null +++ b/packages/db/src/migrations/0099_pipeline_foundation.sql @@ -0,0 +1,141 @@ +CREATE TABLE IF NOT EXISTS "pipelines" ( + "id" uuid PRIMARY KEY DEFAULT gen_random_uuid(), + "company_id" uuid NOT NULL REFERENCES "companies"("id") ON DELETE CASCADE, + "name" text NOT NULL, + "description" text, + "status" text NOT NULL DEFAULT 'active', + "created_at" timestamptz NOT NULL DEFAULT now(), + "updated_at" timestamptz NOT NULL DEFAULT now() +);--> statement-breakpoint + +CREATE INDEX IF NOT EXISTS "pipelines_company_name_idx" + ON "pipelines" ("company_id", "name");--> statement-breakpoint + +CREATE TABLE IF NOT EXISTS "pipeline_stages" ( + "id" uuid PRIMARY KEY DEFAULT gen_random_uuid(), + "pipeline_id" uuid NOT NULL REFERENCES "pipelines"("id") ON DELETE CASCADE, + "name" text NOT NULL, + "kind" text NOT NULL DEFAULT 'open', + "position" integer NOT NULL DEFAULT 0, + "config" jsonb NOT NULL DEFAULT '{"variables":[]}'::jsonb, + "created_at" timestamptz NOT NULL DEFAULT now(), + "updated_at" timestamptz NOT NULL DEFAULT now() +);--> statement-breakpoint + +CREATE TABLE IF NOT EXISTS "pipeline_stage_transitions" ( + "id" uuid PRIMARY KEY DEFAULT gen_random_uuid(), + "pipeline_id" uuid NOT NULL REFERENCES "pipelines"("id") ON DELETE CASCADE, + "from_stage_id" uuid NOT NULL REFERENCES "pipeline_stages"("id") ON DELETE CASCADE, + "to_stage_id" uuid NOT NULL REFERENCES "pipeline_stages"("id") ON DELETE CASCADE, + "config" jsonb NOT NULL DEFAULT '{}'::jsonb, + "created_at" timestamptz NOT NULL DEFAULT now(), + "updated_at" timestamptz NOT NULL DEFAULT now(), + CONSTRAINT "pipeline_stage_transitions_pipeline_from_to_uq" + UNIQUE ("pipeline_id","from_stage_id","to_stage_id") +);--> statement-breakpoint + +CREATE INDEX IF NOT EXISTS "pipeline_stages_pipeline_idx" + ON "pipeline_stages" ("pipeline_id","position");--> statement-breakpoint + +CREATE INDEX IF NOT EXISTS "pipeline_stage_transitions_pipeline_idx" + ON "pipeline_stage_transitions" ("pipeline_id");--> statement-breakpoint + +CREATE INDEX IF NOT EXISTS "pipeline_stage_transitions_from_idx" + ON "pipeline_stage_transitions" ("from_stage_id");--> statement-breakpoint + +CREATE INDEX IF NOT EXISTS "pipeline_stage_transitions_to_idx" + ON "pipeline_stage_transitions" ("to_stage_id");--> statement-breakpoint + +CREATE TABLE IF NOT EXISTS "pipeline_cases" ( + "id" uuid PRIMARY KEY DEFAULT gen_random_uuid(), + "company_id" uuid NOT NULL REFERENCES "companies"("id") ON DELETE CASCADE, + "pipeline_id" uuid NOT NULL REFERENCES "pipelines"("id") ON DELETE CASCADE, + "stage_id" uuid REFERENCES "pipeline_stages"("id") ON DELETE SET NULL, + "parent_case_id" uuid REFERENCES "pipeline_cases"("id") ON DELETE SET NULL, + "title" text NOT NULL, + "status" text NOT NULL DEFAULT 'open', + "fields" jsonb NOT NULL DEFAULT '{}'::jsonb, + "created_by_actor_id" text, + "created_by_actor_type" text, + "last_activity_at" timestamptz, + "created_at" timestamptz NOT NULL DEFAULT now(), + "updated_at" timestamptz NOT NULL DEFAULT now() +);--> statement-breakpoint + +CREATE INDEX IF NOT EXISTS "pipeline_cases_pipeline_idx" + ON "pipeline_cases" ("pipeline_id","stage_id");--> statement-breakpoint + +CREATE INDEX IF NOT EXISTS "pipeline_cases_stage_idx" + ON "pipeline_cases" ("stage_id");--> statement-breakpoint + +CREATE INDEX IF NOT EXISTS "pipeline_cases_company_idx" + ON "pipeline_cases" ("company_id");--> statement-breakpoint + +CREATE INDEX IF NOT EXISTS "pipeline_cases_stage_status_idx" + ON "pipeline_cases" ("pipeline_id","status");--> statement-breakpoint + +CREATE INDEX IF NOT EXISTS "pipeline_cases_parent_idx" + ON "pipeline_cases" ("parent_case_id");--> statement-breakpoint + +CREATE TABLE IF NOT EXISTS "pipeline_case_events" ( + "id" uuid PRIMARY KEY DEFAULT gen_random_uuid(), + "company_id" uuid NOT NULL REFERENCES "companies"("id") ON DELETE CASCADE, + "pipeline_id" uuid NOT NULL REFERENCES "pipelines"("id") ON DELETE CASCADE, + "case_id" uuid NOT NULL REFERENCES "pipeline_cases"("id") ON DELETE CASCADE, + "kind" text NOT NULL, + "summary" text, + "payload" jsonb, + "created_at" timestamptz NOT NULL DEFAULT now() +);--> statement-breakpoint + +CREATE INDEX IF NOT EXISTS "pipeline_case_events_pipeline_idx" + ON "pipeline_case_events" ("pipeline_id","case_id");--> statement-breakpoint + +CREATE INDEX IF NOT EXISTS "pipeline_case_events_case_idx" + ON "pipeline_case_events" ("case_id");--> statement-breakpoint + +CREATE INDEX IF NOT EXISTS "pipeline_case_events_kind_idx" + ON "pipeline_case_events" ("kind");--> statement-breakpoint + +CREATE TABLE IF NOT EXISTS "pipeline_guidance_documents" ( + "id" uuid PRIMARY KEY DEFAULT gen_random_uuid(), + "pipeline_id" uuid NOT NULL REFERENCES "pipelines"("id") ON DELETE CASCADE, + "key" text NOT NULL, + "title" text NOT NULL, + "body" text NOT NULL, + "created_at" timestamptz NOT NULL DEFAULT now(), + "updated_at" timestamptz NOT NULL DEFAULT now(), + CONSTRAINT "pipeline_guidance_documents_pipeline_key_uq" UNIQUE ("pipeline_id", "key") +);--> statement-breakpoint + +CREATE TABLE IF NOT EXISTS "pipeline_case_issue_links" ( + "id" uuid PRIMARY KEY DEFAULT gen_random_uuid(), + "case_id" uuid NOT NULL REFERENCES "pipeline_cases"("id") ON DELETE CASCADE, + "issue_id" uuid NOT NULL REFERENCES "issues"("id") ON DELETE CASCADE, + "role" text NOT NULL DEFAULT 'automation', + "created_at" timestamptz NOT NULL DEFAULT now(), + CONSTRAINT "pipeline_case_issue_links_case_issue_uq" UNIQUE ("case_id","issue_id") +);--> statement-breakpoint + +CREATE INDEX IF NOT EXISTS "pipeline_case_issue_links_case_idx" + ON "pipeline_case_issue_links" ("case_id","issue_id");--> statement-breakpoint + +CREATE INDEX IF NOT EXISTS "pipeline_case_issue_links_issue_idx" + ON "pipeline_case_issue_links" ("issue_id","case_id");--> statement-breakpoint + +CREATE TABLE IF NOT EXISTS "pipeline_case_blockers" ( + "id" uuid PRIMARY KEY DEFAULT gen_random_uuid(), + "case_id" uuid NOT NULL REFERENCES "pipeline_cases"("id") ON DELETE CASCADE, + "blocked_by_case_id" uuid NOT NULL REFERENCES "pipeline_cases"("id") ON DELETE CASCADE, + "reason" text, + "resolved" boolean NOT NULL DEFAULT false, + "created_at" timestamptz NOT NULL DEFAULT now(), + "resolved_at" timestamptz, + CONSTRAINT "pipeline_case_blockers_pair_uq" UNIQUE ("case_id","blocked_by_case_id") +);--> statement-breakpoint + +CREATE INDEX IF NOT EXISTS "pipeline_case_blockers_case_idx" + ON "pipeline_case_blockers" ("case_id");--> statement-breakpoint + +CREATE INDEX IF NOT EXISTS "pipeline_case_blockers_blocked_by_idx" + ON "pipeline_case_blockers" ("blocked_by_case_id"); diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index 96c9209ead3..4e7768a924b 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -694,6 +694,13 @@ "when": 1780534200000, "tag": "0098_project_icon", "breakpoints": true + }, + { + "idx": 99, + "version": "7", + "when": 1781123714191, + "tag": "0099_pipeline_foundation", + "breakpoints": true } ] } diff --git a/packages/db/src/schema/index.ts b/packages/db/src/schema/index.ts index 9d78a33cbd2..4b801a27591 100644 --- a/packages/db/src/schema/index.ts +++ b/packages/db/src/schema/index.ts @@ -24,6 +24,14 @@ export { agentWakeupRequests } from "./agent_wakeup_requests.js"; export { projects } from "./projects.js"; export { projectMemberships } from "./project_memberships.js"; export { projectWorkspaces } from "./project_workspaces.js"; +export { pipelineCaseBlockers } from "./pipelines.js"; +export { pipelineCaseEvents } from "./pipelines.js"; +export { pipelineCaseIssueLinks } from "./pipelines.js"; +export { pipelineCases } from "./pipelines.js"; +export { pipelineGuidanceDocuments } from "./pipelines.js"; +export { pipelineStages } from "./pipelines.js"; +export { pipelineTransitions } from "./pipelines.js"; +export { pipelines } from "./pipelines.js"; export { executionWorkspaces } from "./execution_workspaces.js"; export { environments } from "./environments.js"; export { environmentLeases } from "./environment_leases.js"; diff --git a/packages/db/src/schema/pipelines.ts b/packages/db/src/schema/pipelines.ts new file mode 100644 index 00000000000..cc69cc59c16 --- /dev/null +++ b/packages/db/src/schema/pipelines.ts @@ -0,0 +1,165 @@ +import { + type AnyPgColumn, + boolean, + index, + integer, + jsonb, + pgTable, + text, + timestamp, + uniqueIndex, + uuid, +} from "drizzle-orm/pg-core"; +import type { PipelineStageConfig } from "@paperclipai/shared"; +import { companies } from "./companies.js"; +import { issues } from "./issues.js"; + +export const pipelines = pgTable( + "pipelines", + { + id: uuid("id").primaryKey().defaultRandom(), + companyId: uuid("company_id").notNull().references(() => companies.id, { onDelete: "cascade" }), + name: text("name").notNull(), + description: text("description"), + status: text("status").notNull().default("active"), + createdAt: timestamp("created_at", { withTimezone: true }).notNull().defaultNow(), + updatedAt: timestamp("updated_at", { withTimezone: true }).notNull().defaultNow(), + }, + (table) => ({ + companyNameIdx: index("pipelines_company_name_idx").on(table.companyId, table.name), + }), +); + +export const pipelineStages = pgTable( + "pipeline_stages", + { + id: uuid("id").primaryKey().defaultRandom(), + pipelineId: uuid("pipeline_id").notNull().references(() => pipelines.id, { onDelete: "cascade" }), + name: text("name").notNull(), + kind: text("kind").notNull().default("open"), + position: integer("position").notNull().default(0), + config: jsonb("config").$type().notNull().default({ variables: [] }), + createdAt: timestamp("created_at", { withTimezone: true }).notNull().defaultNow(), + updatedAt: timestamp("updated_at", { withTimezone: true }).notNull().defaultNow(), + }, + (table) => ({ + pipelineIdx: index("pipeline_stages_pipeline_idx").on(table.pipelineId, table.position), + pipelinePositionUq: uniqueIndex("pipeline_stages_pipeline_position_uq").on(table.pipelineId, table.position), + }), +); + +export const pipelineTransitions = pgTable( + "pipeline_stage_transitions", + { + id: uuid("id").primaryKey().defaultRandom(), + pipelineId: uuid("pipeline_id").notNull().references(() => pipelines.id, { onDelete: "cascade" }), + fromStageId: uuid("from_stage_id").notNull().references(() => pipelineStages.id, { onDelete: "cascade" }), + toStageId: uuid("to_stage_id").notNull().references(() => pipelineStages.id, { onDelete: "cascade" }), + config: jsonb("config").notNull().default({}), + createdAt: timestamp("created_at", { withTimezone: true }).notNull().defaultNow(), + updatedAt: timestamp("updated_at", { withTimezone: true }).notNull().defaultNow(), + }, + (table) => ({ + pipelineIdx: index("pipeline_stage_transitions_pipeline_idx").on(table.pipelineId), + fromIdx: index("pipeline_stage_transitions_from_idx").on(table.fromStageId), + toIdx: index("pipeline_stage_transitions_to_idx").on(table.toStageId), + uniqueTransition: uniqueIndex("pipeline_stage_transitions_pipeline_from_to_uq") + .on(table.pipelineId, table.fromStageId, table.toStageId), + }), +); + +export const pipelineCases = pgTable( + "pipeline_cases", + { + id: uuid("id").primaryKey().defaultRandom(), + companyId: uuid("company_id").notNull().references(() => companies.id, { onDelete: "cascade" }), + pipelineId: uuid("pipeline_id").notNull().references(() => pipelines.id, { onDelete: "cascade" }), + stageId: uuid("stage_id").references((): AnyPgColumn => pipelineStages.id, { onDelete: "set null" }), + parentCaseId: uuid("parent_case_id").references((): AnyPgColumn => pipelineCases.id, { onDelete: "set null" }), + title: text("title").notNull(), + status: text("status").notNull().default("open"), + fields: jsonb("fields").$type>().notNull().default({}), + createdByActorId: text("created_by_actor_id"), + createdByActorType: text("created_by_actor_type"), + lastActivityAt: timestamp("last_activity_at", { withTimezone: true }), + createdAt: timestamp("created_at", { withTimezone: true }).notNull().defaultNow(), + updatedAt: timestamp("updated_at", { withTimezone: true }).notNull().defaultNow(), + }, + (table) => ({ + pipelineIdx: index("pipeline_cases_pipeline_idx").on(table.pipelineId, table.stageId), + stageIdx: index("pipeline_cases_stage_idx").on(table.stageId), + companyIdx: index("pipeline_cases_company_idx").on(table.companyId), + stageStatusIdx: index("pipeline_cases_stage_status_idx").on(table.pipelineId, table.status), + parentIdx: index("pipeline_cases_parent_idx").on(table.parentCaseId), + }), +); + +export const pipelineCaseEvents = pgTable( + "pipeline_case_events", + { + id: uuid("id").primaryKey().defaultRandom(), + companyId: uuid("company_id").notNull().references(() => companies.id, { onDelete: "cascade" }), + pipelineId: uuid("pipeline_id").notNull().references(() => pipelines.id, { onDelete: "cascade" }), + caseId: uuid("case_id").notNull().references((): AnyPgColumn => pipelineCases.id, { onDelete: "cascade" }), + kind: text("kind").notNull(), + summary: text("summary"), + payload: jsonb("payload").$type>(), + createdAt: timestamp("created_at", { withTimezone: true }).notNull().defaultNow(), + }, + (table) => ({ + pipelineIdx: index("pipeline_case_events_pipeline_idx").on(table.pipelineId, table.caseId), + caseIdx: index("pipeline_case_events_case_idx").on(table.caseId), + kindIdx: index("pipeline_case_events_kind_idx").on(table.kind), + }), +); + +export const pipelineGuidanceDocuments = pgTable( + "pipeline_guidance_documents", + { + id: uuid("id").primaryKey().defaultRandom(), + pipelineId: uuid("pipeline_id").notNull().references(() => pipelines.id, { onDelete: "cascade" }), + key: text("key").notNull(), + title: text("title").notNull(), + body: text("body").notNull(), + createdAt: timestamp("created_at", { withTimezone: true }).notNull().defaultNow(), + updatedAt: timestamp("updated_at", { withTimezone: true }).notNull().defaultNow(), + }, + (table) => ({ + pipelineIdx: index("pipeline_guidance_documents_pipeline_idx").on(table.pipelineId), + pipelineKeyUq: uniqueIndex("pipeline_guidance_documents_pipeline_key_uq").on(table.pipelineId, table.key), + }), +); + +export const pipelineCaseIssueLinks = pgTable( + "pipeline_case_issue_links", + { + id: uuid("id").primaryKey().defaultRandom(), + caseId: uuid("case_id").notNull().references(() => pipelineCases.id, { onDelete: "cascade" }), + issueId: uuid("issue_id").notNull().references(() => issues.id, { onDelete: "cascade" }), + role: text("role").notNull().default("automation"), + createdAt: timestamp("created_at", { withTimezone: true }).notNull().defaultNow(), + }, + (table) => ({ + caseIdx: index("pipeline_case_issue_links_case_idx").on(table.caseId, table.issueId), + issueIdx: index("pipeline_case_issue_links_issue_idx").on(table.issueId, table.caseId), + caseIssueUq: uniqueIndex("pipeline_case_issue_links_case_issue_uq").on(table.caseId, table.issueId), + }), +); + +export const pipelineCaseBlockers = pgTable( + "pipeline_case_blockers", + { + id: uuid("id").primaryKey().defaultRandom(), + caseId: uuid("case_id").notNull().references((): AnyPgColumn => pipelineCases.id, { onDelete: "cascade" }), + blockedByCaseId: uuid("blocked_by_case_id").notNull().references((): AnyPgColumn => pipelineCases.id, { onDelete: "cascade" }), + reason: text("reason"), + resolved: boolean("resolved").notNull().default(false), + createdAt: timestamp("created_at", { withTimezone: true }).notNull().defaultNow(), + resolvedAt: timestamp("resolved_at", { withTimezone: true }), + }, + (table) => ({ + caseIdx: index("pipeline_case_blockers_case_idx").on(table.caseId), + blockedByIdx: index("pipeline_case_blockers_blocked_by_idx").on(table.blockedByCaseId), + casePairUq: uniqueIndex("pipeline_case_blockers_pair_uq").on(table.caseId, table.blockedByCaseId), + }), +); diff --git a/packages/shared/src/index.ts b/packages/shared/src/index.ts index d31d6287528..62d391deccd 100644 --- a/packages/shared/src/index.ts +++ b/packages/shared/src/index.ts @@ -1144,6 +1144,32 @@ export { runRoutineIntakeFormSchema, pipelineStageConfigSchema, pipelineStageVariableSchema, + pipelineStageRunRoutineSchema, + pipelineStatusSchema, + pipelineStageKindSchema, + pipelineCaseIssueLinkRoleSchema, + createPipelineSchema, + updatePipelineSchema, + pipelineListQuerySchema, + createPipelineStageSchema, + updatePipelineStageSchema, + pipelineTransitionConfigSchema, + pipelineTransitionSchema, + setPipelineTransitionsSchema, + createPipelineGuidanceDocumentSchema, + listPipelineCasesQuerySchema, + listCompanyCaseEventsQuerySchema, + pipelineCaseIngestSchema, + pipelineCaseBatchIngestSchema, + updatePipelineCaseSchema, + transitionCaseSchema, + resolveCaseSuggestionSchema, + reviewCaseSchema, + reviewCasesQuerySchema, + bulkReviewCasesSchema, + createPipelineCaseBlockerSchema, + createPipelineCaseIssueLinkSchema, + openConversationPayloadSchema, rotateRoutineTriggerSecretSchema, routineRevisionSnapshotRoutineV1Schema, routineRevisionSnapshotTriggerV1Schema, @@ -1166,6 +1192,31 @@ export { type RunRoutineIntakeForm, type PipelineStageConfig, type PipelineStageVariable, + type PipelineStageRunRoutineConfig, + type PipelineStatus, + type PipelineStageKind, + type PipelineCaseIssueLinkRole, + type CreatePipeline, + type UpdatePipeline, + type PipelineListQuery, + type CreatePipelineStage, + type UpdatePipelineStage, + type PipelineTransition, + type SetPipelineTransitions, + type PipelineGuidanceDocumentUpsert, + type ListPipelineCasesQuery, + type ListCompanyCaseEventsQuery, + type PipelineCaseIngest, + type PipelineCaseBatchIngest, + type UpdatePipelineCase, + type TransitionCaseInput, + type ResolveSuggestionInput, + type ReviewCaseInput, + type ReviewCasesQuery, + type BulkReviewCasesInput, + type CreatePipelineCaseBlocker, + type CreatePipelineCaseIssueLink, + type OpenConversationPayload, type RotateRoutineTriggerSecret, createCostEventSchema, createFinanceEventSchema, diff --git a/packages/shared/src/validators/index.ts b/packages/shared/src/validators/index.ts index f97cdd06cc3..78ca6f75f9e 100644 --- a/packages/shared/src/validators/index.ts +++ b/packages/shared/src/validators/index.ts @@ -468,8 +468,61 @@ export { export { pipelineStageConfigSchema, pipelineStageVariableSchema, + pipelineStageRunRoutineSchema, + pipelineStatusSchema, + pipelineStageKindSchema, + pipelineCaseIssueLinkRoleSchema, + createPipelineSchema, + updatePipelineSchema, + pipelineListQuerySchema, + createPipelineStageSchema, + updatePipelineStageSchema, + pipelineTransitionConfigSchema, + pipelineTransitionSchema, + setPipelineTransitionsSchema, + createPipelineGuidanceDocumentSchema, + listPipelineCasesQuerySchema, + listCompanyCaseEventsQuerySchema, + pipelineCaseIngestSchema, + pipelineCaseBatchIngestSchema, + updatePipelineCaseSchema, + transitionCaseSchema, + resolveCaseSuggestionSchema, + reviewCaseSchema, + reviewCasesQuerySchema, + bulkReviewCasesSchema, + createPipelineCaseBlockerSchema, + createPipelineCaseIssueLinkSchema, + openConversationPayloadSchema, type PipelineStageConfig, type PipelineStageVariable, + type PipelineStatus, + type PipelineStageKind, + type PipelineCaseIssueLinkRole, + type PipelineStageRunRoutineConfig, + type CreatePipeline, + type UpdatePipeline, + type PipelineListQuery, + type CreatePipelineStage, + type UpdatePipelineStage, + type PipelineTransition, + type SetPipelineTransitions, + type PipelineGuidanceDocumentUpsert, + type ListPipelineCasesQuery, + type ListCompanyCaseEventsQuery, + type PipelineCaseIngest, + type PipelineCaseBatchIngest, + type UpdatePipelineCase, + type TransitionCaseInput, + type ResolveSuggestionInput, + type ReviewCaseInput, + type ReviewCasesQuery, + type BulkReviewCasesInput, + type CreatePipelineCaseBlocker, + type CreatePipelineCaseIssueLink, + type OpenConversationPayload, + type PipelineCaseTransitionConfig, + type PipelineCaseConfig, } from "./pipeline.js"; export { diff --git a/packages/shared/src/validators/pipeline.test.ts b/packages/shared/src/validators/pipeline.test.ts index 79d93e8ea58..ad5c2f84bc8 100644 --- a/packages/shared/src/validators/pipeline.test.ts +++ b/packages/shared/src/validators/pipeline.test.ts @@ -26,4 +26,33 @@ describe("pipeline stage variable schema", () => { }).success, ).toBe(false); }); + + it("accepts run_routine onEnter action", () => { + expect( + pipelineStageConfigSchema.safeParse({ + variables: [], + onEnter: { + run_routine: { + routineId: "11111111-1111-4111-8111-111111111111", + variables: { issue_title: "title" }, + payload: { source: "pipeline" }, + caseFields: { requestedBy: "requester" }, + }, + }, + }).success, + ).toBe(true); + }); + + it("rejects invalid onEnter run_routine configuration", () => { + expect( + pipelineStageConfigSchema.safeParse({ + variables: [], + onEnter: { + run_routine: { + routineId: "not-a-uuid", + }, + }, + }).success, + ).toBe(false); + }); }); diff --git a/packages/shared/src/validators/pipeline.ts b/packages/shared/src/validators/pipeline.ts index 347b2a8dd44..e14556000fe 100644 --- a/packages/shared/src/validators/pipeline.ts +++ b/packages/shared/src/validators/pipeline.ts @@ -2,6 +2,22 @@ import { z } from "zod"; const routineVariableLikeNameSchema = z.string().trim().regex(/^[A-Za-z][A-Za-z0-9_]*$/); +const pipelineStageRunRoutineVariableMapSchema = z.record( + z.string().trim().min(1).max(120), + z.string().trim().max(120), +); + +export const pipelineStageRunRoutineSchema = z.object({ + routineId: z.string().uuid(), + variables: pipelineStageRunRoutineVariableMapSchema.optional().default({}), + payload: z.record(z.string(), z.unknown()).optional().default({}), + caseFields: pipelineStageRunRoutineVariableMapSchema.optional().default({}), +}).strict(); + +const pipelineStageOnEnterSchema = z.object({ + run_routine: pipelineStageRunRoutineSchema, +}).strict(); + export const pipelineStageVariableSchema = z.object({ key: routineVariableLikeNameSchema, label: z.string().trim().max(120), @@ -28,6 +44,7 @@ export const pipelineStageVariableSchema = z.object({ export const pipelineStageConfigSchema = z.object({ variables: z.array(pipelineStageVariableSchema).default([]), + onEnter: pipelineStageOnEnterSchema.optional(), }).passthrough().superRefine((value, ctx) => { const keys = new Set(); value.variables.forEach((variable, index) => { @@ -42,5 +59,221 @@ export const pipelineStageConfigSchema = z.object({ }); }); -export type PipelineStageVariable = z.infer; +export const pipelineStatusSchema = z.enum(["active", "paused", "archived"]); +export const pipelineStageKindSchema = z.enum(["open", "working", "review", "done", "cancelled"]); +export const pipelineCaseIssueLinkRoleSchema = z.enum(["origin", "conversation", "work", "automation"]); + +export const createPipelineSchema = z.object({ + name: z.string().trim().min(1).max(180), + description: z.string().trim().max(5_000).nullable().optional(), + status: pipelineStatusSchema.optional().default("active"), +}).strict(); + +export const updatePipelineSchema = createPipelineSchema.partial().strict(); + +export const pipelineListQuerySchema = z.object({ + includeConnections: z.preprocess((value) => { + if (value === undefined) return undefined; + if (typeof value === "boolean") return value; + if (typeof value === "string") return ["1", "true", "yes", "y"].includes(value.toLowerCase()); + return undefined; + }, z.boolean().optional()), + includeCounts: z.preprocess((value) => { + if (value === undefined) return undefined; + if (typeof value === "boolean") return value; + if (typeof value === "string") return ["1", "true", "yes", "y"].includes(value.toLowerCase()); + return undefined; + }, z.boolean().optional()), + q: z.string().trim().max(200).optional(), +}).strict(); + +export const createPipelineStageSchema = z.object({ + name: z.string().trim().min(1).max(180), + kind: pipelineStageKindSchema.optional().default("open"), + position: z.number().int().nonnegative().optional(), + config: pipelineStageConfigSchema.optional(), +}).strict(); + +export const updatePipelineStageSchema = createPipelineStageSchema.partial().strict(); + +export const pipelineTransitionConfigSchema = z.record(z.unknown()).default({}); + +export const pipelineTransitionSchema = z.object({ + fromStageId: z.string().uuid(), + toStageId: z.string().uuid(), + config: pipelineTransitionConfigSchema, +}).strict(); + +export const setPipelineTransitionsSchema = z.object({ + transitions: z.array(pipelineTransitionSchema), + enforceTransitions: z.boolean().optional().default(true), +}).strict().superRefine((value, ctx) => { + const transitionsByPair = new Map(); + value.transitions.forEach((transition, index) => { + const key = `${transition.fromStageId}:${transition.toStageId}`; + const indexOfFirst = transitionsByPair.get(key); + if (indexOfFirst !== undefined) { + ctx.addIssue({ + code: z.ZodIssueCode.custom, + path: ["transitions", index], + message: `Duplicate transition from ${transition.fromStageId} to ${transition.toStageId}`, + }); + return; + } + transitionsByPair.set(key, index); + if (transition.fromStageId === transition.toStageId) { + ctx.addIssue({ + code: z.ZodIssueCode.custom, + path: ["transitions", index], + message: "Transition destination must differ from source", + }); + } + }); +}); + +export const createPipelineGuidanceDocumentSchema = z.object({ + title: z.string().trim().min(1).max(180), + body: z.string().trim().min(1).max(200_000), +}).strict(); + +export const listPipelineCasesQuerySchema = z.object({ + stageId: z.string().uuid().optional(), + status: z.string().trim().max(80).optional(), + q: z.string().trim().max(200).optional(), + limit: z.preprocess((value) => { + if (value === undefined || value === null || value === "") return undefined; + if (typeof value === "number") return value; + if (typeof value === "string") return Number(value); + return undefined; + }, z.number().int().positive().max(500).default(100)), + offset: z.preprocess((value) => { + if (value === undefined || value === null || value === "") return undefined; + if (typeof value === "number") return value; + if (typeof value === "string") return Number(value); + return undefined; + }, z.number().int().min(0).default(0)), +}).strict(); + +export const listCompanyCaseEventsQuerySchema = z.object({ + pipelineId: z.string().uuid().optional(), + types: z.string().trim().max(500).optional(), + limit: z.preprocess((value) => { + if (value === undefined || value === null || value === "") return undefined; + if (typeof value === "number") return value; + if (typeof value === "string") return Number(value); + return undefined; + }, z.number().int().positive().max(500).default(100)), + offset: z.preprocess((value) => { + if (value === undefined || value === null || value === "") return undefined; + if (typeof value === "number") return value; + if (typeof value === "string") return Number(value); + return undefined; + }, z.number().int().min(0).default(0)), +}).strict(); + +export const pipelineCaseIngestSchema = z.object({ + title: z.string().trim().min(1).max(300), + status: z.string().trim().max(80).optional().default("open"), + stageId: z.string().uuid().optional().nullable(), + parentCaseId: z.string().uuid().optional().nullable(), + fields: z.record(z.string(), z.unknown()).optional().default({}), +}).strict().passthrough(); + +export const pipelineCaseBatchIngestSchema = z.object({ + items: z.array(pipelineCaseIngestSchema).min(1).max(100), +}).strict(); + +export const updatePipelineCaseSchema = z.object({ + title: z.string().trim().min(1).max(300).optional(), + status: z.string().trim().max(80).optional(), + stageId: z.string().uuid().optional().nullable(), + fields: z.record(z.string(), z.unknown()).optional(), +}).strict().passthrough(); + +export const transitionCaseSchema = z.object({ + toStageId: z.string().uuid(), + reason: z.string().trim().max(1_000).optional().nullable().default(null), +}).strict(); + +export const resolveCaseSuggestionSchema = z.object({ + decision: z.enum(["accept", "decline"]), + note: z.string().trim().max(2_000).optional().nullable().default(null), +}).strict(); + +export const reviewCaseSchema = z.object({ + decision: z.enum(["approve", "request_changes", "drop"]), + note: z.string().trim().max(2_000).optional().nullable().default(null), +}).strict(); + +export const reviewCasesQuerySchema = z.object({ + pipelineId: z.string().uuid().optional(), + kind: z.string().trim().max(80).optional(), + limit: z.preprocess((value) => { + if (value === undefined || value === null || value === "") return undefined; + if (typeof value === "number") return value; + if (typeof value === "string") return Number(value); + return undefined; + }, z.number().int().positive().max(500).default(100)), + offset: z.preprocess((value) => { + if (value === undefined || value === null || value === "") return undefined; + if (typeof value === "number") return value; + if (typeof value === "string") return Number(value); + return undefined; + }, z.number().int().min(0).default(0)), +}).strict(); + +export const bulkReviewCasesSchema = z.object({ + caseIds: z.array(z.string().uuid()).min(1).max(100), + decision: z.enum(["approve", "request_changes", "drop"]), + note: z.string().trim().max(2_000).optional().nullable().default(null), +}).strict(); + +export const createPipelineCaseBlockerSchema = z.object({ + blockedByCaseId: z.string().uuid(), + reason: z.string().trim().max(1_000).optional().nullable().default(null), +}).strict(); + +export const createPipelineCaseIssueLinkSchema = z.object({ + issueId: z.string().uuid(), + role: pipelineCaseIssueLinkRoleSchema, +}).strict(); + +export const openConversationPayloadSchema = z.object({ + title: z.string().trim().max(300).optional(), + description: z.string().trim().max(20_000).optional(), + status: z.string().trim().max(80).optional().default("todo"), + assigneeAgentId: z.string().uuid().optional().nullable(), + assigneeUserId: z.string().trim().max(200).optional().nullable(), + projectId: z.string().uuid().optional().nullable(), + role: pipelineCaseIssueLinkRoleSchema.optional().default("conversation"), +}).strict(); + +export type PipelineStatus = z.infer; +export type PipelineStageKind = z.infer; +export type PipelineCaseIssueLinkRole = z.infer; +export type CreatePipeline = z.infer; +export type UpdatePipeline = z.infer; +export type PipelineListQuery = z.infer; +export type CreatePipelineStage = z.infer; +export type UpdatePipelineStage = z.infer; +export type PipelineTransition = z.infer; +export type SetPipelineTransitions = z.infer; +export type PipelineGuidanceDocumentUpsert = z.infer; +export type ListPipelineCasesQuery = z.infer; +export type ListCompanyCaseEventsQuery = z.infer; +export type PipelineCaseIngest = z.infer; +export type PipelineCaseBatchIngest = z.infer; +export type UpdatePipelineCase = z.infer; +export type TransitionCaseInput = z.infer; +export type ResolveSuggestionInput = z.infer; +export type ReviewCaseInput = z.infer; +export type ReviewCasesQuery = z.infer; +export type BulkReviewCasesInput = z.infer; +export type CreatePipelineCaseBlocker = z.infer; +export type CreatePipelineCaseIssueLink = z.infer; +export type OpenConversationPayload = z.infer; +export type PipelineCaseTransitionConfig = Record; +export type PipelineCaseConfig = Record; export type PipelineStageConfig = z.infer; +export type PipelineStageVariable = z.infer; +export type PipelineStageRunRoutineConfig = z.infer; diff --git a/server/src/app.ts b/server/src/app.ts index e91bc00a59b..4743ab40d52 100644 --- a/server/src/app.ts +++ b/server/src/app.ts @@ -39,6 +39,7 @@ import { type InstanceDatabaseBackupService, } from "./routes/instance-database-backups.js"; import { llmRoutes } from "./routes/llms.js"; +import { pipelineRoutes } from "./routes/pipelines.js"; import { authRoutes } from "./routes/auth.js"; import { assetRoutes } from "./routes/assets.js"; import { accessRoutes } from "./routes/access.js"; @@ -228,6 +229,7 @@ export async function createApp( api.use(environmentRoutes(db, { pluginWorkerManager: workerManager })); api.use(executionWorkspaceRoutes(db)); api.use(goalRoutes(db)); + api.use(pipelineRoutes(db)); api.use(approvalRoutes(db, { pluginWorkerManager: workerManager })); api.use(secretRoutes(db)); api.use(costRoutes(db, { pluginWorkerManager: workerManager })); diff --git a/server/src/routes/index.ts b/server/src/routes/index.ts index 940047f29f9..8c2fb92b1c1 100644 --- a/server/src/routes/index.ts +++ b/server/src/routes/index.ts @@ -19,6 +19,7 @@ export { sidebarPreferenceRoutes } from "./sidebar-preferences.js"; export { resourceMembershipRoutes } from "./resource-memberships.js"; export { inboxDismissalRoutes } from "./inbox-dismissals.js"; export { llmRoutes } from "./llms.js"; +export { pipelineRoutes } from "./pipelines.js"; export { accessRoutes } from "./access.js"; export { instanceSettingsRoutes } from "./instance-settings.js"; export { instanceDatabaseBackupRoutes } from "./instance-database-backups.js"; diff --git a/server/src/routes/pipelines.ts b/server/src/routes/pipelines.ts new file mode 100644 index 00000000000..ad5105f4a12 --- /dev/null +++ b/server/src/routes/pipelines.ts @@ -0,0 +1,886 @@ +import { Router } from "express"; +import type { Db } from "@paperclipai/db"; +import { + createPipelineCaseBlockerSchema, + createPipelineCaseIssueLinkSchema, + createPipelineGuidanceDocumentSchema, + openConversationPayloadSchema, + createPipelineSchema, + createPipelineStageSchema, + bulkReviewCasesSchema, + listCompanyCaseEventsQuerySchema, + listPipelineCasesQuerySchema, + pipelineCaseBatchIngestSchema, + pipelineCaseIngestSchema, + pipelineListQuerySchema, + resolveCaseSuggestionSchema, + reviewCaseSchema, + reviewCasesQuerySchema, + setPipelineTransitionsSchema, + transitionCaseSchema, + updatePipelineCaseSchema, + updatePipelineSchema, + updatePipelineStageSchema, +} from "@paperclipai/shared"; +import { extractPipelineIntakeFormFields } from "../services/pipelines.js"; +import { validate } from "../middleware/validate.js"; +import { pipelineService, issueService, logActivity } from "../services/index.js"; +import { assertCompanyAccess, getActorInfo } from "./authz.js"; + +export function pipelineRoutes(db: Db) { + const router = Router(); + const svc = pipelineService(db); + + async function assertPipelineAccess(pipelineId: string, req: Parameters[0]) { + const pipeline = await svc.getPipeline(pipelineId); + if (!pipeline) { + return { pipeline: null as null }; + } + assertCompanyAccess(req, pipeline.companyId); + return { pipeline }; + } + + async function assertCaseAccess(caseId: string, req: Parameters[0]) { + const pipelineCase = await svc.getPipelineCase(caseId); + if (!pipelineCase) { + return { pipelineCase: null as null }; + } + const pipelineResult = await assertPipelineAccess(pipelineCase.pipelineId, req); + if (!pipelineResult.pipeline) { + return { pipelineCase: null as null }; + } + return { pipelineCase, pipeline: pipelineResult.pipeline }; + } + + router.get("/companies/:companyId/pipelines", async (req, res) => { + const companyId = req.params.companyId as string; + assertCompanyAccess(req, companyId); + const query = pipelineListQuerySchema.parse(req.query); + const pipelines = await svc.listPipelines(companyId, query); + res.json(pipelines); + }); + + router.get("/companies/:companyId/case-events", async (req, res) => { + const companyId = req.params.companyId as string; + assertCompanyAccess(req, companyId); + const query = listCompanyCaseEventsQuerySchema.parse(req.query); + const events = await svc.listCompanyCaseEvents(companyId, query); + res.json(events); + }); + + router.get("/pipelines/:pipelineId", async (req, res) => { + const pipelineId = req.params.pipelineId as string; + const pipeline = await svc.getPipelineWithRelations(pipelineId); + if (!pipeline) { + res.status(404).json({ error: "Pipeline not found" }); + return; + } + assertCompanyAccess(req, pipeline.companyId); + res.json(pipeline); + }); + + router.post("/companies/:companyId/pipelines", validate(createPipelineSchema), async (req, res) => { + const companyId = req.params.companyId as string; + assertCompanyAccess(req, companyId); + const pipeline = await svc.createPipeline(companyId, req.body); + + const actor = getActorInfo(req); + await logActivity(db, { + companyId, + actorType: actor.actorType, + actorId: actor.actorId, + agentId: actor.agentId, + action: "pipeline.created", + entityType: "pipeline", + entityId: pipeline.id, + details: req.body, + }); + + res.status(201).json(pipeline); + }); + + router.patch("/pipelines/:pipelineId", validate(updatePipelineSchema), async (req, res) => { + const pipelineId = req.params.pipelineId as string; + const existing = await svc.getPipeline(pipelineId); + if (!existing) { + res.status(404).json({ error: "Pipeline not found" }); + return; + } + assertCompanyAccess(req, existing.companyId); + + const pipeline = await svc.updatePipeline(pipelineId, req.body); + if (!pipeline) { + res.status(404).json({ error: "Pipeline not found" }); + return; + } + + const actor = getActorInfo(req); + await logActivity(db, { + companyId: pipeline.companyId, + actorType: actor.actorType, + actorId: actor.actorId, + agentId: actor.agentId, + action: "pipeline.updated", + entityType: "pipeline", + entityId: pipeline.id, + details: req.body, + }); + + res.json(pipeline); + }); + + router.delete("/pipelines/:pipelineId", async (req, res) => { + const pipelineId = req.params.pipelineId as string; + const existing = await svc.getPipeline(pipelineId); + if (!existing) { + res.status(404).json({ error: "Pipeline not found" }); + return; + } + assertCompanyAccess(req, existing.companyId); + + const pipeline = await svc.removePipeline(pipelineId); + if (!pipeline) { + res.status(404).json({ error: "Pipeline not found" }); + return; + } + + const actor = getActorInfo(req); + await logActivity(db, { + companyId: pipeline.companyId, + actorType: actor.actorType, + actorId: actor.actorId, + agentId: actor.agentId, + action: "pipeline.deleted", + entityType: "pipeline", + entityId: pipeline.id, + }); + + res.json(pipeline); + }); + + router.get("/pipelines/:pipelineId/stages", async (req, res) => { + const pipelineId = req.params.pipelineId as string; + const access = await assertPipelineAccess(pipelineId, req); + if (!access.pipeline) { + res.status(404).json({ error: "Pipeline not found" }); + return; + } + const stages = await svc.listPipelineStages(pipelineId); + res.json(stages); + }); + + router.post("/pipelines/:pipelineId/stages", validate(createPipelineStageSchema), async (req, res) => { + const pipelineId = req.params.pipelineId as string; + const access = await assertPipelineAccess(pipelineId, req); + if (!access.pipeline) { + res.status(404).json({ error: "Pipeline not found" }); + return; + } + + const stage = await svc.createPipelineStage(pipelineId, req.body); + const actor = getActorInfo(req); + await logActivity(db, { + companyId: access.pipeline.companyId, + actorType: actor.actorType, + actorId: actor.actorId, + agentId: actor.agentId, + action: "pipeline_stage.created", + entityType: "pipeline_stage", + entityId: stage.id, + details: { pipelineId, input: req.body }, + }); + + res.status(201).json(stage); + }); + + router.patch("/pipelines/:pipelineId/stages/:stageId", validate(updatePipelineStageSchema), async (req, res) => { + const pipelineId = req.params.pipelineId as string; + const stageId = req.params.stageId as string; + const access = await assertPipelineAccess(pipelineId, req); + if (!access.pipeline) { + res.status(404).json({ error: "Pipeline not found" }); + return; + } + + const stage = await svc.updatePipelineStage(pipelineId, stageId, req.body); + if (!stage) { + res.status(404).json({ error: "Pipeline stage not found" }); + return; + } + + const actor = getActorInfo(req); + await logActivity(db, { + companyId: access.pipeline.companyId, + actorType: actor.actorType, + actorId: actor.actorId, + agentId: actor.agentId, + action: "pipeline_stage.updated", + entityType: "pipeline_stage", + entityId: stage.id, + details: { pipelineId, stageId, input: req.body }, + }); + + res.json(stage); + }); + + router.delete("/pipelines/:pipelineId/stages/:stageId", async (req, res) => { + const pipelineId = req.params.pipelineId as string; + const stageId = req.params.stageId as string; + const access = await assertPipelineAccess(pipelineId, req); + if (!access.pipeline) { + res.status(404).json({ error: "Pipeline not found" }); + return; + } + + const stage = await svc.removePipelineStage(pipelineId, stageId); + if (!stage) { + res.status(404).json({ error: "Pipeline stage not found" }); + return; + } + + const actor = getActorInfo(req); + await logActivity(db, { + companyId: access.pipeline.companyId, + actorType: actor.actorType, + actorId: actor.actorId, + agentId: actor.agentId, + action: "pipeline_stage.deleted", + entityType: "pipeline_stage", + entityId: stage.id, + details: { pipelineId, stageId }, + }); + + res.json(stage); + }); + + router.get("/pipelines/:pipelineId/transitions", async (req, res) => { + const pipelineId = req.params.pipelineId as string; + const access = await assertPipelineAccess(pipelineId, req); + if (!access.pipeline) { + res.status(404).json({ error: "Pipeline not found" }); + return; + } + const transitions = await svc.listPipelineTransitions(pipelineId); + res.json(transitions); + }); + + router.put("/pipelines/:pipelineId/transitions", validate(setPipelineTransitionsSchema), async (req, res) => { + const pipelineId = req.params.pipelineId as string; + const access = await assertPipelineAccess(pipelineId, req); + if (!access.pipeline) { + res.status(404).json({ error: "Pipeline not found" }); + return; + } + const transitions = await svc.setPipelineTransitions(pipelineId, req.body); + + const actor = getActorInfo(req); + await logActivity(db, { + companyId: access.pipeline.companyId, + actorType: actor.actorType, + actorId: actor.actorId, + agentId: actor.agentId, + action: "pipeline.transitions_updated", + entityType: "pipeline", + entityId: pipelineId, + details: req.body, + }); + + res.json({ transitions }); + }); + + router.get("/pipelines/:pipelineId/guidance-documents", async (req, res) => { + const pipelineId = req.params.pipelineId as string; + const access = await assertPipelineAccess(pipelineId, req); + if (!access.pipeline) { + res.status(404).json({ error: "Pipeline not found" }); + return; + } + const docs = await svc.listGuidanceDocuments(pipelineId); + res.json(docs); + }); + + router.get("/pipelines/:pipelineId/guidance-documents/:key", async (req, res) => { + const pipelineId = req.params.pipelineId as string; + const key = req.params.key as string; + const access = await assertPipelineAccess(pipelineId, req); + if (!access.pipeline) { + res.status(404).json({ error: "Pipeline not found" }); + return; + } + + const doc = await svc.getGuidanceDocument(pipelineId, key); + if (!doc) { + res.status(404).json({ error: "Guidance document not found" }); + return; + } + + res.json(doc); + }); + + router.put( + "/pipelines/:pipelineId/guidance-documents/:key", + validate(createPipelineGuidanceDocumentSchema), + async (req, res) => { + const pipelineId = req.params.pipelineId as string; + const key = req.params.key as string; + const access = await assertPipelineAccess(pipelineId, req); + if (!access.pipeline) { + res.status(404).json({ error: "Pipeline not found" }); + return; + } + const doc = await svc.upsertGuidanceDocument(pipelineId, key, req.body); + + const actor = getActorInfo(req); + await logActivity(db, { + companyId: access.pipeline.companyId, + actorType: actor.actorType, + actorId: actor.actorId, + agentId: actor.agentId, + action: "pipeline_guidance_document.upserted", + entityType: "pipeline_guidance_document", + entityId: doc.id, + details: { pipelineId, key, ...req.body }, + }); + + res.json(doc); + }, + ); + + router.delete("/pipelines/:pipelineId/guidance-documents/:key", async (req, res) => { + const pipelineId = req.params.pipelineId as string; + const key = req.params.key as string; + const access = await assertPipelineAccess(pipelineId, req); + if (!access.pipeline) { + res.status(404).json({ error: "Pipeline not found" }); + return; + } + + const doc = await svc.deleteGuidanceDocument(pipelineId, key); + if (!doc) { + res.status(404).json({ error: "Guidance document not found" }); + return; + } + + const actor = getActorInfo(req); + await logActivity(db, { + companyId: access.pipeline.companyId, + actorType: actor.actorType, + actorId: actor.actorId, + agentId: actor.agentId, + action: "pipeline_guidance_document.deleted", + entityType: "pipeline_guidance_document", + entityId: doc.id, + details: { pipelineId, key }, + }); + + res.json(doc); + }); + + router.get("/pipelines/:pipelineId/cases", async (req, res) => { + const pipelineId = req.params.pipelineId as string; + const access = await assertPipelineAccess(pipelineId, req); + if (!access.pipeline) { + res.status(404).json({ error: "Pipeline not found" }); + return; + } + + const query = listPipelineCasesQuerySchema.parse(req.query); + const cases = await svc.listPipelineCases(pipelineId, query); + res.json(cases); + }); + + router.post("/pipelines/:pipelineId/cases/ingest", validate(pipelineCaseIngestSchema), async (req, res) => { + const pipelineId = req.params.pipelineId as string; + const access = await assertPipelineAccess(pipelineId, req); + if (!access.pipeline) { + res.status(404).json({ error: "Pipeline not found" }); + return; + } + + const actor = getActorInfo(req); + const created = await svc.ingestPipelineCase(pipelineId, req.body, { + actorId: actor.actorId, + actorType: actor.actorType, + }); + + await logActivity(db, { + companyId: access.pipeline.companyId, + actorType: actor.actorType, + actorId: actor.actorId, + agentId: actor.agentId, + action: "pipeline_case.ingested", + entityType: "pipeline_case", + entityId: created.id, + details: req.body, + }); + + res.status(201).json(created); + }); + + router.post( + "/pipelines/:pipelineId/cases/batch", + validate(pipelineCaseBatchIngestSchema), + async (req, res) => { + const pipelineId = req.params.pipelineId as string; + const access = await assertPipelineAccess(pipelineId, req); + if (!access.pipeline) { + res.status(404).json({ error: "Pipeline not found" }); + return; + } + + const actor = getActorInfo(req); + const cases = await svc.ingestPipelineCases(pipelineId, req.body.items, { + actorId: actor.actorId, + actorType: actor.actorType, + }); + + await logActivity(db, { + companyId: access.pipeline.companyId, + actorType: actor.actorType, + actorId: actor.actorId, + agentId: actor.agentId, + action: "pipeline_case.batch_ingested", + entityType: "pipeline", + entityId: pipelineId, + details: { caseIds: cases.map((item) => item.id), count: cases.length }, + }); + + res.status(201).json({ cases }); + }, + ); + + router.get("/cases/:caseId", async (req, res) => { + const caseId = req.params.caseId as string; + const caseAccess = await assertCaseAccess(caseId, req); + if (!caseAccess.pipelineCase || !caseAccess.pipeline) { + res.status(404).json({ error: "Case not found" }); + return; + } + + res.json(caseAccess.pipelineCase); + }); + + router.patch("/cases/:caseId", validate(updatePipelineCaseSchema), async (req, res) => { + const caseId = req.params.caseId as string; + const caseAccess = await assertCaseAccess(caseId, req); + if (!caseAccess.pipelineCase || !caseAccess.pipeline) { + res.status(404).json({ error: "Case not found" }); + return; + } + + const updated = await svc.updatePipelineCase(caseId, req.body); + if (!updated) { + res.status(404).json({ error: "Case not found" }); + return; + } + + const actor = getActorInfo(req); + await logActivity(db, { + companyId: caseAccess.pipeline.companyId, + actorType: actor.actorType, + actorId: actor.actorId, + agentId: actor.agentId, + action: "pipeline_case.updated", + entityType: "pipeline_case", + entityId: caseId, + details: req.body, + }); + + res.json(updated); + }); + + router.post("/cases/:caseId/transition", validate(transitionCaseSchema), async (req, res) => { + const caseId = req.params.caseId as string; + const caseAccess = await assertCaseAccess(caseId, req); + if (!caseAccess.pipelineCase || !caseAccess.pipeline) { + res.status(404).json({ error: "Case not found" }); + return; + } + + const actor = getActorInfo(req); + const updated = await svc.transitionCaseStage(caseId, req.body.toStageId, { + actorId: actor.actorId, + actorType: actor.actorType, + reason: req.body.reason, + }); + if (!updated) { + res.status(404).json({ error: "Case not found" }); + return; + } + + await logActivity(db, { + companyId: caseAccess.pipeline.companyId, + actorType: actor.actorType, + actorId: actor.actorId, + agentId: actor.agentId, + action: "pipeline_case.transitioned", + entityType: "pipeline_case", + entityId: caseId, + details: req.body, + }); + + res.json(updated); + }); + + router.post("/cases/:caseId/suggest-transition", validate(transitionCaseSchema), async (req, res) => { + const caseId = req.params.caseId as string; + const caseAccess = await assertCaseAccess(caseId, req); + if (!caseAccess.pipelineCase || !caseAccess.pipeline) { + res.status(404).json({ error: "Case not found" }); + return; + } + + const updated = await svc.suggestTransition(caseId, req.body.toStageId, req.body.reason); + if (!updated) { + res.status(404).json({ error: "Case not found" }); + return; + } + + const actor = getActorInfo(req); + await logActivity(db, { + companyId: caseAccess.pipeline.companyId, + actorType: actor.actorType, + actorId: actor.actorId, + agentId: actor.agentId, + action: "pipeline_case.transition_suggested", + entityType: "pipeline_case", + entityId: caseId, + details: req.body, + }); + + res.json(updated); + }); + + router.post("/cases/:caseId/resolve-suggestion", validate(resolveCaseSuggestionSchema), async (req, res) => { + const caseId = req.params.caseId as string; + const caseAccess = await assertCaseAccess(caseId, req); + if (!caseAccess.pipelineCase || !caseAccess.pipeline) { + res.status(404).json({ error: "Case not found" }); + return; + } + + const updated = await svc.resolveSuggestion(caseId, req.body.decision, req.body.note); + if (!updated) { + res.status(404).json({ error: "Case not found" }); + return; + } + + const actor = getActorInfo(req); + await logActivity(db, { + companyId: caseAccess.pipeline.companyId, + actorType: actor.actorType, + actorId: actor.actorId, + agentId: actor.agentId, + action: "pipeline_case.suggestion_resolved", + entityType: "pipeline_case", + entityId: caseId, + details: req.body, + }); + + res.json(updated); + }); + + router.post("/cases/:caseId/review", validate(reviewCaseSchema), async (req, res) => { + const caseId = req.params.caseId as string; + const caseAccess = await assertCaseAccess(caseId, req); + if (!caseAccess.pipelineCase || !caseAccess.pipeline) { + res.status(404).json({ error: "Case not found" }); + return; + } + + const updated = await svc.reviewCase(caseId, req.body.decision, req.body.note); + if (!updated) { + res.status(404).json({ error: "Case not found" }); + return; + } + + const actor = getActorInfo(req); + await logActivity(db, { + companyId: caseAccess.pipeline.companyId, + actorType: actor.actorType, + actorId: actor.actorId, + agentId: actor.agentId, + action: "pipeline_case.reviewed", + entityType: "pipeline_case", + entityId: caseId, + details: req.body, + }); + + res.json(updated); + }); + + router.get("/companies/:companyId/review-cases", async (req, res) => { + const companyId = req.params.companyId as string; + assertCompanyAccess(req, companyId); + const query = reviewCasesQuerySchema.parse(req.query); + const rows = await svc.listReviewCases(companyId, query); + res.json(rows); + }); + + router.post("/companies/:companyId/review-cases/bulk", validate(bulkReviewCasesSchema), async (req, res) => { + const companyId = req.params.companyId as string; + assertCompanyAccess(req, companyId); + + const rows = await svc.bulkReviewCases(companyId, req.body.caseIds, req.body.decision, req.body.note); + + const actor = getActorInfo(req); + await logActivity(db, { + companyId, + actorType: actor.actorType, + actorId: actor.actorId, + agentId: actor.agentId, + action: "pipeline_case.reviewed_bulk", + entityType: "pipeline_case", + entityId: companyId, + details: { + count: rows.length, + decision: req.body.decision, + }, + }); + + res.json({ cases: rows }); + }); + + router.get("/cases/:caseId/blockers", async (req, res) => { + const caseId = req.params.caseId as string; + const caseAccess = await assertCaseAccess(caseId, req); + if (!caseAccess.pipelineCase || !caseAccess.pipeline) { + res.status(404).json({ error: "Case not found" }); + return; + } + + const blockers = await svc.listCaseBlockers(caseId); + res.json(blockers); + }); + + router.post("/cases/:caseId/blockers", validate(createPipelineCaseBlockerSchema), async (req, res) => { + const caseId = req.params.caseId as string; + const caseAccess = await assertCaseAccess(caseId, req); + if (!caseAccess.pipelineCase || !caseAccess.pipeline) { + res.status(404).json({ error: "Case not found" }); + return; + } + + const blocker = await svc.createCaseBlocker(caseId, req.body.blockedByCaseId, req.body.reason); + if (!blocker) { + res.status(404).json({ error: "Case not found" }); + return; + } + + const actor = getActorInfo(req); + await logActivity(db, { + companyId: caseAccess.pipeline.companyId, + actorType: actor.actorType, + actorId: actor.actorId, + agentId: actor.agentId, + action: "pipeline_case_blocker.created", + entityType: "pipeline_case_blocker", + entityId: blocker.id, + details: req.body, + }); + + res.status(201).json(blocker); + }); + + router.post("/cases/:caseId/blockers/:blockerId/resolve", async (req, res) => { + const caseId = req.params.caseId as string; + const blockerId = req.params.blockerId as string; + const caseAccess = await assertCaseAccess(caseId, req); + if (!caseAccess.pipelineCase || !caseAccess.pipeline) { + res.status(404).json({ error: "Case not found" }); + return; + } + + const blocker = await svc.resolveCaseBlocker(caseId, blockerId); + if (!blocker) { + res.status(404).json({ error: "Blocker not found" }); + return; + } + + const actor = getActorInfo(req); + await logActivity(db, { + companyId: caseAccess.pipeline.companyId, + actorType: actor.actorType, + actorId: actor.actorId, + agentId: actor.agentId, + action: "pipeline_case_blocker.resolved", + entityType: "pipeline_case_blocker", + entityId: blocker.id, + }); + + res.json(blocker); + }); + + router.get("/cases/:caseId/issue-links", async (req, res) => { + const caseId = req.params.caseId as string; + const caseAccess = await assertCaseAccess(caseId, req); + if (!caseAccess.pipelineCase || !caseAccess.pipeline) { + res.status(404).json({ error: "Case not found" }); + return; + } + const links = await svc.listCaseIssueLinks(caseId); + res.json(links); + }); + + router.post("/cases/:caseId/issue-links", validate(createPipelineCaseIssueLinkSchema), async (req, res) => { + const caseId = req.params.caseId as string; + const caseAccess = await assertCaseAccess(caseId, req); + if (!caseAccess.pipelineCase || !caseAccess.pipeline) { + res.status(404).json({ error: "Case not found" }); + return; + } + + const link = await svc.linkCaseIssue(caseId, req.body.issueId, req.body.role); + if (!link) { + res.status(404).json({ error: "Case not found" }); + return; + } + + const actor = getActorInfo(req); + await logActivity(db, { + companyId: caseAccess.pipeline.companyId, + actorType: actor.actorType, + actorId: actor.actorId, + agentId: actor.agentId, + action: "pipeline_case_issue_link.created", + entityType: "pipeline_case_issue_link", + entityId: link.id, + details: req.body, + }); + + res.status(201).json(link); + }); + + 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 caseAccess = await assertCaseAccess(caseId, req); + if (!caseAccess.pipelineCase || !caseAccess.pipeline) { + res.status(404).json({ error: "Case not found" }); + return; + } + + const link = await svc.unlinkCaseIssue(linkId, caseId); + if (!link) { + res.status(404).json({ error: "Issue link not found" }); + return; + } + + const actor = getActorInfo(req); + await logActivity(db, { + companyId: caseAccess.pipeline.companyId, + actorType: actor.actorType, + actorId: actor.actorId, + agentId: actor.agentId, + action: "pipeline_case_issue_link.deleted", + entityType: "pipeline_case_issue_link", + entityId: link.id, + details: { caseId, linkId }, + }); + + res.json(link); + }); + + router.get("/pipelines/:pipelineId/events", async (req, res) => { + const pipelineId = req.params.pipelineId as string; + const access = await assertPipelineAccess(pipelineId, req); + if (!access.pipeline) { + res.status(404).json({ error: "Pipeline not found" }); + return; + } + const events = await svc.listPipelineEvents(pipelineId); + res.json(events); + }); + + router.get("/cases/:caseId/events", async (req, res) => { + const caseId = req.params.caseId as string; + const caseAccess = await assertCaseAccess(caseId, req); + if (!caseAccess.pipelineCase || !caseAccess.pipeline) { + res.status(404).json({ error: "Case not found" }); + return; + } + + const events = await svc.listCaseEvents(caseId); + res.json(events); + }); + + router.get("/cases/:caseId/children", async (req, res) => { + const caseId = req.params.caseId as string; + const caseAccess = await assertCaseAccess(caseId, req); + if (!caseAccess.pipelineCase || !caseAccess.pipeline) { + res.status(404).json({ error: "Case not found" }); + return; + } + + const children = await svc.listCaseChildren(caseId); + res.json(children); + }); + + router.get("/pipelines/:pipelineId/intake-form", async (req, res) => { + const pipelineId = req.params.pipelineId as string; + const access = await assertPipelineAccess(pipelineId, req); + if (!access.pipeline) { + res.status(404).json({ error: "Pipeline not found" }); + return; + } + const pipeline = await svc.getPipelineWithRelations(pipelineId); + const firstStage = pipeline?.stages?.[0] ?? null; + const fields = firstStage ? extractPipelineIntakeFormFields([firstStage]) : []; + res.json({ + pipelineId, + stageId: firstStage?.id ?? null, + fields, + }); + }); + + router.get("/companies/:companyId/pipelines-attention", async (req, res) => { + const companyId = req.params.companyId as string; + assertCompanyAccess(req, companyId); + const rows = await svc.listPipelinesAttention(companyId); + res.json(rows); + }); + + // Optional compatibility for pipeline issue conversion. + router.post("/cases/:caseId/open-conversation", validate(openConversationPayloadSchema), async (req, res) => { + const caseId = req.params.caseId as string; + const caseAccess = await assertCaseAccess(caseId, req); + if (!caseAccess.pipelineCase || !caseAccess.pipeline) { + res.status(404).json({ error: "Case not found" }); + return; + } + const payload = req.body; + const issuePayload = { + title: payload.title ?? caseAccess.pipelineCase.title, + description: payload.description ?? null, + status: payload.status, + projectId: payload.projectId ?? null, + assigneeAgentId: payload.assigneeAgentId ?? null, + assigneeUserId: payload.assigneeUserId ?? null, + }; + const issue = await issueService(db).create(caseAccess.pipeline.companyId, { + ...issuePayload, + }); + + const link = await svc.linkCaseIssue(caseId, issue.id, payload.role); + + const actorInfo = getActorInfo(req); + await logActivity(db, { + companyId: caseAccess.pipeline.companyId, + actorType: actorInfo.actorType, + actorId: actorInfo.actorId, + agentId: actorInfo.agentId, + action: "pipeline_case.open_conversation", + entityType: "issue", + entityId: issue.id, + details: { caseId }, + }); + + res.status(201).json({ + ...link, + issueId: issue.id, + caseId, + }); + }); + + return router; +} diff --git a/server/src/routes/routines.ts b/server/src/routes/routines.ts index 48baca8d3c0..67c209954e4 100644 --- a/server/src/routes/routines.ts +++ b/server/src/routes/routines.ts @@ -114,6 +114,16 @@ export function routineRoutes( return { payload, control }; } + function pickRoutineValueFields(input: Record) { + const picked: Record = {}; + for (const [key, value] of Object.entries(input)) { + if (typeof value === "string" || typeof value === "number" || typeof value === "boolean") { + picked[key] = value; + } + } + return picked; + } + router.get("/companies/:companyId/routines", async (req, res) => { const companyId = req.params.companyId as string; assertCompanyAccess(req, companyId); @@ -370,12 +380,14 @@ export function routineRoutes( const caseFields = Object.hasOwn(control, "caseFields") ? (control.caseFields as Record | null) : payload; + const variablePayload = pickRoutineValueFields(payload); + const normalizedCaseFields = caseFields ? pickRoutineValueFields(caseFields) : null; const run = await svc.runRoutine(routine.id, { ...control, source: "api", payload: runPayload, - variables: payload, - caseFields, + variables: variablePayload, + caseFields: normalizedCaseFields, }, { agentId: req.actor.type === "agent" ? req.actor.agentId : null, userId: req.actor.type === "board" ? req.actor.userId ?? null : null, diff --git a/server/src/services/index.ts b/server/src/services/index.ts index 7affa2e9a52..1a317f46236 100644 --- a/server/src/services/index.ts +++ b/server/src/services/index.ts @@ -63,6 +63,7 @@ export type { export { boardAuthService } from "./board-auth.js"; export { instanceSettingsService } from "./instance-settings.js"; export { cloudUpstreamService, reconcileCloudUpstreamRunsOnStartup } from "./cloud-upstreams.js"; +export { pipelineService } from "./pipelines.js"; export { companyPortabilityService } from "./company-portability.js"; export { teamsCatalogService } from "./teams-catalog.js"; export { environmentService } from "./environments.js"; diff --git a/server/src/services/pipelines.test.ts b/server/src/services/pipelines.test.ts new file mode 100644 index 00000000000..81a5624ccd6 --- /dev/null +++ b/server/src/services/pipelines.test.ts @@ -0,0 +1,127 @@ +import { describe, expect, it, vi } from "vitest"; +import { extractPipelineIntakeFormFields, mapConfiguredFields, triggerStageOnEnter, validateStageRequiredFields } from "./pipelines.js"; + +describe("pipeline service helpers", () => { + it("derives intake fields from first stage variables", () => { + const stageRows = [ + { + pipelineId: "pipeline-1", + config: { + variables: [ + { key: "customer", label: "Customer", type: "text", required: true, options: [], showInAddForm: false }, + { key: "urgency", label: "Urgency", type: "select", options: ["low", "high"], required: false, showInAddForm: false }, + ], + }, + }, + { + pipelineId: "pipeline-1", + config: { + variables: [ + { key: "ignored", label: "Ignored", type: "text", required: true, options: [], showInAddForm: false }, + ], + }, + }, + ] as Array<{ + id: string; + pipelineId: string; + name: string; + kind: string; + position: number; + config: { + variables: Array<{ + key: string; + label: string; + type: "text" | "select" | "multiline"; + required: boolean; + options: string[]; + showInAddForm: boolean; + }>; + }; + createdAt: Date; + updatedAt: Date; + }>; + + expect(extractPipelineIntakeFormFields(stageRows)).toEqual([ + { key: "customer", label: "Customer", type: "text", required: true, options: [] }, + { key: "urgency", label: "Urgency", type: "select", required: false, options: ["low", "high"] }, + ]); + }); + + it("validates required intake fields for pipeline stage variables", () => { + const variables = [ + { key: "customer", label: "Customer", type: "text" as const, required: true, options: [], showInAddForm: false }, + { key: "notes", label: "Notes", type: "multiline" as const, required: false, options: [], showInAddForm: false }, + ]; + expect(() => + validateStageRequiredFields({ customer: "Acme" }, variables), + ).not.toThrow(); + expect(() => + validateStageRequiredFields({ notes: "Urgent follow-up" }, variables), + ).toThrowError(/Missing required field\(s\): customer/); + }); + + it("maps configured fields and triggers run_routine with resolved payload and actor", async () => { + const runRoutine = vi.fn().mockResolvedValue({ id: "run-1" }); + const routineSvc = { runRoutine }; + const caseRow = { + id: "case-1", + title: "Follow up issue", + pipelineId: "pipeline-1", + } as { id: string; title: string; pipelineId: string }; + const stage = { + id: "stage-1", + pipelineId: "pipeline-1", + config: { + variables: [], + onEnter: { + run_routine: { + routineId: "11111111-1111-4111-8111-111111111111", + variables: { urgency: "priority" }, + payload: { source: "pipeline" }, + caseFields: { customer: "customer_name" }, + }, + }, + }, + } as { + id: string; + pipelineId: string; + config: { variables: unknown[]; onEnter: { run_routine: Record } }; + }; + const fields = { + customer_name: "Acme", + priority: "high", + notes: "Escalate quickly", + }; + + await triggerStageOnEnter(null as any, routineSvc as never, stage as never, caseRow as never, fields, { + actorId: "agent-1", + actorType: "agent", + }); + + expect(runRoutine).toHaveBeenCalledWith( + "11111111-1111-4111-8111-111111111111", + { + source: "api", + payload: { + source: "pipeline", + pipelineId: "pipeline-1", + caseId: "case-1", + caseTitle: "Follow up issue", + customer: "Acme", + }, + variables: { urgency: "high" }, + caseFields: { customer: "Acme" }, + }, + { + agentId: "agent-1", + userId: null, + }, + ); + }); + + it("maps configured fields without unknown entries", () => { + expect(mapConfiguredFields({ destination: "source" }, { source: "value", other: "ignore" })).toEqual({ + destination: "value", + }); + }); +}); diff --git a/server/src/services/pipelines.ts b/server/src/services/pipelines.ts new file mode 100644 index 00000000000..3f8800b9098 --- /dev/null +++ b/server/src/services/pipelines.ts @@ -0,0 +1,1237 @@ +import { and, asc, count, desc, eq, ilike, inArray, max, ne, sql } from "drizzle-orm"; +import type { Db } from "@paperclipai/db"; +import { + issues, + pipelineCaseBlockers, + pipelineCaseEvents, + pipelineCaseIssueLinks, + pipelineCases, + pipelineGuidanceDocuments, + pipelineStages, + pipelineTransitions, + pipelines, +} from "@paperclipai/db"; +import { + pipelineStageRunRoutineSchema, + type ListCompanyCaseEventsQuery, + type PipelineStageConfig, + type PipelineStageRunRoutineConfig, + type PipelineStageVariable, + type ReviewCasesQuery, +} from "@paperclipai/shared"; +import { conflict, notFound, unprocessable } from "../errors.js"; +import { routineService } from "./routines.js"; + +type PipelineRow = typeof pipelines.$inferSelect; +type PipelineStageRow = typeof pipelineStages.$inferSelect; +type PipelineTransitionInsert = typeof pipelineTransitions.$inferInsert; +type PipelineTransitionSelect = typeof pipelineTransitions.$inferSelect; +type PipelineCaseRow = typeof pipelineCases.$inferSelect; +type PipelineCaseInsert = typeof pipelineCases.$inferInsert; +type PipelineCaseEventInsert = typeof pipelineCaseEvents.$inferInsert; +type PipelineCaseEvent = typeof pipelineCaseEvents.$inferSelect; +type PipelineCaseIssueLink = typeof pipelineCaseIssueLinks.$inferSelect; +type PipelineGuidanceDocument = typeof pipelineGuidanceDocuments.$inferSelect; +type PipelineCaseBlocker = typeof pipelineCaseBlockers.$inferSelect; +type PipelineDb = Db | Parameters[0]>[0]; +type RoutineVariableValue = string | number | boolean; + +type PipelineInput = Omit; +type PipelineStageInput = Omit; +type PipelineCaseActor = { + actorId: string; + actorType: "user" | "agent" | "system"; +}; +type PipelineCaseTransitionOptions = PipelineCaseActor & { + reason?: string | null; +}; + +export type PipelineCaseEventWithContext = PipelineCaseEvent & { + caseTitle: string | null; + pipelineName: string | null; + fromStageName: string | null; + toStageName: string | null; +}; + +export type PipelineIntakeField = { + key: string; + label: string; + type: PipelineStageVariable["type"]; + required?: boolean; + options?: string[]; +}; + +function isNonNullableObject(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + +function normalizeStageConfig(config: PipelineStageConfig | unknown): PipelineStageConfig { + if (!isNonNullableObject(config)) { + return { variables: [] }; + } + if (!("variables" in config) || !Array.isArray((config as { variables?: unknown }).variables)) { + return { ...config, variables: [] }; + } + return config as PipelineStageConfig; +} + +const companyCaseEventTypeAliases: Record = { + review_decided: ["review_decided", "case.reviewed"], + transition_forced: ["transition_forced", "case.transitioned"], +}; + +function normalizeCompanyCaseEventTypes(types: string | undefined): string[] { + if (!types) return []; + const requested = types + .split(",") + .map((type) => type.trim()) + .filter(Boolean); + const normalized = requested.flatMap((type) => companyCaseEventTypeAliases[type] ?? [type]); + return Array.from(new Set(normalized)); +} + +function eventPayloadStageIds(payload: Record | null | undefined): string[] { + const ids = [payload?.fromStageId, payload?.toStageId].filter( + (value): value is string => typeof value === "string" && value.length > 0, + ); + return Array.from(new Set(ids)); +} + +export function extractPipelineIntakeFormFields(stages: PipelineStageRow[]): PipelineIntakeField[] { + const first = stages[0]; + if (!first) return []; + const stageConfig = normalizeStageConfig(first.config); + return stageConfig.variables.map((variable) => ({ + key: variable.key, + label: variable.label, + type: variable.type, + required: variable.required, + options: variable.options, + })); +} + +export function validateStageRequiredFields( + fields: Record, + stageVariables: PipelineStageVariable[], +) { + const missing = stageVariables + .filter((variable) => variable.required) + .filter((variable) => { + const value = fields[variable.key]; + if (value == null) return true; + if (typeof value === "string") return value.trim().length === 0; + return false; + }) + .map((variable) => variable.key); + if (missing.length > 0) { + throw unprocessable(`Missing required field(s): ${missing.join(", ")}`); + } +} + +function coerceCaseFields(fields: Record | null | undefined) { + return fields ?? {}; +} + +export function mapConfiguredFields(mapping: Record, fields: Record) { + const mapped: Record = {}; + for (const [toKey, fromKey] of Object.entries(mapping)) { + if (Object.hasOwn(fields, fromKey)) { + mapped[toKey] = fields[fromKey]; + } + } + return mapped; +} + +function isRoutineVariableValue(value: unknown): value is RoutineVariableValue { + return typeof value === "string" || typeof value === "number" || typeof value === "boolean"; +} + +function pickRoutineVariableValues(fields: Record) { + const picked: Record = {}; + for (const [key, value] of Object.entries(fields)) { + if (isRoutineVariableValue(value)) { + picked[key] = value; + } + } + return picked; +} + +function mapConfiguredRoutineValues(mapping: Record, fields: Record) { + const mapped: Record = {}; + for (const [toKey, fromKey] of Object.entries(mapping)) { + const value = fields[fromKey]; + if (isRoutineVariableValue(value)) { + mapped[toKey] = value; + } + } + return mapped; +} + +export function parseStageOnEnterRoutineConfig(config: PipelineStageConfig | unknown): PipelineStageRunRoutineConfig | null { + const stageConfig = normalizeStageConfig(config); + if (!isNonNullableObject(stageConfig.onEnter)) return null; + const raw = (stageConfig.onEnter as { run_routine?: unknown }).run_routine; + if (raw == null || typeof raw !== "object") return null; + const parsed = pipelineStageRunRoutineSchema.safeParse(raw); + if (!parsed.success) return null; + return parsed.data; +} + +function makeRoutineActorInput(actor?: PipelineCaseActor) { + if (!actor) return undefined; + if (actor.actorType === "agent") { + return { agentId: actor.actorId, userId: null }; + } + if (actor.actorType === "system") { + return { agentId: null, userId: null }; + } + return { agentId: null, userId: actor.actorId }; +} + +export async function triggerStageOnEnter( + _db: Db, + svc: ReturnType, + stage: PipelineStageRow, + caseRow: PipelineCaseRow, + fields: Record, + actor?: PipelineCaseActor, +) { + const onEnter = parseStageOnEnterRoutineConfig(stage.config); + if (!onEnter) return; + const mappedPayloadCaseFields = mapConfiguredFields(onEnter.caseFields ?? {}, fields); + const mappedCaseFields = mapConfiguredRoutineValues(onEnter.caseFields ?? {}, fields); + const mappedVariables = mapConfiguredRoutineValues(onEnter.variables, fields); + await svc.runRoutine(onEnter.routineId, { + source: "api", + payload: { + ...onEnter.payload, + pipelineId: stage.pipelineId, + caseId: caseRow.id, + caseTitle: caseRow.title, + ...mappedPayloadCaseFields, + }, + variables: mappedVariables, + caseFields: Object.keys(mappedCaseFields).length > 0 ? mappedCaseFields : pickRoutineVariableValues(fields), + }, makeRoutineActorInput(actor)); +} + +async function resolvePipelineIntakeStage(db: PipelineDb, pipelineId: string, stageId?: string | null) { + if (stageId) { + const row = await db + .select() + .from(pipelineStages) + .where(and(eq(pipelineStages.pipelineId, pipelineId), eq(pipelineStages.id, stageId))) + .then((rows) => rows[0] ?? null); + if (!row) { + throw unprocessable("Stage does not belong to this pipeline"); + } + return row; + } + + const first = await db + .select() + .from(pipelineStages) + .where(eq(pipelineStages.pipelineId, pipelineId)) + .orderBy(asc(pipelineStages.position), asc(pipelineStages.createdAt)) + .limit(1); + if (first.length === 0) { + throw unprocessable("Pipeline has no stages"); + } + const row = first[0]; + if (!row) { + throw unprocessable("Pipeline has no stages"); + } + return row; +} + +function validateStageRequiredFieldsForIngest(stage: PipelineStageRow, fields: Record) { + const config = normalizeStageConfig(stage.config); + const requiredVariables = Array.isArray(config.variables) ? config.variables : []; + validateStageRequiredFields(fields, requiredVariables); +} + +type PipelineTransitionPayload = { + transitions: Array<{ + fromStageId: string; + toStageId: string; + config?: Record; + }>; + enforceTransitions?: boolean; +}; + +type PipelineListOptions = { + q?: string; + includeConnections?: boolean; + includeCounts?: boolean; +}; + +type PipelineCaseListOptions = { + stageId?: string; + status?: string; + q?: string; + limit: number; + offset: number; +}; + +type PipelineWithRelations = PipelineRow & { + stages?: PipelineStageRow[]; + transitions?: PipelineTransitionSelect[]; + guidanceDocuments?: PipelineGuidanceDocument[]; + caseCount?: number; + openCaseCount?: number; + connections?: { + upstreamPipelineIds?: string[]; + downstreamPipelineIds?: string[]; + }; +}; + +const DEFAULT_CASE_LIST_LIMIT = 100; +const MAX_CASE_LIST_LIMIT = 500; + +function parseQueryValue(value: unknown) { + return typeof value === "string" && value.trim().length > 0 ? value.trim() : undefined; +} + +function clampCaseLimit(value: number | undefined) { + if (typeof value !== "number" || !Number.isFinite(value)) return DEFAULT_CASE_LIST_LIMIT; + if (value <= 0) return 1; + if (value > MAX_CASE_LIST_LIMIT) return MAX_CASE_LIST_LIMIT; + return Math.floor(value); +} + +async function findPipeline(db: PipelineDb, pipelineId: string) { + return db + .select() + .from(pipelines) + .where(eq(pipelines.id, pipelineId)) + .then((rows) => rows[0] ?? null); +} + +async function findPipelineCase(db: PipelineDb, caseId: string) { + return db + .select() + .from(pipelineCases) + .where(eq(pipelineCases.id, caseId)) + .then((rows) => rows[0] ?? null); +} + +async function assertStageBelongsToPipeline(db: PipelineDb, pipelineId: string, stageId: string | undefined | null) { + if (!stageId) return; + const row = await db + .select({ id: pipelineStages.id }) + .from(pipelineStages) + .where(and(eq(pipelineStages.id, stageId), eq(pipelineStages.pipelineId, pipelineId))) + .then((rows) => rows[0] ?? null); + if (!row) { + throw unprocessable("Stage does not belong to this pipeline"); + } +} + +async function assertCaseBelongsToPipeline(db: PipelineDb, pipelineId: string, caseId: string | undefined | null) { + if (!caseId) return; + const row = await db + .select({ pipelineId: pipelineCases.pipelineId }) + .from(pipelineCases) + .where(eq(pipelineCases.id, caseId)) + .then((rows) => rows[0] ?? null); + if (!row) { + throw unprocessable("Reference case not found"); + } + if (row.pipelineId !== pipelineId) { + throw unprocessable("Reference case must belong to this pipeline"); + } +} + +async function assertIssueBelongsToPipeline(db: PipelineDb, companyId: string, issueId: string) { + const row = await db + .select({ companyId: issues.companyId }) + .from(issues) + .where(eq(issues.id, issueId)) + .then((rows) => rows[0] ?? null); + if (!row) { + throw unprocessable("Issue not found"); + } + if (row.companyId !== companyId) { + throw unprocessable("Issue must belong to this company"); + } +} + +async function writeCaseEvent(db: PipelineDb, row: PipelineCaseEventInsert) { + const [event] = await db.insert(pipelineCaseEvents).values(row).returning(); + return event; +} + +export function pipelineService(db: Db) { + const routineSvc = routineService(db); + + return { + listPipelines: async ( + companyId: string, + query: PipelineListOptions = {}, + ): Promise => { + const whereClauses = [eq(pipelines.companyId, companyId)] as Parameters[0][]; + const q = parseQueryValue(query.q); + if (q) { + whereClauses.push(ilike(pipelines.name, `%${q}%`)); + } + + const rows = await db + .select() + .from(pipelines) + .where(and(...whereClauses)) + .orderBy(asc(pipelines.name), asc(pipelines.createdAt)); + + if (!query.includeCounts && !query.includeConnections) { + return rows; + } + + if (rows.length === 0) { + return rows as PipelineWithRelations[]; + } + + const pipelineIds = rows.map((row) => row.id); + const [caseCounts, openCaseCounts] = await Promise.all([ + db + .select({ + pipelineId: pipelineCases.pipelineId, + count: count(), + }) + .from(pipelineCases) + .where(inArray(pipelineCases.pipelineId, pipelineIds)) + .groupBy(pipelineCases.pipelineId), + db + .select({ + pipelineId: pipelineCases.pipelineId, + count: count(), + }) + .from(pipelineCases) + .where(and(inArray(pipelineCases.pipelineId, pipelineIds), ne(pipelineCases.status, "done"), ne(pipelineCases.status, "cancelled"))) + .groupBy(pipelineCases.pipelineId), + ]); + + const caseCountByPipeline = new Map(caseCounts.map((row) => [row.pipelineId, Number(row.count)])); + const openCaseCountByPipeline = new Map(openCaseCounts.map((row) => [row.pipelineId, Number(row.count)])); + + return rows.map((pipeline) => ({ + ...pipeline, + caseCount: caseCountByPipeline.get(pipeline.id) ?? 0, + openCaseCount: openCaseCountByPipeline.get(pipeline.id) ?? 0, + ...(query.includeConnections + ? { + connections: { + upstreamPipelineIds: [], + downstreamPipelineIds: [], + }, + } + : {}), + })); + }, + + getPipeline: async (pipelineId: string) => findPipeline(db, pipelineId), + + getPipelineWithRelations: async (pipelineId: string) => { + const pipeline = await findPipeline(db, pipelineId); + if (!pipeline) return null; + const [stages, transitions, guidanceDocuments] = await Promise.all([ + db + .select() + .from(pipelineStages) + .where(eq(pipelineStages.pipelineId, pipelineId)) + .orderBy(asc(pipelineStages.position), asc(pipelineStages.createdAt)), + db + .select() + .from(pipelineTransitions) + .where(eq(pipelineTransitions.pipelineId, pipelineId)) + .orderBy(asc(pipelineTransitions.fromStageId), asc(pipelineTransitions.toStageId)), + db + .select() + .from(pipelineGuidanceDocuments) + .where(eq(pipelineGuidanceDocuments.pipelineId, pipelineId)) + .orderBy(asc(pipelineGuidanceDocuments.key)), + ]); + + return { + ...pipeline, + stages, + transitions, + guidanceDocuments, + } satisfies PipelineWithRelations; + }, + + createPipeline: async (companyId: string, input: PipelineInput) => { + const [pipeline] = await db + .insert(pipelines) + .values({ ...input, companyId }) + .returning(); + if (!pipeline) { + throw conflict("Failed to create pipeline"); + } + return pipeline as PipelineRow; + }, + + updatePipeline: async (pipelineId: string, input: Partial) => { + const [pipeline] = await db + .update(pipelines) + .set({ ...input, updatedAt: new Date() }) + .where(eq(pipelines.id, pipelineId)) + .returning(); + return pipeline ? pipeline as PipelineRow : null; + }, + + removePipeline: async (pipelineId: string) => { + const [pipeline] = await db + .delete(pipelines) + .where(eq(pipelines.id, pipelineId)) + .returning(); + return pipeline ? pipeline as PipelineRow : null; + }, + + listPipelineStages: async (pipelineId: string) => + db + .select() + .from(pipelineStages) + .where(eq(pipelineStages.pipelineId, pipelineId)) + .orderBy(asc(pipelineStages.position), asc(pipelineStages.createdAt)), + + createPipelineStage: async (pipelineId: string, input: PipelineStageInput) => { + const pipeline = await findPipeline(db, pipelineId); + if (!pipeline) { + throw notFound("Pipeline not found"); + } + + const nextPos = input.position !== undefined + ? input.position + : await db + .select({ maxPos: max(pipelineStages.position) }) + .from(pipelineStages) + .where(eq(pipelineStages.pipelineId, pipelineId)) + .then((rows) => { + const maxPos = rows[0]?.maxPos; + if (typeof maxPos === "number") { + return maxPos + 1; + } + return 0; + }); + + try { + const [stage] = await db + .insert(pipelineStages) + .values({ + ...input, + pipelineId, + position: nextPos, + config: input.config ?? { variables: [] }, + }) + .returning(); + return stage as PipelineStageRow; + } catch (err) { + if ((err as { code?: unknown }).code === "23505") { + throw conflict("Pipeline stage with this position already exists"); + } + throw err; + } + }, + + updatePipelineStage: async (pipelineId: string, stageId: string, input: Partial) => { + if (Object.keys(input).length === 0) { + return await db + .select() + .from(pipelineStages) + .where(and(eq(pipelineStages.id, stageId), eq(pipelineStages.pipelineId, pipelineId))) + .then((rows) => rows[0] ?? null); + } + + try { + const [stage] = await db + .update(pipelineStages) + .set({ ...input, updatedAt: new Date() }) + .where(and(eq(pipelineStages.id, stageId), eq(pipelineStages.pipelineId, pipelineId))) + .returning(); + return stage ? stage as PipelineStageRow : null; + } catch (err) { + if ((err as { code?: unknown }).code === "23505") { + throw conflict("Pipeline stage position is already in use"); + } + throw err; + } + }, + + removePipelineStage: async (pipelineId: string, stageId: string) => { + const [removed] = await db + .delete(pipelineStages) + .where(and(eq(pipelineStages.id, stageId), eq(pipelineStages.pipelineId, pipelineId))) + .returning(); + return removed ? removed as PipelineStageRow : null; + }, + + listPipelineTransitions: async (pipelineId: string) => + db + .select() + .from(pipelineTransitions) + .where(eq(pipelineTransitions.pipelineId, pipelineId)) + .orderBy(asc(pipelineTransitions.fromStageId), asc(pipelineTransitions.toStageId)), + + setPipelineTransitions: async (pipelineId: string, payload: PipelineTransitionPayload) => { + const pipeline = await findPipeline(db, pipelineId); + if (!pipeline) { + throw notFound("Pipeline not found"); + } + + const enforceTransitions = payload.enforceTransitions !== false; + const transitions = payload.transitions.map((transition) => ({ + ...transition, + pipelineId, + config: transition.config ?? {}, + })) as PipelineTransitionInsert[]; + + if (enforceTransitions && transitions.length > 0) { + const stageIds = Array.from(new Set(transitions.flatMap((transition) => [transition.fromStageId, transition.toStageId]))); + if (stageIds.length > 0) { + const known = await db + .select({ id: pipelineStages.id }) + .from(pipelineStages) + .where(and(eq(pipelineStages.pipelineId, pipelineId), inArray(pipelineStages.id, stageIds))); + if (known.length !== stageIds.length) { + throw unprocessable("Transition stage references must belong to this pipeline"); + } + } + } + + return db.transaction(async (tx) => { + await tx.delete(pipelineTransitions).where(eq(pipelineTransitions.pipelineId, pipelineId)); + if (transitions.length === 0) { + return []; + } + return tx + .insert(pipelineTransitions) + .values(transitions) + .returning() + .then((rows) => rows as PipelineTransitionSelect[]); + }); + }, + + listPipelineCases: async ( + pipelineId: string, + query: PipelineCaseListOptions, + ): Promise => { + const queryPipeline = await findPipeline(db, pipelineId); + if (!queryPipeline) { + throw notFound("Pipeline not found"); + } + const where = [eq(pipelineCases.pipelineId, queryPipeline.id)] as Parameters[0][]; + const stageId = parseQueryValue(query.stageId); + const status = parseQueryValue(query.status); + const q = parseQueryValue(query.q); + if (stageId) { + where.push(eq(pipelineCases.stageId, stageId)); + } + if (status) { + where.push(eq(pipelineCases.status, status)); + } + if (q) { + where.push(ilike(pipelineCases.title, `%${q}%`)); + } + + const limit = clampCaseLimit(query.limit); + const offset = Math.max(0, Number.isFinite(query.offset) ? Math.floor(query.offset) : 0); + + return db + .select() + .from(pipelineCases) + .where(and(...where)) + .orderBy(desc(sql`coalesce(${pipelineCases.lastActivityAt}, ${pipelineCases.updatedAt}, ${pipelineCases.createdAt})`), asc(pipelineCases.id)) + .limit(limit) + .offset(offset); + }, + + getPipelineCase: async (caseId: string) => findPipelineCase(db, caseId), + + ingestPipelineCase: async ( + pipelineId: string, + input: Omit, + actor?: PipelineCaseActor, + ) => { + const stage = await resolvePipelineIntakeStage(db, pipelineId, input.stageId); + validateStageRequiredFieldsForIngest(stage, coerceCaseFields(input.fields)); + const [createdCase] = await db.transaction(async (tx) => { + await assertCaseBelongsToPipeline(tx, pipelineId, input.parentCaseId); + const pipeline = await findPipeline(tx, pipelineId); + if (!pipeline) { + throw notFound("Pipeline not found"); + } + + const [created] = await tx + .insert(pipelineCases) + .values({ + ...input, + companyId: pipeline.companyId, + pipelineId, + status: input.status ?? "open", + stageId: stage.id, + fields: coerceCaseFields(input.fields), + createdByActorId: actor?.actorId, + createdByActorType: actor?.actorType, + lastActivityAt: new Date(), + }) + .returning(); + if (!created) { + throw conflict("Failed to create case"); + } + + await writeCaseEvent(tx, { + companyId: pipeline.companyId, + pipelineId, + caseId: created.id, + kind: "case.ingested", + summary: `Case ingested: ${created.title}`, + payload: { + status: created.status, + }, + }); + + return [created] as [PipelineCaseRow]; + }); + await triggerStageOnEnter( + db, + routineSvc, + stage, + createdCase, + coerceCaseFields(input.fields), + actor, + ); + return createdCase; + }, + + ingestPipelineCases: async ( + pipelineId: string, + items: Array<{ + title: string; + status?: string; + stageId?: string | null; + parentCaseId?: string | null; + fields?: Record; + }>, + actor?: PipelineCaseActor, + ) => { + const pipeline = await findPipeline(db, pipelineId); + if (!pipeline) { + throw notFound("Pipeline not found"); + } + let defaultStage: PipelineStageRow | null | undefined; + const created = await db.transaction(async (tx) => { + const created: Array<{ caseRow: PipelineCaseRow; stage: PipelineStageRow; fields: Record }> = []; + for (const item of items) { + const stage = item.stageId + ? await resolvePipelineIntakeStage(tx, pipelineId, item.stageId) + : defaultStage ?? (defaultStage = await resolvePipelineIntakeStage(tx, pipelineId)); + const fields = coerceCaseFields(item.fields); + validateStageRequiredFieldsForIngest(stage, fields); + const next: PipelineCaseInsert = { + ...item, + companyId: pipeline.companyId, + pipelineId, + stageId: stage.id, + status: item.status ?? "open", + fields, + createdByActorId: actor?.actorId, + createdByActorType: actor?.actorType, + lastActivityAt: new Date(), + } as PipelineCaseInsert; + await assertCaseBelongsToPipeline(tx, pipelineId, item.parentCaseId ?? null); + const [createdCase] = await tx + .insert(pipelineCases) + .values(next) + .returning(); + if (!createdCase) { + throw conflict("Failed to create case"); + } + await writeCaseEvent(tx, { + companyId: pipeline.companyId, + pipelineId, + caseId: createdCase.id, + kind: "case.ingested", + summary: `Case ingested: ${createdCase.title}`, + payload: { + status: createdCase.status, + }, + }); + created.push({ + caseRow: createdCase, + stage, + fields, + }); + } + return created; + }); + for (const { caseRow, stage, fields } of created) { + await triggerStageOnEnter( + db, + routineSvc, + stage, + caseRow, + fields, + actor, + ); + } + return created.map((entry) => entry.caseRow); + }, + + updatePipelineCase: async ( + caseId: string, + input: Partial>, + ) => { + const current = await findPipelineCase(db, caseId); + if (!current) return null; + await assertStageBelongsToPipeline(db, current.pipelineId, input.stageId); + const [updated] = await db + .update(pipelineCases) + .set({ ...input, lastActivityAt: new Date(), updatedAt: new Date() }) + .where(eq(pipelineCases.id, caseId)) + .returning(); + if (!updated) return null; + await writeCaseEvent(db, { + companyId: current.companyId, + pipelineId: current.pipelineId, + caseId: current.id, + kind: "case.updated", + summary: `Case updated: ${current.title}`, + payload: input, + }); + return updated as PipelineCaseRow; + }, + + transitionCaseStage: async (caseId: string, toStageId: string, options?: PipelineCaseTransitionOptions) => { + const current = await findPipelineCase(db, caseId); + if (!current) return null; + await assertStageBelongsToPipeline(db, current.pipelineId, toStageId); + if (current.stageId === toStageId) { + return current; + } + const nextStage = await db + .select() + .from(pipelineStages) + .where(and(eq(pipelineStages.pipelineId, current.pipelineId), eq(pipelineStages.id, toStageId))) + .then((rows) => rows[0] ?? null); + if (!nextStage) { + throw notFound("Stage not found"); + } + const [updated] = await db + .update(pipelineCases) + .set({ stageId: toStageId, lastActivityAt: new Date(), updatedAt: new Date() }) + .where(eq(pipelineCases.id, caseId)) + .returning(); + if (!updated) return null; + await writeCaseEvent(db, { + companyId: current.companyId, + pipelineId: current.pipelineId, + caseId, + kind: "case.transitioned", + summary: `Case transitioned: ${current.title}`, + payload: { + fromStageId: current.stageId, + toStageId, + reason: options?.reason ?? null, + }, + }); + await triggerStageOnEnter( + db, + routineSvc, + nextStage, + updated, + coerceCaseFields(updated.fields as Record | null | undefined), + options, + ); + return updated as PipelineCaseRow; + }, + + createCaseBlocker: async (caseId: string, blockedByCaseId: string, reason?: string | null) => { + const current = await findPipelineCase(db, caseId); + if (!current) return null; + if (caseId === blockedByCaseId) { + throw unprocessable("A case cannot block itself"); + } + const blockerCase = await findPipelineCase(db, blockedByCaseId); + if (!blockerCase) { + throw unprocessable("Blocked-by case not found"); + } + if (blockerCase.pipelineId !== current.pipelineId) { + throw unprocessable("Blocked-by case must belong to this pipeline"); + } + try { + const [created] = await db + .insert(pipelineCaseBlockers) + .values({ + caseId, + blockedByCaseId, + reason: reason ?? null, + resolved: false, + }) + .returning(); + return created as PipelineCaseBlocker; + } catch (err) { + if ((err as { code?: unknown }).code === "23505") { + throw conflict("Blocker already exists"); + } + throw err; + } + }, + + listCaseBlockers: async (caseId: string) => { + return db + .select() + .from(pipelineCaseBlockers) + .where(eq(pipelineCaseBlockers.caseId, caseId)) + .orderBy(asc(pipelineCaseBlockers.createdAt)); + }, + + resolveCaseBlocker: async (caseId: string, blockerId: string) => { + const [updated] = await db + .update(pipelineCaseBlockers) + .set({ resolved: true, resolvedAt: new Date() }) + .where(and(eq(pipelineCaseBlockers.id, blockerId), eq(pipelineCaseBlockers.caseId, caseId))) + .returning(); + return updated ? updated as PipelineCaseBlocker : null; + }, + + listGuidanceDocuments: async (pipelineId: string) => + db + .select() + .from(pipelineGuidanceDocuments) + .where(eq(pipelineGuidanceDocuments.pipelineId, pipelineId)) + .orderBy(asc(pipelineGuidanceDocuments.key)), + + getGuidanceDocument: async (pipelineId: string, key: string) => { + return db + .select() + .from(pipelineGuidanceDocuments) + .where(and(eq(pipelineGuidanceDocuments.pipelineId, pipelineId), eq(pipelineGuidanceDocuments.key, key))) + .then((rows) => rows[0] ?? null); + }, + + upsertGuidanceDocument: async ( + pipelineId: string, + key: string, + input: Pick, + ) => { + const existing = await db + .select({ id: pipelineGuidanceDocuments.id }) + .from(pipelineGuidanceDocuments) + .where(and(eq(pipelineGuidanceDocuments.pipelineId, pipelineId), eq(pipelineGuidanceDocuments.key, key))) + .then((rows) => rows[0] ?? null); + if (existing) { + const [updated] = await db + .update(pipelineGuidanceDocuments) + .set({ + title: input.title, + body: input.body, + updatedAt: new Date(), + }) + .where(and(eq(pipelineGuidanceDocuments.pipelineId, pipelineId), eq(pipelineGuidanceDocuments.key, key))) + .returning(); + return updated as PipelineGuidanceDocument; + } + const [created] = await db + .insert(pipelineGuidanceDocuments) + .values({ + pipelineId, + key, + title: input.title, + body: input.body, + }) + .returning(); + return created as PipelineGuidanceDocument; + }, + + deleteGuidanceDocument: async (pipelineId: string, key: string) => { + const [deleted] = await db + .delete(pipelineGuidanceDocuments) + .where(and(eq(pipelineGuidanceDocuments.pipelineId, pipelineId), eq(pipelineGuidanceDocuments.key, key))) + .returning(); + return deleted ? deleted as PipelineGuidanceDocument : null; + }, + + linkCaseIssue: async (caseId: string, issueId: string, role: string) => { + const current = await findPipelineCase(db, caseId); + if (!current) return null; + await assertIssueBelongsToPipeline(db, current.companyId, issueId); + try { + const [created] = await db + .insert(pipelineCaseIssueLinks) + .values({ + caseId, + issueId, + role, + }) + .returning(); + return created as PipelineCaseIssueLink; + } catch (err) { + if ((err as { code?: unknown }).code === "23505") { + throw conflict("Case issue link already exists"); + } + throw err; + } + }, + + listCaseIssueLinks: async (caseId: string) => { + return db + .select() + .from(pipelineCaseIssueLinks) + .where(eq(pipelineCaseIssueLinks.caseId, caseId)) + .orderBy(asc(pipelineCaseIssueLinks.createdAt)); + }, + + unlinkCaseIssue: async (linkId: string, caseId: string) => { + const [deleted] = await db + .delete(pipelineCaseIssueLinks) + .where(and(eq(pipelineCaseIssueLinks.id, linkId), eq(pipelineCaseIssueLinks.caseId, caseId))) + .returning(); + return deleted ? deleted as PipelineCaseIssueLink : null; + }, + + listCaseEvents: async (caseId: string): Promise => { + const current = await findPipelineCase(db, caseId); + if (!current) return []; + return db + .select() + .from(pipelineCaseEvents) + .where(eq(pipelineCaseEvents.caseId, current.id)) + .orderBy(desc(pipelineCaseEvents.createdAt)); + }, + + listPipelineEvents: async (pipelineId: string): Promise => { + return db + .select() + .from(pipelineCaseEvents) + .where(eq(pipelineCaseEvents.pipelineId, pipelineId)) + .orderBy(desc(pipelineCaseEvents.createdAt)); + }, + + listCompanyCaseEvents: async ( + companyId: string, + query: ListCompanyCaseEventsQuery, + ): Promise => { + const conditions = [eq(pipelineCaseEvents.companyId, companyId)]; + const types = normalizeCompanyCaseEventTypes(query.types); + if (types.length > 0) { + conditions.push(inArray(pipelineCaseEvents.kind, types)); + } + if (query.pipelineId) { + conditions.push(eq(pipelineCaseEvents.pipelineId, query.pipelineId)); + } + + const rows = await db + .select({ + event: pipelineCaseEvents, + caseTitle: pipelineCases.title, + pipelineName: pipelines.name, + }) + .from(pipelineCaseEvents) + .leftJoin(pipelineCases, eq(pipelineCaseEvents.caseId, pipelineCases.id)) + .leftJoin(pipelines, eq(pipelineCaseEvents.pipelineId, pipelines.id)) + .where(and(...conditions)) + .orderBy(desc(pipelineCaseEvents.createdAt)) + .limit(query.limit) + .offset(query.offset); + + const stageIds = Array.from( + new Set(rows.flatMap((row) => eventPayloadStageIds(row.event.payload))), + ); + const stageRows = stageIds.length > 0 + ? await db + .select({ id: pipelineStages.id, name: pipelineStages.name }) + .from(pipelineStages) + .where(inArray(pipelineStages.id, stageIds)) + : []; + const stageNameById = new Map(stageRows.map((stage) => [stage.id, stage.name])); + + return rows.map((row) => { + const payload = row.event.payload; + const fromStageId = typeof payload?.fromStageId === "string" ? payload.fromStageId : null; + const toStageId = typeof payload?.toStageId === "string" ? payload.toStageId : null; + return { + ...row.event, + caseTitle: row.caseTitle, + pipelineName: row.pipelineName, + fromStageName: fromStageId ? stageNameById.get(fromStageId) ?? null : null, + toStageName: toStageId ? stageNameById.get(toStageId) ?? null : null, + }; + }); + }, + + listCaseChildren: async (caseId: string) => { + return db + .select() + .from(pipelineCases) + .where(eq(pipelineCases.parentCaseId, caseId)) + .orderBy(asc(pipelineCases.createdAt)); + }, + + suggestTransition: async (caseId: string, toStageId: string, reason?: string | null) => { + const current = await findPipelineCase(db, caseId); + if (!current) return null; + await assertStageBelongsToPipeline(db, current.pipelineId, toStageId); + const fields = { ...(current.fields ?? {}) } as Record; + fields.nextSuggestedStageId = toStageId; + if (reason) { + fields.nextSuggestedStageReason = reason; + } + const [updated] = await db + .update(pipelineCases) + .set({ + fields, + lastActivityAt: new Date(), + updatedAt: new Date(), + }) + .where(eq(pipelineCases.id, caseId)) + .returning(); + if (!updated) return null; + await writeCaseEvent(db, { + companyId: current.companyId, + pipelineId: current.pipelineId, + caseId, + kind: "case.suggested", + summary: `Case suggestion added for ${current.title}`, + payload: { toStageId, reason }, + }); + return updated; + }, + + resolveSuggestion: async (caseId: string, decision: "accept" | "decline", note?: string | null) => { + const current = await findPipelineCase(db, caseId); + if (!current) return null; + const fields = { ...(current.fields ?? {}) } as Record; + fields.suggestionResolution = { + decision, + note, + resolvedAt: new Date().toISOString(), + }; + const [updated] = await db + .update(pipelineCases) + .set({ + fields, + lastActivityAt: new Date(), + updatedAt: new Date(), + }) + .where(eq(pipelineCases.id, caseId)) + .returning(); + if (!updated) return null; + await writeCaseEvent(db, { + companyId: current.companyId, + pipelineId: current.pipelineId, + caseId, + kind: "case.suggestion_resolved", + summary: `Case suggestion ${decision}`, + payload: { decision, note }, + }); + return updated; + }, + + reviewCase: async ( + caseId: string, + decision: "approve" | "request_changes" | "drop", + note?: string | null, + ) => { + const current = await findPipelineCase(db, caseId); + if (!current) return null; + + const nextStatus = decision === "approve" ? "done" : decision === "drop" ? "cancelled" : current.status; + const payload = { ...(current.fields ?? {}) } as Record; + payload.review = { + decision, + note, + reviewedAt: new Date().toISOString(), + }; + const [updated] = await db + .update(pipelineCases) + .set({ + status: nextStatus, + fields: payload, + lastActivityAt: new Date(), + updatedAt: new Date(), + }) + .where(eq(pipelineCases.id, caseId)) + .returning(); + if (!updated) return null; + await writeCaseEvent(db, { + companyId: current.companyId, + pipelineId: current.pipelineId, + caseId, + kind: "case.reviewed", + summary: `Case ${decision}`, + payload: { decision, note }, + }); + return updated; + }, + + listReviewCases: async (companyId: string, query: ReviewCasesQuery) => { + const filters = [ + eq(pipelineCases.companyId, companyId), + ne(pipelineCases.status, "done"), + ne(pipelineCases.status, "cancelled"), + ]; + if (query.pipelineId) { + filters.push(eq(pipelineCases.pipelineId, query.pipelineId)); + } + filters.push(eq(pipelineStages.kind, query.kind ?? "review")); + + return db + .select({ + id: pipelineCases.id, + companyId: pipelineCases.companyId, + pipelineId: pipelineCases.pipelineId, + stageId: pipelineCases.stageId, + parentCaseId: pipelineCases.parentCaseId, + title: pipelineCases.title, + status: pipelineCases.status, + fields: pipelineCases.fields, + createdByActorId: pipelineCases.createdByActorId, + createdByActorType: pipelineCases.createdByActorType, + lastActivityAt: pipelineCases.lastActivityAt, + createdAt: pipelineCases.createdAt, + updatedAt: pipelineCases.updatedAt, + }) + .from(pipelineCases) + .innerJoin(pipelineStages, eq(pipelineCases.stageId, pipelineStages.id)) + .where(and(...filters)) + .orderBy(desc(sql`coalesce(${pipelineCases.lastActivityAt}, ${pipelineCases.updatedAt}, ${pipelineCases.createdAt})`)) + .limit(query.limit) + .offset(query.offset); + }, + + bulkReviewCases: async ( + companyId: string, + caseIds: string[], + decision: "approve" | "request_changes" | "drop", + note?: string | null, + ) => { + const rows = await db + .select({ id: pipelineCases.id }) + .from(pipelineCases) + .where(and(eq(pipelineCases.companyId, companyId), inArray(pipelineCases.id, caseIds))); + + const updated: PipelineCaseRow[] = []; + for (const row of rows) { + const reviewed = await pipelineService(db).reviewCase(row.id, decision, note); + if (reviewed) updated.push(reviewed); + } + return updated; + }, + + listPipelinesAttention: async (companyId: string) => { + const attention = await db + .select({ + pipelineId: pipelineCases.pipelineId, + id: pipelineCases.id, + title: pipelineCases.title, + }) + .from(pipelineCases) + .where(and( + eq(pipelineCases.companyId, companyId), + ne(pipelineCases.status, "done"), + ne(pipelineCases.status, "cancelled"), + )) + .orderBy(desc(sql`coalesce(${pipelineCases.lastActivityAt}, ${pipelineCases.updatedAt}, ${pipelineCases.createdAt})`)); + return attention; + }, + }; +} diff --git a/ui/src/App.tsx b/ui/src/App.tsx index 52c4a7f2781..8d005320da4 100644 --- a/ui/src/App.tsx +++ b/ui/src/App.tsx @@ -13,6 +13,14 @@ import { Projects } from "./pages/Projects"; import { ProjectDetail } from "./pages/ProjectDetail"; import { ProjectWorkspaceDetail } from "./pages/ProjectWorkspaceDetail"; import { Workspaces } from "./pages/Workspaces"; +import { + LearningsStub, + PipelineBoardStub, + PipelineItemStub, + Pipelines, + PipelineSettingsStub, + ReviewQueueStub, +} from "./pages/Pipelines"; import { Issues } from "./pages/Issues"; import { Search } from "./pages/Search"; import { IssueDetail } from "./pages/IssueDetail"; @@ -117,6 +125,12 @@ function boardRoutes() { } /> } /> } /> + } /> + } /> + } /> + } /> + } /> + } /> } /> } /> } /> diff --git a/ui/src/api/pipelines.ts b/ui/src/api/pipelines.ts index d14afb35839..6c4945b5c5b 100644 --- a/ui/src/api/pipelines.ts +++ b/ui/src/api/pipelines.ts @@ -77,6 +77,7 @@ export interface PipelineCase { id: string; pipelineId: string; stageId: string | null; + parentCaseId?: string | null; title: string; activeWork?: boolean | number | null; status?: string | null; diff --git a/ui/src/components/Sidebar.tsx b/ui/src/components/Sidebar.tsx index 592888ad286..4f0ca081119 100644 --- a/ui/src/components/Sidebar.tsx +++ b/ui/src/components/Sidebar.tsx @@ -6,6 +6,9 @@ import { DollarSign, History, Search, + ListChecks, + FolderKanban, + BookOpenText, SquarePen, Network, Boxes, @@ -164,6 +167,9 @@ export function Sidebar() { + + + diff --git a/ui/src/lib/pipeline-item-detail.ts b/ui/src/lib/pipeline-item-detail.ts new file mode 100644 index 00000000000..02c60f13854 --- /dev/null +++ b/ui/src/lib/pipeline-item-detail.ts @@ -0,0 +1,211 @@ +import type { PipelineCase, PipelineCaseEvent, PipelineStage } from "../api/pipelines"; + +export type PendingTransitionBannerState = + | { visible: false } + | { + visible: true; + toStageId: string; + toStageName: string; + reason: string | null; + }; + +export type PipelineItemEventPresentation = { + sentence: string; + tone: "neutral" | "positive" | "attention"; +}; + +export type PipelineItemFieldEntry = { + key: string; + label: string; + value: string; +}; + +const INTERNAL_FIELD_KEYS = new Set([ + "activeWork", + "blockedByCaseIds", + "childrenSummary", + "nextSuggestedStageId", + "nextSuggestedStageName", + "nextSuggestedStageReason", + "review", + "suggestionResolution", + "thisChanged", + "upstreamChanged", + "upstreamDrift", + "changeAcknowledgedAt", +]); + +function asRecord(value: unknown): Record { + return typeof value === "object" && value !== null && !Array.isArray(value) + ? value as Record + : {}; +} + +function asString(value: unknown): string | null { + return typeof value === "string" && value.trim().length > 0 ? value.trim() : null; +} + +function stageNameById(stages: PipelineStage[]): Map { + return new Map(stages.map((stage) => [stage.id, stage.name])); +} + +function payloadText(event: PipelineCaseEvent, ...keys: string[]): string | null { + const payload = asRecord(event.payload); + for (const key of keys) { + const value = asString(payload[key]); + if (value) return value; + } + return null; +} + +function sentenceSubject(event: PipelineCaseEvent): string { + return event.caseTitle ?? payloadText(event, "itemTitle", "title") ?? "This item"; +} + +export function humanizePipelineItemStatus(status: string | null | undefined): string { + const normalized = status?.trim().toLowerCase(); + if (!normalized || normalized === "open") return "Open"; + if (normalized === "done" || normalized === "finished" || normalized === "complete") return "Finished"; + if (normalized === "cancelled" || normalized === "canceled" || normalized === "dropped") return "Removed"; + if (normalized === "in_review" || normalized === "review") return "Needs review"; + if (normalized === "in_progress" || normalized === "working") return "In motion"; + return normalized + .split(/[_\s-]+/) + .filter(Boolean) + .map((part) => part.charAt(0).toUpperCase() + part.slice(1)) + .join(" "); +} + +export function getPendingTransitionBannerState( + item: PipelineCase | null | undefined, + stages: PipelineStage[], +): PendingTransitionBannerState { + if (!item) return { visible: false }; + const fields = asRecord(item.fields); + if (asRecord(fields.suggestionResolution).decision) return { visible: false }; + + const toStageId = asString(fields.nextSuggestedStageId); + if (!toStageId) return { visible: false }; + + return { + visible: true, + toStageId, + toStageName: + asString(fields.nextSuggestedStageName) ?? + stageNameById(stages).get(toStageId) ?? + "the next stage", + reason: asString(fields.nextSuggestedStageReason), + }; +} + +export function itemHasChangedNotice(item: PipelineCase | null | undefined): boolean { + if (!item) return false; + const fields = asRecord(item.fields); + if (fields.changeAcknowledgedAt) return false; + return item.thisChanged === true || fields.thisChanged === true || fields.upstreamChanged === true || !!fields.upstreamDrift; +} + +function formatReviewDecision(decision: string | null): string { + if (decision === "request_changes") return "requested changes"; + if (decision === "drop") return "removed this item"; + return "approved this item"; +} + +export function formatPipelineItemEvent(event: PipelineCaseEvent): PipelineItemEventPresentation { + const title = sentenceSubject(event); + + if (event.kind === "case.ingested") { + const stage = event.toStageName ?? payloadText(event, "toStageName", "stageName"); + return { + tone: "neutral", + sentence: stage ? `${title} was added to ${stage}.` : `${title} was added.`, + }; + } + + if (event.kind === "case.updated") { + return { + tone: "neutral", + sentence: `${title} was updated.`, + }; + } + + if (event.kind === "case.transitioned") { + const fromStage = event.fromStageName ?? payloadText(event, "fromStageName"); + const toStage = event.toStageName ?? payloadText(event, "toStageName", "stageName", "targetStageName"); + const fromCopy = fromStage ? ` from ${fromStage}` : ""; + const toCopy = toStage ? ` to ${toStage}` : ""; + return { + tone: "positive", + sentence: `${title} moved${fromCopy}${toCopy}.`, + }; + } + + if (event.kind === "case.suggested") { + const toStage = event.toStageName ?? payloadText(event, "toStageName", "stageName", "targetStageName"); + const reason = payloadText(event, "reason", "note"); + const reasonCopy = reason ? ` ${reason}` : ""; + return { + tone: "attention", + sentence: toStage + ? `An agent suggested moving ${title} to ${toStage}.${reasonCopy}` + : `An agent suggested moving ${title}.${reasonCopy}`, + }; + } + + if (event.kind === "case.suggestion_resolved") { + const decision = payloadText(event, "decision"); + return { + tone: decision === "accept" ? "positive" : "neutral", + sentence: decision === "accept" + ? `You approved the suggestion for ${title}.` + : `You kept ${title} where it is.`, + }; + } + + if (event.kind === "case.reviewed") { + return { + tone: "neutral", + sentence: `You ${formatReviewDecision(payloadText(event, "decision"))}.`, + }; + } + + return { + tone: "neutral", + sentence: `${title} changed.`, + }; +} + +function humanizeFieldLabel(key: string): string { + return key + .replace(/([a-z0-9])([A-Z])/g, "$1 $2") + .replace(/[_-]+/g, " ") + .trim() + .split(/\s+/) + .filter(Boolean) + .map((part) => part.charAt(0).toUpperCase() + part.slice(1)) + .join(" "); +} + +function formatFieldValue(value: unknown): string | null { + if (value === null || value === undefined || value === "") return null; + if (typeof value === "string") return value; + if (typeof value === "number" || typeof value === "boolean") return String(value); + if (Array.isArray(value)) { + const values = value.map(formatFieldValue).filter((entry): entry is string => Boolean(entry)); + return values.length > 0 ? values.join(", ") : null; + } + const record = asRecord(value); + return asString(record.label) ?? asString(record.name) ?? asString(record.title) ?? "Added details"; +} + +export function displayPipelineItemFields(fields: Record | null | undefined): PipelineItemFieldEntry[] { + const record = asRecord(fields); + return Object.entries(record) + .filter(([key]) => !INTERNAL_FIELD_KEYS.has(key)) + .map(([key, value]) => ({ + key, + label: humanizeFieldLabel(key), + value: formatFieldValue(value), + })) + .filter((entry): entry is PipelineItemFieldEntry => Boolean(entry.value)); +} diff --git a/ui/src/lib/pipeline-learnings.test.ts b/ui/src/lib/pipeline-learnings.test.ts new file mode 100644 index 00000000000..8bc545c457e --- /dev/null +++ b/ui/src/lib/pipeline-learnings.test.ts @@ -0,0 +1,66 @@ +import { describe, expect, it } from "vitest"; +import type { PipelineCaseEvent } from "../api/pipelines"; +import { formatLearningEvent } from "./pipeline-learnings"; + +function event(overrides: Partial): PipelineCaseEvent { + return { + id: "event-1", + companyId: "company-1", + pipelineId: "pipeline-1", + caseId: "item-1", + kind: "review_decided", + createdAt: "2026-06-10T12:00:00.000Z", + payload: {}, + caseTitle: "Launch tweet", + pipelineName: "Content production", + ...overrides, + }; +} + +describe("formatLearningEvent", () => { + it("translates review decisions with a target stage", () => { + const result = formatLearningEvent(event({ + kind: "review_decided", + toStageName: "Publish", + payload: { actorName: "Dotta", decision: "approve" }, + })); + + expect(result.kind).toBe("review"); + expect(result.sentence).toBe("Dotta approved 'Launch tweet' moving to Publish."); + expect(result.sentence).not.toContain("review_decided"); + }); + + it("translates review decisions without optional context", () => { + const result = formatLearningEvent(event({ + kind: "case.reviewed", + payload: { decision: "request_changes" }, + })); + + expect(result.sentence).toBe("Someone sent back 'Launch tweet'."); + expect(result.sentence).not.toContain("case.reviewed"); + }); + + it("translates forced moves with a reason", () => { + const result = formatLearningEvent(event({ + kind: "transition_forced", + fromStageName: "Drafting", + toStageName: "Done", + payload: { reason: "Ready for the campaign" }, + })); + + expect(result.kind).toBe("forced_move"); + expect(result.sentence).toBe("'Launch tweet' was moved by hand from Drafting to Done - reason: Ready for the campaign."); + expect(result.sentence).not.toContain("transition_forced"); + }); + + it("translates forced moves without a reason", () => { + const result = formatLearningEvent(event({ + kind: "case.transitioned", + toStageName: "Done", + payload: {}, + })); + + expect(result.sentence).toBe("'Launch tweet' was moved by hand to Done."); + expect(result.sentence).not.toContain("case.transitioned"); + }); +}); diff --git a/ui/src/lib/pipeline-learnings.ts b/ui/src/lib/pipeline-learnings.ts new file mode 100644 index 00000000000..69173d49f3c --- /dev/null +++ b/ui/src/lib/pipeline-learnings.ts @@ -0,0 +1,88 @@ +import type { PipelineCaseEvent } from "../api/pipelines"; + +export type LearningEventPresentation = { + sentence: string; + kind: "review" | "forced_move" | "unknown"; +}; + +function asRecord(value: unknown): Record { + return typeof value === "object" && value !== null && !Array.isArray(value) + ? value as Record + : {}; +} + +function asString(value: unknown): string | null { + return typeof value === "string" && value.trim().length > 0 ? value.trim() : null; +} + +function eventItemTitle(event: PipelineCaseEvent): string { + const payload = asRecord(event.payload); + return ( + event.caseTitle ?? + asString(payload.itemTitle) ?? + asString(payload.caseTitle) ?? + asString(payload.title) ?? + "Untitled item" + ); +} + +function eventActorName(event: PipelineCaseEvent): string { + const payload = asRecord(event.payload); + return ( + asString(payload.actorName) ?? + asString(payload.reviewerName) ?? + asString(payload.decidedByName) ?? + "Someone" + ); +} + +function payloadText(event: PipelineCaseEvent, ...keys: string[]): string | null { + const payload = asRecord(event.payload); + for (const key of keys) { + const value = asString(payload[key]); + if (value) return value; + } + return null; +} + +function reviewVerb(decision: string | null): string { + if (decision === "request_changes") return "sent back"; + if (decision === "drop") return "dropped"; + return "approved"; +} + +export function formatLearningEvent(event: PipelineCaseEvent): LearningEventPresentation { + const payload = asRecord(event.payload); + const title = eventItemTitle(event); + + if (event.kind === "review_decided" || event.kind === "case.reviewed") { + const actor = eventActorName(event); + const decision = asString(payload.decision); + const toStageName = event.toStageName ?? payloadText(event, "toStageName", "stageName", "targetStageName"); + const stageCopy = toStageName ? ` moving to ${toStageName}` : ""; + const note = payloadText(event, "reason", "note"); + const noteCopy = note ? ` - note: ${note}` : ""; + return { + kind: "review", + sentence: `${actor} ${reviewVerb(decision)} '${title}'${stageCopy}${noteCopy}.`, + }; + } + + if (event.kind === "transition_forced" || event.kind === "case.transitioned") { + const fromStageName = event.fromStageName ?? payloadText(event, "fromStageName"); + const toStageName = event.toStageName ?? payloadText(event, "toStageName", "stageName", "targetStageName"); + const fromCopy = fromStageName ? ` from ${fromStageName}` : ""; + const toCopy = toStageName ? ` to ${toStageName}` : ""; + const reason = payloadText(event, "reason", "note"); + const reasonCopy = reason ? ` - reason: ${reason}` : ""; + return { + kind: "forced_move", + sentence: `'${title}' was moved by hand${fromCopy}${toCopy}${reasonCopy}.`, + }; + } + + return { + kind: "unknown", + sentence: `'${title}' changed.`, + }; +} diff --git a/ui/src/lib/queryKeys.ts b/ui/src/lib/queryKeys.ts index 30a8e83774e..097957fb187 100644 --- a/ui/src/lib/queryKeys.ts +++ b/ui/src/lib/queryKeys.ts @@ -136,6 +136,19 @@ export const queryKeys = { list: (companyId: string) => ["projects", companyId] as const, detail: (id: string) => ["projects", "detail", id] as const, }, + pipelines: { + list: (companyId: string) => ["pipelines", companyId] as const, + detail: (id: string) => ["pipelines", "detail", id] as const, + cases: (id: string) => ["pipelines", "cases", id] as const, + caseDetail: (id: string) => ["pipelines", "case", id] as const, + caseChildren: (id: string) => ["pipelines", "case", id, "children"] as const, + caseEvents: (id: string) => ["pipelines", "case", id, "events"] as const, + caseIssueLinks: (id: string) => ["pipelines", "case", id, "issue-links"] as const, + attention: (companyId: string) => ["pipelines", companyId, "attention"] as const, + reviewCases: (companyId: string) => ["pipelines", companyId, "review-cases"] as const, + learnings: (companyId: string, pipelineId: string, offset: number) => + ["pipelines", companyId, "learnings", pipelineId, offset] as const, + }, goals: { list: (companyId: string) => ["goals", companyId] as const, detail: (id: string) => ["goals", "detail", id] as const, diff --git a/ui/src/pages/Pipelines.test.tsx b/ui/src/pages/Pipelines.test.tsx new file mode 100644 index 00000000000..388de56c737 --- /dev/null +++ b/ui/src/pages/Pipelines.test.tsx @@ -0,0 +1,355 @@ +// @vitest-environment jsdom + +import { act, useState } from "react"; +import { createRoot } from "react-dom/client"; +import { QueryClient, QueryClientProvider } from "@tanstack/react-query"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import type { Pipeline, PipelineAttentionItem, PipelineCase } from "../api/pipelines"; +import { pipelinesApi } from "../api/pipelines"; +import { buildReviewQueueRows, PipelinesIndexTable, ReviewQueueStub } from "./Pipelines"; + +const mockCompany = vi.hoisted(() => ({ selectedCompanyId: "company-1" as string | null })); + +vi.mock("@/lib/router", () => ({ + Link: ({ + children, + to, + ...props + }: React.AnchorHTMLAttributes & { to: string }) => ( + {children} + ), + useNavigate: () => vi.fn(), + useParams: () => ({}), +})); + +vi.mock("../context/CompanyContext", () => ({ + useCompany: () => ({ selectedCompanyId: mockCompany.selectedCompanyId }), +})); + +vi.mock("../context/BreadcrumbContext", () => ({ + useBreadcrumbs: () => ({ setBreadcrumbs: vi.fn() }), +})); + +// eslint-disable-next-line @typescript-eslint/no-explicit-any +(globalThis as any).IS_REACT_ACT_ENVIRONMENT = true; + +function makePipeline(overrides: Partial): Pipeline { + return { + id: "pipeline-1", + companyId: "company-1", + name: "Pipeline", + description: null, + status: "active", + attentionCount: 0, + inMotionCount: 0, + openItemCount: 0, + lastActivityAt: null, + ...overrides, + }; +} + +function connectedPipelines() { + return [ + makePipeline({ + id: "release", + name: "Release", + description: "the launch this work is building toward", + openItemCount: 1, + connections: {}, + }), + makePipeline({ + id: "features", + name: "Features", + attentionCount: 1, + openItemCount: 4, + connections: { feedsIntoPipelineId: "release" }, + }), + makePipeline({ + id: "content", + name: "Content production", + attentionCount: 2, + inMotionCount: 3, + openItemCount: 7, + connections: { feedsIntoPipelineId: "features" }, + }), + ]; +} + +function renderTable({ + pipelines, + connectionsAvailable, + search = "", +}: { + pipelines: Pipeline[]; + connectionsAvailable: boolean; + search?: string; +}) { + const container = document.createElement("div"); + document.body.appendChild(container); + const root = createRoot(container); + + function Harness() { + const [viewMode, setViewMode] = useState<"nested" | "flat">("nested"); + const [query, setQuery] = useState(search); + + return ( + + ); + } + + act(() => { + root.render(); + }); + + return { container, root }; +} + +describe("PipelinesIndexTable", () => { + beforeEach(() => { + document.body.innerHTML = ""; + }); + + afterEach(() => { + document.body.innerHTML = ""; + }); + + it("nests connected pipelines under the work they feed", () => { + const { container, root } = renderTable({ + pipelines: connectedPipelines(), + connectionsAvailable: true, + }); + + const content = container.textContent ?? ""; + expect(content.indexOf("Release")).toBeLessThan(content.indexOf("Features")); + expect(content.indexOf("Features")).toBeLessThan(content.indexOf("Content production")); + expect(content).toContain("feeds into Release"); + expect(content).toContain("feeds into Features"); + + const collapse = container.querySelector('button[aria-label="Collapse Release"]'); + expect(collapse).not.toBeNull(); + + act(() => { + root.unmount(); + }); + }); + + it("switches between nested and flat views when connection data exists", () => { + const { container, root } = renderTable({ + pipelines: connectedPipelines(), + connectionsAvailable: true, + }); + + expect(container.textContent).toContain("feeds into Release"); + + const flatButton = Array.from(container.querySelectorAll("button")).find((button) => + button.textContent?.includes("Flat list"), + ); + expect(flatButton).toBeTruthy(); + + act(() => { + flatButton?.dispatchEvent(new MouseEvent("click", { bubbles: true })); + }); + + expect(container.textContent).not.toContain("feeds into Release"); + expect(container.textContent).not.toContain("feeds into Features"); + + act(() => { + root.unmount(); + }); + }); + + it("disables the nested toggle until the connections field lands", () => { + const { container, root } = renderTable({ + pipelines: [ + makePipeline({ id: "support", name: "Support knowledge base" }), + makePipeline({ id: "sales", name: "Sales decks" }), + ], + connectionsAvailable: false, + }); + + const nestedButton = Array.from(container.querySelectorAll("button")).find((button) => + button.textContent?.includes("Nested"), + ) as HTMLButtonElement | undefined; + expect(nestedButton?.disabled).toBe(true); + expect(container.textContent).toContain("Support knowledge base"); + expect(container.textContent).toContain("Sales decks"); + + act(() => { + root.unmount(); + }); + }); + + it("renders attention and in-motion copy only when nonzero", () => { + const { container, root } = renderTable({ + pipelines: [ + makePipeline({ + id: "hiring", + name: "Hiring pipeline", + attentionCount: 3, + inMotionCount: 2, + openItemCount: 18, + }), + makePipeline({ + id: "recap", + name: "Quarterly board recap", + attentionCount: 0, + inMotionCount: 0, + openItemCount: 0, + status: "paused", + }), + ], + connectionsAvailable: false, + }); + + const content = container.textContent ?? ""; + expect(content).toContain("3 to review"); + expect(content).toContain("2 in motion"); + expect(content).toContain("18 open"); + expect(content).toContain("Paused"); + expect(content).not.toContain("0 to review"); + expect(content).not.toContain("0 in motion"); + + act(() => { + root.unmount(); + }); + }); + + it("shows an empty state when search filters out every pipeline", () => { + const { container, root } = renderTable({ + pipelines: [makePipeline({ id: "press", name: "Press outreach" })], + connectionsAvailable: false, + search: "customer", + }); + + expect(container.textContent).toContain("No pipelines match your search."); + + act(() => { + root.unmount(); + }); + }); +}); + +function makeCase(overrides: Partial): PipelineCase { + return { + id: "item-1", + pipelineId: "pipeline-1", + stageId: "stage-1", + title: "Launch blog post", + status: "open", + fields: {}, + createdAt: "2026-06-10T10:00:00.000Z", + updatedAt: "2026-06-10T10:00:00.000Z", + lastActivityAt: "2026-06-10T10:00:00.000Z", + ...overrides, + }; +} + +function renderReviewQueue() { + const container = document.createElement("div"); + document.body.appendChild(container); + const root = createRoot(container); + const queryClient = new QueryClient({ + defaultOptions: { queries: { retry: false }, mutations: { retry: false } }, + }); + + act(() => { + root.render( + + + , + ); + }); + + return { container, root, queryClient }; +} + +async function flushQueries() { + await Promise.resolve(); + await Promise.resolve(); + await new Promise((resolve) => setTimeout(resolve, 0)); +} + +describe("ReviewQueue", () => { + beforeEach(() => { + document.body.innerHTML = ""; + mockCompany.selectedCompanyId = "company-1"; + vi.spyOn(pipelinesApi, "list").mockResolvedValue([ + makePipeline({ id: "pipeline-1", name: "Content production" }), + ]); + vi.spyOn(pipelinesApi, "listAttention").mockResolvedValue([]); + vi.spyOn(pipelinesApi, "listReviewCases").mockResolvedValue([]); + vi.spyOn(pipelinesApi, "reviewCase").mockResolvedValue(makeCase({ id: "review-1" })); + vi.spyOn(pipelinesApi, "resolveSuggestion").mockResolvedValue(makeCase({ id: "suggestion-1" })); + vi.spyOn(pipelinesApi, "bulkReviewCases").mockResolvedValue({ cases: [] }); + }); + + afterEach(() => { + vi.restoreAllMocks(); + document.body.innerHTML = ""; + }); + + it("groups attention and review rows into the daily triage sections", () => { + const rows = buildReviewQueueRows({ + pipelines: [makePipeline({ id: "pipeline-1", name: "Content production" })], + reviewCases: [makeCase({ id: "review-1", title: "Final launch post" })], + attentionItems: [ + { + id: "suggestion-1", + pipelineId: "pipeline-1", + caseId: "suggestion-1", + kind: "suggestion", + title: "Draft launch post", + summary: "Drafting agent thinks Draft launch post is ready to move forward.", + createdAt: "2026-06-10T11:00:00.000Z", + } satisfies PipelineAttentionItem, + { + id: "heads-up-1", + pipelineId: "pipeline-1", + kind: "drift", + title: "Launch tweet", + summary: "Launch tweet changed upstream.", + createdAt: "2026-06-10T09:00:00.000Z", + } satisfies PipelineAttentionItem, + ], + }); + + expect(rows.map((row) => row.kind)).toEqual(["suggestion", "review", "headsUp"]); + expect(rows.find((row) => row.kind === "review")?.pipelineName).toBe("Content production"); + expect(rows.map((row) => row.prompt).join(" ")).not.toMatch(/\bcase\b/i); + }); + + it("approves a final-review row through the review endpoint", async () => { + vi.mocked(pipelinesApi.listReviewCases).mockResolvedValue([ + makeCase({ id: "review-1", title: "Final launch post" }), + ]); + const { container, root, queryClient } = renderReviewQueue(); + + await flushQueries(); + + expect(container.textContent).toContain("Final calls"); + const approveButton = Array.from(container.querySelectorAll("button")).find( + (button) => button.textContent === "Approve", + ); + expect(approveButton).toBeTruthy(); + + approveButton?.dispatchEvent(new MouseEvent("click", { bubbles: true })); + await flushQueries(); + + expect(pipelinesApi.reviewCase).toHaveBeenCalledWith("review-1", { + decision: "approve", + note: null, + }); + + act(() => { + root.unmount(); + }); + queryClient.clear(); + }); +}); diff --git a/ui/src/pages/Pipelines.tsx b/ui/src/pages/Pipelines.tsx index d8f97701b90..9f836daa506 100644 --- a/ui/src/pages/Pipelines.tsx +++ b/ui/src/pages/Pipelines.tsx @@ -1,12 +1,21 @@ -import { useEffect, useMemo, useState, type FormEvent } from "react"; +import { useEffect, useMemo, useRef, useState, type FormEvent } from "react"; import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query"; import { + AlertTriangle, BookOpenText, + Check, + CheckCircle2, ChevronDown, ChevronRight, + ExternalLink, + GitBranch, Hexagon, + Loader2, + MessageSquare, + MoreHorizontal, Plus, Search, + Trash2, } from "lucide-react"; import { DndContext, @@ -27,14 +36,17 @@ import { } from "@dnd-kit/sortable"; import type { Pipeline, + PipelineAttentionItem, PipelineCase, PipelineCaseEvent, PipelineConnectionRef, PipelineConnections, + PipelineIssueLink, PipelineStage, PipelineTransition, } from "../api/pipelines"; import { pipelinesApi } from "../api/pipelines"; +import { issuesApi } from "../api/issues"; import { useCompany } from "../context/CompanyContext"; import { useBreadcrumbs } from "../context/BreadcrumbContext"; import { useToastActions } from "../context/ToastContext"; @@ -47,11 +59,26 @@ import { DialogHeader, DialogTitle, } from "@/components/ui/dialog"; +import { + DropdownMenu, + DropdownMenuContent, + DropdownMenuItem, + DropdownMenuTrigger, +} from "@/components/ui/dropdown-menu"; import { Input } from "@/components/ui/input"; import { Select, SelectContent, SelectItem, SelectTrigger, SelectValue } from "@/components/ui/select"; import { Textarea } from "@/components/ui/textarea"; import { EmptyState } from "../components/EmptyState"; +import { IssueChatThread } from "../components/IssueChatThread"; import { PageSkeleton } from "../components/PageSkeleton"; +import { hasBlockingShortcutDialog, isKeyboardShortcutTextInputTarget } from "../lib/keyboardShortcuts"; +import { + displayPipelineItemFields, + formatPipelineItemEvent, + getPendingTransitionBannerState, + humanizePipelineItemStatus, + itemHasChangedNotice, +} from "../lib/pipeline-item-detail"; import { formatLearningEvent } from "../lib/pipeline-learnings"; import { queryKeys } from "../lib/queryKeys"; import { cn, formatNumber, formatShortDate, relativeTime } from "../lib/utils"; @@ -225,6 +252,127 @@ function formatPipelineActivity(value: string | Date | null) { return new Date(value).toLocaleDateString("en-US", { month: "short", day: "numeric" }); } +type ReviewQueueKind = "suggestion" | "review" | "headsUp"; + +export interface ReviewQueueRow { + id: string; + caseId: string | null; + pipelineId: string; + pipelineName: string; + title: string; + prompt: string; + kind: ReviewQueueKind; + createdAt: string | Date | null; + caseItem?: PipelineCase; +} + +const REVIEW_QUEUE_SECTION_LABELS: Record = { + suggestion: "Suggestions to review", + review: "Final calls", + headsUp: "Heads-up", +}; + +const REVIEW_QUEUE_SECTION_ORDER: ReviewQueueKind[] = ["suggestion", "review", "headsUp"]; + +function attentionItemKind(item: PipelineAttentionItem): ReviewQueueKind { + const rawKind = String(item.kind ?? "").toLowerCase(); + if (rawKind.includes("suggest")) return "suggestion"; + if (rawKind.includes("review") || rawKind.includes("final")) return "review"; + return "headsUp"; +} + +function hasPendingSuggestion(caseItem: PipelineCase) { + const fields = caseItem.fields ?? {}; + return Boolean(fields.nextSuggestedStageId && !fields.suggestionResolution); +} + +function humanizeFieldLabel(key: string) { + return key + .replace(/([a-z])([A-Z])/g, "$1 $2") + .replace(/[_-]+/g, " ") + .replace(/\b\w/g, (char) => char.toUpperCase()); +} + +function reviewCasePrompt(caseItem: PipelineCase) { + const fields = caseItem.fields ?? {}; + if (typeof fields.decision === "string" && fields.decision.trim()) { + return fields.decision.trim(); + } + return `Decide whether ${caseItem.title} is ready to move forward.`; +} + +function suggestionPrompt(item: PipelineAttentionItem, pipelineName: string) { + return item.summary ?? `${pipelineName} thinks ${item.title} is ready to move forward.`; +} + +function headsUpPrompt(item: PipelineAttentionItem) { + return item.summary ?? `${item.title} needs a quick look before work continues.`; +} + +export function buildReviewQueueRows({ + attentionItems, + reviewCases, + pipelines, +}: { + attentionItems: PipelineAttentionItem[]; + reviewCases: PipelineCase[]; + pipelines: Pipeline[]; +}): ReviewQueueRow[] { + const pipelineNameById = new Map(pipelines.map((pipeline) => [pipeline.id, pipeline.name])); + const reviewCaseById = new Map(reviewCases.map((caseItem) => [caseItem.id, caseItem])); + const rows = new Map(); + + for (const item of attentionItems) { + const caseId = item.caseId ?? item.id ?? null; + const caseItem = caseId ? reviewCaseById.get(caseId) : undefined; + const kind = caseItem ? "review" : attentionItemKind(item); + const pipelineName = pipelineNameById.get(item.pipelineId) ?? "Pipeline"; + rows.set(`${kind}:${caseId ?? item.id}`, { + id: `${kind}:${caseId ?? item.id}`, + caseId, + pipelineId: item.pipelineId, + pipelineName, + title: item.title, + prompt: + kind === "suggestion" + ? suggestionPrompt(item, pipelineName) + : kind === "review" && caseItem + ? reviewCasePrompt(caseItem) + : headsUpPrompt(item), + kind, + createdAt: item.createdAt ?? caseItem?.lastActivityAt ?? caseItem?.updatedAt ?? null, + caseItem, + }); + } + + for (const caseItem of reviewCases) { + const kind: ReviewQueueKind = hasPendingSuggestion(caseItem) ? "suggestion" : "review"; + const pipelineName = pipelineNameById.get(caseItem.pipelineId) ?? "Pipeline"; + const id = `${kind}:${caseItem.id}`; + if (rows.has(id)) continue; + rows.set(id, { + id, + caseId: caseItem.id, + pipelineId: caseItem.pipelineId, + pipelineName, + title: caseItem.title, + prompt: + kind === "suggestion" + ? `${pipelineName} thinks ${caseItem.title} is ready to move forward.` + : reviewCasePrompt(caseItem), + kind, + createdAt: caseItem.lastActivityAt ?? caseItem.updatedAt ?? caseItem.createdAt ?? null, + caseItem, + }); + } + + return [...rows.values()].sort((left, right) => { + const leftTime = left.createdAt ? new Date(left.createdAt).getTime() : 0; + const rightTime = right.createdAt ? new Date(right.createdAt).getTime() : 0; + return rightTime - leftTime; + }); +} + const LEARNINGS_PAGE_SIZE = 100; const LEARNING_EVENT_TYPES = "review_decided,transition_forced"; @@ -596,7 +744,7 @@ function PipelineCaseCard({ const isWorking = isWorkingCase(caseItem); const blockerCount = getOpenBlockerCount(caseItem); const hasNeedsAttention = blockerCount > 0; - const hasThisChanged = hasThisChanged(caseItem); + const hasChangedNotice = hasThisChanged(caseItem); const childrenSummary = getChildrenSummaryCount(caseItem); const { attributes, @@ -642,7 +790,7 @@ function PipelineCaseCard({ Needs attention ) : null} - {hasThisChanged ? ( + {hasChangedNotice ? ( This changed @@ -958,7 +1106,7 @@ export function PipelineBoardStub() { onError: (error) => { pushToast({ title: "Move blocked", - body: error instanceof Error ? error.message : "Unable to move case", + body: error instanceof Error ? error.message : "Unable to move item", tone: "error", }); queryClient.invalidateQueries({ queryKey: queryKeys.pipelines.detail(pipelineId!) }); @@ -1072,7 +1220,7 @@ export function PipelineBoardStub() {

{pipeline.name}

{pipeline.description ?

{pipeline.description}

: null} -

{cases.length} total case{cases.length === 1 ? "" : "s"}

+

{cases.length} total item{cases.length === 1 ? "" : "s"}