Add pipeline upstream drift notices

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

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

@@ -124,6 +124,28 @@ describeEmbeddedPostgres("pipelineService", () => {
     return count ?? 0;
   }
 
+  async function seedLinkedIssue(input: {
+    companyId: string;
+    caseId: string;
+    role: "origin" | "conversation" | "work" | "automation";
+    status?: "backlog" | "todo" | "in_progress" | "in_review" | "done" | "blocked" | "cancelled";
+    title?: string;
+  }) {
+    const [issue] = await db.insert(issues).values({
+      companyId: input.companyId,
+      title: input.title ?? `${input.role} issue`,
+      status: input.status ?? "todo",
+      priority: "medium",
+    }).returning();
+    await db.insert(pipelineCaseIssueLinks).values({
+      companyId: input.companyId,
+      caseId: input.caseId,
+      issueId: issue!.id,
+      role: input.role,
+    });
+    return issue!;
+  }
+
   it("seeds default stages and protects non-empty stage deletion", async () => {
     const { company, pipeline, byKey } = await seedPipeline();
 
@@ -478,6 +500,162 @@ describeEmbeddedPostgres("pipelineService", () => {
     expect(moved.case.version).toBe(2);
   });
 
+  it("posts upstream drift notices to active dependent work issues only", async () => {
+    const { company, pipeline } = await seedPipeline();
+    const upstream = await svc.ingestCase({
+      companyId: company.id,
+      pipelineId: pipeline.id,
+      caseKey: "draft",
+      title: "Draft",
+      actor: userActor,
+    });
+    const workDependent = await svc.ingestCase({
+      companyId: company.id,
+      pipelineId: pipeline.id,
+      caseKey: "asset-work",
+      title: "Asset work",
+      blockedByCaseIds: [upstream.case.id],
+      actor: userActor,
+    });
+    const conversationDependent = await svc.ingestCase({
+      companyId: company.id,
+      pipelineId: pipeline.id,
+      caseKey: "asset-conversation",
+      title: "Asset conversation",
+      blockedByCaseIds: [upstream.case.id],
+      actor: userActor,
+    });
+    const workIssue = await seedLinkedIssue({
+      companyId: company.id,
+      caseId: workDependent.case.id,
+      role: "work",
+      title: "Asset work issue",
+    });
+    const conversationIssue = await seedLinkedIssue({
+      companyId: company.id,
+      caseId: conversationDependent.case.id,
+      role: "conversation",
+      title: "Conversation issue",
+    });
+
+    const updated = await svc.patchCaseContent({
+      companyId: company.id,
+      caseId: upstream.case.id,
+      title: "Draft v2",
+      expectedVersion: 1,
+      actor: userActor,
+    });
+
+    expect(updated.version).toBe(2);
+    const workComments = await db.select().from(issueComments).where(eq(issueComments.issueId, workIssue.id));
+    expect(workComments).toHaveLength(1);
+    expect(workComments[0]!.authorType).toBe("system");
+    expect(workComments[0]!.body).toBe(
+      `Upstream case [draft](/PAP/pipelines/${pipeline.id}/cases/${upstream.case.id}) changed (v1→v2).`,
+    );
+    const conversationComments = await db.select().from(issueComments).where(eq(issueComments.issueId, conversationIssue.id));
+    expect(conversationComments).toHaveLength(0);
+  });
+
+  it("skips upstream drift notices for terminal dependents and dependents without work issues", async () => {
+    const { company, pipeline } = await seedPipeline();
+    const upstream = await svc.ingestCase({
+      companyId: company.id,
+      pipelineId: pipeline.id,
+      caseKey: "source",
+      title: "Source",
+      actor: userActor,
+    });
+    const terminalDependent = await svc.ingestCase({
+      companyId: company.id,
+      pipelineId: pipeline.id,
+      stageKey: "done",
+      caseKey: "terminal-dependent",
+      title: "Terminal dependent",
+      actor: userActor,
+    });
+    await svc.replaceBlockers({
+      companyId: company.id,
+      caseId: terminalDependent.case.id,
+      blockedByCaseIds: [upstream.case.id],
+      actor: userActor,
+    });
+    const noWorkDependent = await svc.ingestCase({
+      companyId: company.id,
+      pipelineId: pipeline.id,
+      caseKey: "no-work-dependent",
+      title: "No work dependent",
+      blockedByCaseIds: [upstream.case.id],
+      actor: userActor,
+    });
+    const terminalIssue = await seedLinkedIssue({
+      companyId: company.id,
+      caseId: terminalDependent.case.id,
+      role: "work",
+      title: "Terminal work issue",
+    });
+    const conversationIssue = await seedLinkedIssue({
+      companyId: company.id,
+      caseId: noWorkDependent.case.id,
+      role: "conversation",
+      title: "Non-work issue",
+    });
+
+    await svc.patchCaseContent({
+      companyId: company.id,
+      caseId: upstream.case.id,
+      summary: "Updated source",
+      expectedVersion: 1,
+      actor: userActor,
+    });
+
+    const terminalComments = await db.select().from(issueComments).where(eq(issueComments.issueId, terminalIssue.id));
+    expect(terminalComments).toHaveLength(0);
+    const conversationComments = await db.select().from(issueComments).where(eq(issueComments.issueId, conversationIssue.id));
+    expect(conversationComments).toHaveLength(0);
+  });
+
+  it("does not bump versions or notify dependents on no-op content patches", async () => {
+    const { company, pipeline } = await seedPipeline();
+    const upstream = await svc.ingestCase({
+      companyId: company.id,
+      pipelineId: pipeline.id,
+      caseKey: "noop-source",
+      title: "No-op source",
+      fields: { channel: "blog" },
+      actor: userActor,
+    });
+    const dependent = await svc.ingestCase({
+      companyId: company.id,
+      pipelineId: pipeline.id,
+      caseKey: "noop-dependent",
+      title: "No-op dependent",
+      blockedByCaseIds: [upstream.case.id],
+      actor: userActor,
+    });
+    const workIssue = await seedLinkedIssue({
+      companyId: company.id,
+      caseId: dependent.case.id,
+      role: "work",
+      title: "No-op work issue",
+    });
+    const beforeEvents = await eventCount(upstream.case.id);
+
+    const patched = await svc.patchCaseContent({
+      companyId: company.id,
+      caseId: upstream.case.id,
+      title: "No-op source",
+      fields: { channel: "blog" },
+      expectedVersion: 1,
+      actor: userActor,
+    });
+
+    expect(patched.version).toBe(1);
+    expect(await eventCount(upstream.case.id)).toBe(beforeEvents);
+    const comments = await db.select().from(issueComments).where(eq(issueComments.issueId, workIssue.id));
+    expect(comments).toHaveLength(0);
+  });
+
   it("resolves in-batch forward blocker case keys", async () => {
     const { company, pipeline } = await seedPipeline();
 

server/src/services/pipelines.ts

@@ -1,4 +1,5 @@
 import { randomUUID } from "node:crypto";
+import { isDeepStrictEqual } from "node:util";
 import { and, asc, desc, eq, inArray, isNull, ne, or, sql } from "drizzle-orm";
 import { alias } from "drizzle-orm/pg-core";
 import type { Db } from "@paperclipai/db";
@@ -588,6 +589,53 @@ async function handleBlockersResolved(db: PipelineDb, companyId: string, blocker
   }
 }
 
+async function notifyDependentWorkIssuesOfUpstreamContentChange(
+  db: PipelineDb,
+  input: {
+    companyId: string;
+    upstreamCase: typeof pipelineCases.$inferSelect;
+    previousVersion: number;
+    version: number;
+  },
+) {
+  const rows = await db
+    .selectDistinctOn([issues.id], { issueId: issues.id })
+    .from(pipelineCaseBlockers)
+    .innerJoin(pipelineCases, eq(pipelineCaseBlockers.caseId, pipelineCases.id))
+    .innerJoin(pipelineCaseIssueLinks, eq(pipelineCaseIssueLinks.caseId, pipelineCases.id))
+    .innerJoin(issues, eq(pipelineCaseIssueLinks.issueId, issues.id))
+    .where(and(
+      eq(pipelineCaseBlockers.companyId, input.companyId),
+      eq(pipelineCaseBlockers.blockedByCaseId, input.upstreamCase.id),
+      eq(pipelineCases.companyId, input.companyId),
+      isNull(pipelineCases.terminalKind),
+      eq(pipelineCaseIssueLinks.companyId, input.companyId),
+      eq(pipelineCaseIssueLinks.role, "work"),
+      eq(issues.companyId, input.companyId),
+      ne(issues.status, "done"),
+      ne(issues.status, "cancelled"),
+      isNull(issues.hiddenAt),
+    ));
+
+  if (rows.length === 0) return;
+
+  const upstreamLink = buildCaseDeepLink({
+    pipelineId: input.upstreamCase.pipelineId,
+    caseId: input.upstreamCase.id,
+  });
+  const body = `Upstream case [${input.upstreamCase.caseKey}](${upstreamLink}) changed (v${input.previousVersion}→v${input.version}).`;
+
+  for (const row of rows) {
+    await db.insert(issueComments).values({
+      companyId: input.companyId,
+      issueId: row.issueId,
+      authorType: "system",
+      body,
+    });
+    await db.update(issues).set({ updatedAt: nowDate() }).where(eq(issues.id, row.issueId));
+  }
+}
+
 async function validateBlockerSet(
   db: PipelineDb,
   input: { companyId: string; caseId: string; blockedByCaseIds: string[] },
@@ -863,14 +911,23 @@ export function pipelineService(db: Db, deps: { heartbeat?: IssueAssignmentWakeu
         parentCaseId: input.parentCaseId,
       });
     }
+    const titleChanged = input.title !== undefined && input.title !== current.title;
+    const summaryChanged = input.summary !== undefined && input.summary !== current.summary;
+    const fieldsChanged = input.fields !== undefined && !isDeepStrictEqual(input.fields, current.fields);
+    const parentCaseChanged = input.parentCaseId !== undefined && input.parentCaseId !== current.parentCaseId;
+    const contentChanged = titleChanged || summaryChanged || fieldsChanged;
+    if (!contentChanged && !parentCaseChanged) {
+      return { case: current, event: null };
+    }
+
     const patch: Partial<typeof pipelineCases.$inferInsert> = {
       version: current.version + 1,
       updatedAt: nowDate(),
     };
-    if (input.title !== undefined) patch.title = input.title;
-    if (input.summary !== undefined) patch.summary = input.summary;
-    if (input.fields !== undefined) patch.fields = input.fields;
-    if (input.parentCaseId !== undefined) patch.parentCaseId = input.parentCaseId;
+    if (titleChanged) patch.title = input.title;
+    if (summaryChanged) patch.summary = input.summary;
+    if (fieldsChanged) patch.fields = input.fields;
+    if (parentCaseChanged) patch.parentCaseId = input.parentCaseId;
 
     const [updated] = await tx
       .update(pipelineCases)
@@ -890,10 +947,10 @@ export function pipelineService(db: Db, deps: { heartbeat?: IssueAssignmentWakeu
       payload: {
         previousVersion: current.version,
         version: updated.version,
-        parentCaseChanged: input.parentCaseId !== undefined && input.parentCaseId !== current.parentCaseId,
+        parentCaseChanged,
       },
     });
-    if (input.parentCaseId !== undefined && input.parentCaseId !== current.parentCaseId) {
+    if (parentCaseChanged) {
       const terminalDelta = isTerminalKind(current.terminalKind) ? 1 : 0;
       await adjustParentCounts(tx, {
         parentCaseId: current.parentCaseId,
@@ -909,6 +966,14 @@ export function pipelineService(db: Db, deps: { heartbeat?: IssueAssignmentWakeu
         await handleChildrenTerminal(tx, input.companyId, input.parentCaseId);
       }
     }
+    if (contentChanged) {
+      await notifyDependentWorkIssuesOfUpstreamContentChange(tx, {
+        companyId: input.companyId,
+        upstreamCase: updated,
+        previousVersion: current.version,
+        version: updated.version,
+      });
+    }
     return { case: updated, event };
   }