Add pipelines aggregation endpoints

Adds the PAP-10830 backend aggregation layer for the pipelines UI:
attention feed (suggestions/reviews/headsUp), company-wide case-events
learnings feed, recursive case children tree with rollup counts,
activeWork on case detail and case lists, and observed cross-pipeline
connections on the pipelines list. Introduces a first-class
upstream_drift case event so unresolved drift is queryable instead of
living only in issue comments.

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

packages/db/src/migrations/0102_pipeline_upstream_drift_event.sql

@@ -0,0 +1,21 @@
+ALTER TABLE "pipeline_case_events" DROP CONSTRAINT "pipeline_case_events_type_check";--> statement-breakpoint
+ALTER TABLE "pipeline_case_events" ADD CONSTRAINT "pipeline_case_events_type_check" CHECK ("pipeline_case_events"."type" in (
+        'ingested',
+        'updated',
+        'claimed',
+        'lease_released',
+        'lease_expired',
+        'transitioned',
+        'transition_suggested',
+        'suggestion_resolved',
+        'review_decided',
+        'conversation_opened',
+        'issue_linked',
+        'issue_unlinked',
+        'automation_executed',
+        'automation_failed',
+        'blockers_set',
+        'blockers_resolved',
+        'children_terminal',
+        'upstream_drift'
+      ));

packages/db/src/migrations/meta/_journal.json

@@ -715,6 +715,13 @@
       "when": 1781098769536,
       "tag": "0101_special_bloodstorm",
       "breakpoints": true
+    },
+    {
+      "idx": 102,
+      "version": "7",
+      "when": 1781145600000,
+      "tag": "0102_pipeline_upstream_drift_event",
+      "breakpoints": true
     }
   ]
-}
+}

packages/db/src/schema/pipeline_case_events.ts

@@ -44,7 +44,8 @@ export const pipelineCaseEvents = pgTable(
         'automation_failed',
         'blockers_set',
         'blockers_resolved',
-        'children_terminal'
+        'children_terminal',
+        'upstream_drift'
       )`,
     ),
     actorTypeCheck: check("pipeline_case_events_actor_type_check", sql`${table.actorType} in ('user', 'agent', 'system')`),

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

@@ -0,0 +1,479 @@
+import { randomUUID } from "node:crypto";
+import express from "express";
+import request from "supertest";
+import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest";
+import {
+  activityLog,
+  agents,
+  companies,
+  createDb,
+  issueComments,
+  issues,
+  pipelineAutomationExecutions,
+  pipelineCaseBlockers,
+  pipelineCaseEvents,
+  pipelineCaseIssueLinks,
+  pipelineCases,
+  pipelineStages,
+  pipelineTransitions,
+  pipelines,
+} from "@paperclipai/db";
+import {
+  getEmbeddedPostgresTestSupport,
+  startEmbeddedPostgresTestDatabase,
+} from "./helpers/embedded-postgres.js";
+import { errorHandler } from "../middleware/error-handler.js";
+import { pipelineRoutes } from "../routes/pipelines.js";
+
+const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
+const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe.sequential : describe.skip;
+
+if (!embeddedPostgresSupport.supported) {
+  console.warn(
+    `Skipping embedded Postgres pipeline aggregation tests on this host: ${embeddedPostgresSupport.reason ?? "unsupported environment"}`,
+  );
+}
+
+describeEmbeddedPostgres("pipeline aggregation routes", () => {
+  let db!: ReturnType<typeof createDb>;
+  let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
+  const noopHeartbeat = { wakeup: async () => null };
+
+  beforeAll(async () => {
+    tempDb = await startEmbeddedPostgresTestDatabase("paperclip-pipelines-aggregation-");
+    db = createDb(tempDb.connectionString);
+  }, 20_000);
+
+  afterEach(async () => {
+    await db.delete(activityLog);
+    await db.delete(pipelineAutomationExecutions);
+    await db.delete(pipelineCaseBlockers);
+    await db.delete(pipelineCaseIssueLinks);
+    await db.delete(pipelineCaseEvents);
+    await db.delete(pipelineCases);
+    await db.delete(pipelineTransitions);
+    await db.delete(pipelineStages);
+    await db.delete(issueComments);
+    await db.delete(issues);
+    await db.delete(pipelines);
+    await db.delete(agents);
+    await db.delete(companies);
+  });
+
+  afterAll(async () => {
+    await tempDb?.cleanup();
+  });
+
+  function app(actor: Express.Request["actor"]) {
+    const instance = express();
+    instance.use(express.json());
+    instance.use((req, _res, next) => {
+      req.actor = actor;
+      next();
+    });
+    instance.use("/api", pipelineRoutes(db, { heartbeat: noopHeartbeat }));
+    instance.use(errorHandler);
+    return instance;
+  }
+
+  async function seedCompany(name = "Aggregation Co") {
+    const [company] = await db.insert(companies).values({
+      name,
+      issuePrefix: `P${randomUUID().replace(/-/g, "").slice(0, 6).toUpperCase()}`,
+    }).returning();
+    return company!;
+  }
+
+  async function seedAgent(companyId: string, name = "Pipeline Agent") {
+    const [agent] = await db.insert(agents).values({
+      companyId,
+      name,
+      role: "engineer",
+      adapterType: "codex_local",
+    }).returning();
+    return agent!;
+  }
+
+  const boardActor: Express.Request["actor"] = {
+    type: "board",
+    userId: "board-user",
+    source: "local_implicit",
+    isInstanceAdmin: true,
+  };
+
+  function agentActor(companyId: string, agentId: string): Express.Request["actor"] {
+    return {
+      type: "agent",
+      agentId,
+      companyId,
+      runId: randomUUID(),
+      source: "agent_key",
+    };
+  }
+
+  const reviewStages = [
+    { key: "intake", name: "Intake", kind: "open", position: 100 },
+    { key: "drafting", name: "Drafting", kind: "working", position: 200 },
+    {
+      key: "review_human",
+      name: "Final review",
+      kind: "review",
+      position: 300,
+      config: { approveToStageKey: "done", rejectToStageKey: "cancelled", requestChangesToStageKey: "drafting", requireRejectReason: true, reviewerKind: "human" },
+    },
+    {
+      key: "review_any",
+      name: "Open review",
+      kind: "review",
+      position: 400,
+      config: { approveToStageKey: "done", rejectToStageKey: "cancelled", requireRejectReason: false, reviewerKind: "any" },
+    },
+    { key: "done", name: "Done", kind: "done", position: 900 },
+    { key: "cancelled", name: "Cancelled", kind: "cancelled", position: 1000 },
+  ];
+
+  async function seedPipeline(http: ReturnType<typeof request>, companyId: string, key: string, name: string) {
+    const created = await http
+      .post(`/api/companies/${companyId}/pipelines`)
+      .send({ key, name, stages: reviewStages })
+      .expect(201);
+    return created.body as { id: string; stages: Array<{ id: string; key: string }> };
+  }
+
+  it("groups attention into suggestions, reviews, and heads-up with approver filtering", async () => {
+    const company = await seedCompany();
+    const agent = await seedAgent(company.id, "Drafting Agent");
+    const board = request(app(boardActor));
+    const asAgent = request(app(agentActor(company.id, agent.id)));
+
+    const pipeline = await seedPipeline(board, company.id, "content", "Content production");
+
+    const suggested = await board
+      .post(`/api/pipelines/${pipeline.id}/cases`)
+      .send({ caseKey: "suggested", title: "Launch blog post" })
+      .expect(201);
+    await asAgent
+      .post(`/api/cases/${suggested.body.case.id}/suggest-transition`)
+      .send({ toStageKey: "review_human", rationale: "Draft is ready for review", confidence: 0.9 })
+      .expect(200);
+
+    const humanReview = await board
+      .post(`/api/pipelines/${pipeline.id}/cases`)
+      .send({ caseKey: "human-review", title: "Needs a human decision" })
+      .expect(201);
+    await board
+      .post(`/api/cases/${humanReview.body.case.id}/transition`)
+      .send({ toStageKey: "review_human", expectedVersion: 1 })
+      .expect(200);
+
+    const anyReview = await board
+      .post(`/api/pipelines/${pipeline.id}/cases`)
+      .send({ caseKey: "any-review", title: "Anyone can decide" })
+      .expect(201);
+    await board
+      .post(`/api/cases/${anyReview.body.case.id}/transition`)
+      .send({ toStageKey: "review_any", expectedVersion: 1 })
+      .expect(200);
+
+    const upstream = await board
+      .post(`/api/pipelines/${pipeline.id}/cases`)
+      .send({ caseKey: "upstream", title: "Release plan" })
+      .expect(201);
+    const dependent = await board
+      .post(`/api/pipelines/${pipeline.id}/cases`)
+      .send({ caseKey: "dependent", title: "Dependent draft", blockedByCaseIds: [upstream.body.case.id] })
+      .expect(201);
+    const [workIssue] = await db.insert(issues).values({
+      companyId: company.id,
+      title: "Write dependent draft",
+      status: "in_progress",
+      priority: "medium",
+      assigneeAgentId: agent.id,
+    }).returning();
+    await board
+      .post(`/api/cases/${dependent.body.case.id}/issue-links`)
+      .send({ issueId: workIssue!.id, role: "work" })
+      .expect(201);
+    await board
+      .patch(`/api/cases/${upstream.body.case.id}`)
+      .send({ summary: "Scope changed", expectedVersion: 1 })
+      .expect(200);
+
+    const boardFeed = await board.get(`/api/companies/${company.id}/pipelines-attention`).expect(200);
+
+    expect(boardFeed.body.counts).toEqual({ suggestions: 1, reviews: 2, headsUp: 1 });
+    const suggestion = boardFeed.body.suggestions[0];
+    expect(suggestion.case.id).toBe(suggested.body.case.id);
+    expect(suggestion.case.pipeline.name).toBe("Content production");
+    expect(suggestion.suggestion.toStageKey).toBe("review_human");
+    expect(suggestion.suggestion.toStageName).toBe("Final review");
+    expect(suggestion.suggestion.rationale).toBe("Draft is ready for review");
+    expect(suggestion.suggestion.suggestedBy).toEqual({ agentId: agent.id, agentName: "Drafting Agent" });
+
+    const reviewCaseIds = boardFeed.body.reviews.map((entry: { case: { id: string } }) => entry.case.id);
+    expect(reviewCaseIds).toEqual(expect.arrayContaining([humanReview.body.case.id, anyReview.body.case.id]));
+    const humanEntry = boardFeed.body.reviews.find((entry: { case: { id: string } }) => entry.case.id === humanReview.body.case.id);
+    expect(humanEntry.review).toMatchObject({
+      reviewerKind: "human",
+      approveToStageKey: "done",
+      rejectToStageKey: "cancelled",
+      requestChangesToStageKey: "drafting",
+      requireRejectReason: true,
+      expectedVersion: 2,
+    });
+
+    const headsUp = boardFeed.body.headsUp[0];
+    expect(headsUp.case.id).toBe(dependent.body.case.id);
+    expect(headsUp.drift.upstream).toMatchObject({
+      caseId: upstream.body.case.id,
+      caseKey: "upstream",
+      title: "Release plan",
+      pipelineId: pipeline.id,
+      pipelineName: "Content production",
+    });
+    expect(headsUp.drift.previousVersion).toBe(1);
+    expect(headsUp.drift.version).toBe(2);
+    expect(headsUp.workIssue).toMatchObject({ issueId: workIssue!.id, status: "in_progress" });
+    expect(headsUp.activeWork).toMatchObject({ issueId: workIssue!.id, agentId: agent.id, agentName: "Drafting Agent" });
+
+    // Agent callers only await review stages with reviewerKind "any".
+    const agentFeed = await asAgent.get(`/api/companies/${company.id}/pipelines-attention`).expect(200);
+    expect(agentFeed.body.reviews.map((entry: { case: { id: string } }) => entry.case.id)).toEqual([anyReview.body.case.id]);
+
+    // Touching the dependent case resolves the drift notice.
+    await board
+      .patch(`/api/cases/${dependent.body.case.id}`)
+      .send({ title: "Dependent draft v2", expectedVersion: 1 })
+      .expect(200);
+    const resolvedFeed = await board.get(`/api/companies/${company.id}/pipelines-attention`).expect(200);
+    expect(resolvedFeed.body.headsUp).toEqual([]);
+
+    // Terminal cases drop out of the suggestions group.
+    await board
+      .post(`/api/cases/${anyReview.body.case.id}/review`)
+      .send({ decision: "approve", expectedVersion: 2 })
+      .expect(200);
+    const afterApprove = await board.get(`/api/companies/${company.id}/pipelines-attention`).expect(200);
+    expect(afterApprove.body.reviews.map((entry: { case: { id: string } }) => entry.case.id)).toEqual([humanReview.body.case.id]);
+  });
+
+  it("serves the company-wide learnings feed with joins, filtering, and pagination", async () => {
+    const company = await seedCompany();
+    const board = request(app(boardActor));
+    const pipeline = await seedPipeline(board, company.id, "content", "Content production");
+
+    const first = await board
+      .post(`/api/pipelines/${pipeline.id}/cases`)
+      .send({ caseKey: "first", title: "First item" })
+      .expect(201);
+    await board
+      .post(`/api/cases/${first.body.case.id}/transition`)
+      .send({ toStageKey: "review_human", expectedVersion: 1 })
+      .expect(200);
+    await board
+      .post(`/api/cases/${first.body.case.id}/review`)
+      .send({ decision: "reject", reason: "Wrong audience", expectedVersion: 2 })
+      .expect(200);
+
+    const second = await board
+      .post(`/api/pipelines/${pipeline.id}/cases`)
+      .send({ caseKey: "second", title: "Second item" })
+      .expect(201);
+    await board
+      .post(`/api/cases/${second.body.case.id}/transition`)
+      .send({ toStageKey: "review_human", expectedVersion: 1 })
+      .expect(200);
+    await board
+      .post(`/api/cases/${second.body.case.id}/review`)
+      .send({ decision: "request_changes", reason: "Tighten the intro", expectedVersion: 2 })
+      .expect(200);
+
+    const feed = await board
+      .get(`/api/companies/${company.id}/case-events?types=review_decided`)
+      .expect(200);
+    expect(feed.body.items).toHaveLength(2);
+    expect(feed.body.items.every((item: { type: string }) => item.type === "review_decided")).toBe(true);
+    // Newest first.
+    expect(feed.body.items[0].case.caseKey).toBe("second");
+    expect(feed.body.items[0].payload).toMatchObject({ decision: "request_changes", reason: "Tighten the intro" });
+    expect(feed.body.items[0].pipeline).toMatchObject({ id: pipeline.id, name: "Content production" });
+    expect(feed.body.items[0].fromStage).toMatchObject({ key: "review_human" });
+    expect(feed.body.items[0].toStage).toMatchObject({ key: "drafting" });
+    expect(feed.body.items[1].case.caseKey).toBe("first");
+    expect(feed.body.items[1].payload).toMatchObject({ decision: "reject", reason: "Wrong audience" });
+
+    const page = await board
+      .get(`/api/companies/${company.id}/case-events?types=review_decided,transition_forced&limit=1`)
+      .expect(200);
+    expect(page.body.items).toHaveLength(1);
+    expect(page.body.pagination).toMatchObject({ limit: 1, offset: 0, nextOffset: 1, hasMore: true });
+    const nextPage = await board
+      .get(`/api/companies/${company.id}/case-events?types=review_decided&limit=1&offset=1`)
+      .expect(200);
+    expect(nextPage.body.items[0].case.caseKey).toBe("first");
+    expect(nextPage.body.pagination.hasMore).toBe(false);
+
+    await board.get(`/api/companies/${company.id}/case-events?types=Bad%20Type!`).expect(400);
+    await board.get(`/api/companies/${company.id}/case-events?limit=0`).expect(400);
+  });
+
+  it("returns the recursive children tree with per-node rollups and a flat summary", async () => {
+    const company = await seedCompany();
+    const board = request(app(boardActor));
+    const release = await seedPipeline(board, company.id, "release", "Release train");
+    const content = await seedPipeline(board, company.id, "content", "Content production");
+
+    const parent = await board
+      .post(`/api/pipelines/${release.id}/cases`)
+      .send({ caseKey: "launch", title: "Launch v2" })
+      .expect(201);
+    const doneChild = await board
+      .post(`/api/pipelines/${content.id}/cases`)
+      .send({ caseKey: "blog", title: "Blog post", parentCaseId: parent.body.case.id })
+      .expect(201);
+    await board
+      .post(`/api/cases/${doneChild.body.case.id}/transition`)
+      .send({ toStageKey: "done", expectedVersion: 1 })
+      .expect(200);
+    const openChild = await board
+      .post(`/api/pipelines/${content.id}/cases`)
+      .send({ caseKey: "video", title: "Video", parentCaseId: parent.body.case.id })
+      .expect(201);
+    const droppedGrandchild = await board
+      .post(`/api/pipelines/${content.id}/cases`)
+      .send({ caseKey: "short", title: "Short clip", parentCaseId: openChild.body.case.id })
+      .expect(201);
+    await board
+      .post(`/api/cases/${droppedGrandchild.body.case.id}/transition`)
+      .send({ toStageKey: "cancelled", expectedVersion: 1 })
+      .expect(200);
+
+    const tree = await board.get(`/api/cases/${parent.body.case.id}/children`).expect(200);
+    expect(tree.body.rollup).toEqual({ total: 3, done: 1, dropped: 1, inMotion: 1 });
+    expect(tree.body.truncated).toBe(false);
+    expect(tree.body.totalNodes).toBe(4);
+    expect(tree.body.case.pipeline).toMatchObject({ id: release.id, name: "Release train" });
+    expect(tree.body.childGroups).toHaveLength(1);
+    const group = tree.body.childGroups[0];
+    expect(group.pipeline).toMatchObject({ id: content.id, name: "Content production" });
+    expect(group.cases.map((node: { caseKey: string }) => node.caseKey)).toEqual(["blog", "video"]);
+    const blogNode = group.cases[0];
+    expect(blogNode.terminalKind).toBe("done");
+    expect(blogNode.stage).toMatchObject({ key: "done", kind: "done" });
+    expect(blogNode.rollup).toEqual({ total: 0, done: 0, dropped: 0, inMotion: 0 });
+    const videoNode = group.cases[1];
+    expect(videoNode.rollup).toEqual({ total: 1, done: 0, dropped: 1, inMotion: 0 });
+    expect(videoNode.childGroups[0].cases[0]).toMatchObject({ caseKey: "short", terminalKind: "cancelled" });
+
+    const detail = await board.get(`/api/cases/${parent.body.case.id}`).expect(200);
+    expect(detail.body.childrenSummary).toMatchObject({ total: 2, done: 1, dropped: 0, inMotion: 1 });
+
+    await board.get(`/api/cases/${randomUUID()}/children`).expect(404);
+  });
+
+  it("enriches case detail and pipeline case lists with activeWork", async () => {
+    const company = await seedCompany();
+    const agent = await seedAgent(company.id, "Working Agent");
+    const board = request(app(boardActor));
+    const pipeline = await seedPipeline(board, company.id, "content", "Content production");
+
+    const idleCase = await board
+      .post(`/api/pipelines/${pipeline.id}/cases`)
+      .send({ caseKey: "idle", title: "Idle item" })
+      .expect(201);
+    const workedCase = await board
+      .post(`/api/pipelines/${pipeline.id}/cases`)
+      .send({ caseKey: "worked", title: "Worked item" })
+      .expect(201);
+
+    const [todoIssue] = await db.insert(issues).values({
+      companyId: company.id,
+      title: "Not started yet",
+      status: "todo",
+      priority: "medium",
+      assigneeAgentId: agent.id,
+    }).returning();
+    const [activeIssue] = await db.insert(issues).values({
+      companyId: company.id,
+      title: "Write the draft",
+      identifier: "AGG-1",
+      status: "in_progress",
+      priority: "medium",
+      assigneeAgentId: agent.id,
+      startedAt: new Date("2026-06-10T12:00:00Z"),
+    }).returning();
+    await board
+      .post(`/api/cases/${workedCase.body.case.id}/issue-links`)
+      .send({ issueId: todoIssue!.id, role: "work" })
+      .expect(201);
+    await board
+      .post(`/api/cases/${workedCase.body.case.id}/issue-links`)
+      .send({ issueId: activeIssue!.id, role: "work" })
+      .expect(201);
+
+    const detail = await board.get(`/api/cases/${workedCase.body.case.id}`).expect(200);
+    expect(detail.body.activeWork).toMatchObject({
+      issueId: activeIssue!.id,
+      issueIdentifier: "AGG-1",
+      issueTitle: "Write the draft",
+      agentId: agent.id,
+      agentName: "Working Agent",
+    });
+    expect(detail.body.activeWork.startedAt).toBeTruthy();
+
+    const idleDetail = await board.get(`/api/cases/${idleCase.body.case.id}`).expect(200);
+    expect(idleDetail.body.activeWork).toBeNull();
+
+    const list = await board.get(`/api/pipelines/${pipeline.id}/cases`).expect(200);
+    const listedWorked = list.body.find((row: { case: { id: string } }) => row.case.id === workedCase.body.case.id);
+    const listedIdle = list.body.find((row: { case: { id: string } }) => row.case.id === idleCase.body.case.id);
+    expect(listedWorked.activeWork).toMatchObject({ issueId: activeIssue!.id, agentName: "Working Agent" });
+    expect(listedIdle.activeWork).toBeNull();
+  });
+
+  it("derives pipeline connections from observed cross-pipeline case parentage", async () => {
+    const company = await seedCompany();
+    const board = request(app(boardActor));
+    const release = await seedPipeline(board, company.id, "release", "Release train");
+    const content = await seedPipeline(board, company.id, "content", "Content production");
+    const assets = await seedPipeline(board, company.id, "assets", "Asset production");
+    const lonely = await seedPipeline(board, company.id, "lonely", "Unconnected");
+
+    const releaseCase = await board
+      .post(`/api/pipelines/${release.id}/cases`)
+      .send({ caseKey: "launch", title: "Launch" })
+      .expect(201);
+    const contentCase = await board
+      .post(`/api/pipelines/${content.id}/cases`)
+      .send({ caseKey: "blog", title: "Blog", parentCaseId: releaseCase.body.case.id })
+      .expect(201);
+    await board
+      .post(`/api/pipelines/${assets.id}/cases`)
+      .send({ caseKey: "thumb", title: "Thumbnail", parentCaseId: contentCase.body.case.id })
+      .expect(201);
+    // Same-pipeline parentage must not create a self connection.
+    await board
+      .post(`/api/pipelines/${content.id}/cases`)
+      .send({ caseKey: "blog-child", title: "Blog child", parentCaseId: contentCase.body.case.id })
+      .expect(201);
+
+    const listed = await board.get(`/api/companies/${company.id}/pipelines`).expect(200);
+    const byId = new Map(listed.body.map((row: { id: string }) => [row.id, row]));
+    expect((byId.get(release.id) as { connections: unknown }).connections).toEqual({
+      upstreamPipelineIds: [],
+      downstreamPipelineIds: [content.id],
+    });
+    expect((byId.get(content.id) as { connections: unknown }).connections).toEqual({
+      upstreamPipelineIds: [release.id],
+      downstreamPipelineIds: [assets.id],
+    });
+    expect((byId.get(assets.id) as { connections: unknown }).connections).toEqual({
+      upstreamPipelineIds: [content.id],
+      downstreamPipelineIds: [],
+    });
+    expect((byId.get(lonely.id) as { connections: unknown }).connections).toEqual({
+      upstreamPipelineIds: [],
+      downstreamPipelineIds: [],
+    });
+  });
+});

server/src/routes/pipelines.ts

@@ -27,6 +27,20 @@ import {
   type PipelineStageConfig,
   type PipelineStageKind,
 } from "../services/pipelines.js";
+import {
+  COMPANY_CASE_EVENTS_DEFAULT_LIMIT,
+  COMPANY_CASE_EVENTS_MAX_LIMIT,
+  COMPANY_CASE_EVENTS_MAX_TYPES,
+  getCaseChildrenTree,
+  getDirectChildrenSummary,
+  listCompanyCaseEvents,
+  listPipelineAttention,
+  loadActiveWorkForCases,
+  loadPipelineConnections,
+  PIPELINE_ATTENTION_DEFAULT_LIMIT,
+  PIPELINE_ATTENTION_MAX_LIMIT,
+  type AttentionCaller,
+} from "../services/pipelines-aggregation.js";
 import { assertCompanyAccess } from "./authz.js";
 
 const stageKindSchema = z.enum(["open", "working", "review", "done", "cancelled"]);
@@ -184,6 +198,34 @@ function actorForMutation(req: Request): PipelineActor {
   throw unauthorized();
 }
 
+function attentionCallerFor(req: Request): AttentionCaller {
+  if (req.actor.type === "agent") {
+    if (!req.actor.agentId) throw unauthorized();
+    return { type: "agent", agentId: req.actor.agentId };
+  }
+  if (req.actor.type === "board") {
+    return { type: "user", userId: req.actor.userId ?? "board" };
+  }
+  throw unauthorized();
+}
+
+function parseEventTypesQuery(value: unknown): string[] | undefined {
+  if (value === undefined) return undefined;
+  const raw = Array.isArray(value) ? value : [value];
+  const types = raw
+    .flatMap((item) => String(item).split(","))
+    .map((item) => item.trim())
+    .filter((item) => item.length > 0);
+  if (types.length === 0) return undefined;
+  if (types.length > COMPANY_CASE_EVENTS_MAX_TYPES) {
+    throw badRequest(`types accepts at most ${COMPANY_CASE_EVENTS_MAX_TYPES} values`);
+  }
+  for (const type of types) {
+    if (!/^[a-z_]{1,64}$/.test(type)) throw badRequest(`Invalid event type: ${type}`);
+  }
+  return [...new Set(types)];
+}
+
 function parseOptionalNonNegativeInteger(value: unknown, name: string) {
   if (value === undefined) return null;
   if (Array.isArray(value)) throw badRequest(`${name} must be a single integer`);
@@ -287,7 +329,34 @@ export function pipelineRoutes(db: Db, options: Parameters<typeof pipelineServic
       .where(eq(pipelines.companyId, companyId))
       .groupBy(pipelines.id)
       .orderBy(asc(pipelines.createdAt));
-    res.json(rows.map((row) => ({ ...row.pipeline, stageCount: row.stageCount, openCaseCount: row.openCaseCount })));
+    const connections = await loadPipelineConnections(db, companyId);
+    res.json(rows.map((row) => ({
+      ...row.pipeline,
+      stageCount: row.stageCount,
+      openCaseCount: row.openCaseCount,
+      connections: connections.get(row.pipeline.id) ?? { upstreamPipelineIds: [], downstreamPipelineIds: [] },
+    })));
+  });
+
+  router.get("/companies/:companyId/pipelines-attention", async (req, res) => {
+    const companyId = req.params.companyId as string;
+    assertPipelineCompanyAccess(req, companyId);
+    const caller = attentionCallerFor(req);
+    const requestedLimit = parseOptionalNonNegativeInteger(req.query.limit, "limit");
+    if (requestedLimit === 0) throw badRequest("limit must be a positive integer");
+    const limit = Math.min(requestedLimit ?? PIPELINE_ATTENTION_DEFAULT_LIMIT, PIPELINE_ATTENTION_MAX_LIMIT);
+    res.json(await listPipelineAttention(db, { companyId, caller, limit }));
+  });
+
+  router.get("/companies/:companyId/case-events", async (req, res) => {
+    const companyId = req.params.companyId as string;
+    assertPipelineCompanyAccess(req, companyId);
+    const types = parseEventTypesQuery(req.query.types);
+    const requestedLimit = parseOptionalNonNegativeInteger(req.query.limit, "limit");
+    if (requestedLimit === 0) throw badRequest("limit must be a positive integer");
+    const limit = Math.min(requestedLimit ?? COMPANY_CASE_EVENTS_DEFAULT_LIMIT, COMPANY_CASE_EVENTS_MAX_LIMIT);
+    const offset = parseOptionalNonNegativeInteger(req.query.offset, "offset") ?? 0;
+    res.json(await listCompanyCaseEvents(db, { companyId, types, limit, offset }));
   });
 
   router.post("/companies/:companyId/pipelines", validate(createPipelineSchema), async (req, res) => {
@@ -557,7 +626,8 @@ export function pipelineRoutes(db: Db, options: Parameters<typeof pipelineServic
         q ? or(ilike(pipelineCases.title, `%${q}%`), ilike(pipelineCases.summary, `%${q}%`)) : undefined,
       ))
       .orderBy(asc(pipelineCases.createdAt));
-    res.json(rows);
+    const activeWork = await loadActiveWorkForCases(db, companyId, rows.map((row) => row.case.id));
+    res.json(rows.map((row) => ({ ...row, activeWork: activeWork.get(row.case.id) ?? null })));
   });
 
   router.get("/cases/:caseId", async (req, res) => {
@@ -792,6 +862,12 @@ export function pipelineRoutes(db: Db, options: Parameters<typeof pipelineServic
     res.json(await svc.listCaseEventsPage(companyId, caseId, pagination));
   });
 
+  router.get("/cases/:caseId/children", async (req, res) => {
+    const caseId = req.params.caseId as string;
+    const companyId = await assertCaseAccess(db, req, caseId);
+    res.json(await getCaseChildrenTree(db, companyId, caseId));
+  });
+
   router.get("/cases/:caseId/rollup", async (req, res) => {
     const caseId = req.params.caseId as string;
     const companyId = await assertCaseAccess(db, req, caseId);
@@ -846,12 +922,13 @@ async function getCaseDetail(db: Db, companyId: string, caseId: string) {
     .limit(1)
     .then((rows) => rows[0] ?? null);
   if (!row) throw notFound("Pipeline case not found");
-  const [allowedNextStages, links, blockers, blocks, children] = await Promise.all([
+  const [allowedNextStages, links, blockers, blocks, childrenCounts, activeWorkByCase] = await Promise.all([
     db.select().from(pipelineStages).where(eq(pipelineStages.pipelineId, row.case.pipelineId)).orderBy(asc(pipelineStages.position)),
     db.select().from(pipelineCaseIssueLinks).where(and(eq(pipelineCaseIssueLinks.companyId, companyId), eq(pipelineCaseIssueLinks.caseId, caseId))),
     db.select().from(pipelineCaseBlockers).where(and(eq(pipelineCaseBlockers.companyId, companyId), eq(pipelineCaseBlockers.caseId, caseId))),
     db.select().from(pipelineCaseBlockers).where(and(eq(pipelineCaseBlockers.companyId, companyId), eq(pipelineCaseBlockers.blockedByCaseId, caseId))),
-    db.select().from(pipelineCases).where(and(eq(pipelineCases.companyId, companyId), eq(pipelineCases.parentCaseId, caseId))),
+    getDirectChildrenSummary(db, companyId, caseId),
+    loadActiveWorkForCases(db, companyId, [caseId]),
   ]);
   return {
     ...row,
@@ -862,8 +939,10 @@ async function getCaseDetail(db: Db, companyId: string, caseId: string) {
     childrenSummary: {
       childCount: row.case.childCount,
       terminalChildCount: row.case.terminalChildCount,
-      loadedChildren: children.length,
+      loadedChildren: childrenCounts.total,
+      ...childrenCounts,
     },
+    activeWork: activeWorkByCase.get(caseId) ?? null,
     pendingSuggestion: row.case.pendingSuggestion,
   };
 }

server/src/services/pipelines-aggregation.ts

@@ -0,0 +1,543 @@
+import { and, asc, desc, eq, inArray, isNotNull, isNull, ne, sql } from "drizzle-orm";
+import { alias } from "drizzle-orm/pg-core";
+import type { Db } from "@paperclipai/db";
+import {
+  agents,
+  issues,
+  pipelineCaseEvents,
+  pipelineCaseIssueLinks,
+  pipelineCases,
+  pipelineStages,
+  pipelines,
+} from "@paperclipai/db";
+import { notFound } from "../errors.js";
+
+export const PIPELINE_ATTENTION_DEFAULT_LIMIT = 50;
+export const PIPELINE_ATTENTION_MAX_LIMIT = 100;
+export const COMPANY_CASE_EVENTS_DEFAULT_LIMIT = 50;
+export const COMPANY_CASE_EVENTS_MAX_LIMIT = 100;
+export const COMPANY_CASE_EVENTS_MAX_TYPES = 10;
+export const CASE_CHILDREN_TREE_MAX_NODES = 1_000;
+export const CASE_CHILDREN_TREE_MAX_DEPTH = 10;
+
+export type AttentionCaller =
+  | { type: "user"; userId: string }
+  | { type: "agent"; agentId: string };
+
+type CaseRow = typeof pipelineCases.$inferSelect;
+type StageRow = typeof pipelineStages.$inferSelect;
+type PipelineRow = typeof pipelines.$inferSelect;
+
+export type ActiveWork = {
+  issueId: string;
+  issueIdentifier: string | null;
+  issueTitle: string;
+  agentId: string;
+  agentName: string;
+  startedAt: Date | null;
+};
+
+function caseDisplay(row: { case: CaseRow; stage: StageRow; pipeline: PipelineRow }) {
+  return {
+    id: row.case.id,
+    caseKey: row.case.caseKey,
+    title: row.case.title,
+    summary: row.case.summary,
+    version: row.case.version,
+    terminalKind: row.case.terminalKind,
+    parentCaseId: row.case.parentCaseId,
+    updatedAt: row.case.updatedAt,
+    createdAt: row.case.createdAt,
+    pipeline: { id: row.pipeline.id, key: row.pipeline.key, name: row.pipeline.name },
+    stage: { id: row.stage.id, key: row.stage.key, name: row.stage.name, kind: row.stage.kind },
+  };
+}
+
+// Current review-stage approver semantics: reviewerKind "human" awaits any
+// human caller, "any" awaits anyone. B1 (stage requireApproval + named
+// approver) extends this predicate when it lands — keep the seam here.
+function reviewStageAwaitsCaller(config: Record<string, unknown>, caller: AttentionCaller) {
+  const reviewerKind = typeof config.reviewerKind === "string" ? config.reviewerKind : "human";
+  if (reviewerKind === "any") return true;
+  return caller.type === "user";
+}
+
+function boundedLimit(limit: number | undefined, fallback: number, max: number) {
+  return Math.min(max, Math.max(1, Math.floor(limit ?? fallback)));
+}
+
+export async function listPipelineAttention(
+  db: Db,
+  input: { companyId: string; caller: AttentionCaller; limit?: number },
+) {
+  const limit = boundedLimit(input.limit, PIPELINE_ATTENTION_DEFAULT_LIMIT, PIPELINE_ATTENTION_MAX_LIMIT);
+
+  const suggestionAgent = alias(agents, "suggestion_agent");
+  const suggestionToStage = alias(pipelineStages, "suggestion_to_stage");
+  const suggestionRows = await db
+    .select({
+      case: pipelineCases,
+      pipeline: pipelines,
+      stage: pipelineStages,
+      toStage: suggestionToStage,
+      suggestingAgent: { id: suggestionAgent.id, name: suggestionAgent.name },
+    })
+    .from(pipelineCases)
+    .innerJoin(pipelines, eq(pipelineCases.pipelineId, pipelines.id))
+    .innerJoin(pipelineStages, eq(pipelineCases.stageId, pipelineStages.id))
+    .leftJoin(suggestionToStage, and(
+      eq(suggestionToStage.pipelineId, pipelineCases.pipelineId),
+      eq(suggestionToStage.key, sql`${pipelineCases.pendingSuggestion}->>'toStageKey'`),
+    ))
+    .leftJoin(suggestionAgent, sql`${suggestionAgent.id}::text = ${pipelineCases.pendingSuggestion}->>'suggestedByAgentId'`)
+    .where(and(
+      eq(pipelineCases.companyId, input.companyId),
+      eq(pipelines.companyId, input.companyId),
+      isNull(pipelineCases.terminalKind),
+      isNotNull(pipelineCases.pendingSuggestion),
+    ))
+    .orderBy(desc(pipelineCases.updatedAt))
+    .limit(limit);
+
+  const suggestions = suggestionRows.map((row) => {
+    const suggestion = row.case.pendingSuggestion!;
+    return {
+      case: caseDisplay(row),
+      suggestion: {
+        id: suggestion.id,
+        fromStageKey: row.stage.key,
+        fromStageName: row.stage.name,
+        toStageKey: suggestion.toStageKey,
+        toStageName: row.toStage?.name ?? null,
+        rationale: suggestion.rationale,
+        confidence: suggestion.confidence ?? null,
+        createdAt: suggestion.createdAt,
+        suggestedBy: row.suggestingAgent?.id
+          ? { agentId: row.suggestingAgent.id, agentName: row.suggestingAgent.name }
+          : null,
+      },
+    };
+  });
+
+  const reviewRows = await db
+    .select({ case: pipelineCases, pipeline: pipelines, stage: pipelineStages })
+    .from(pipelineCases)
+    .innerJoin(pipelines, eq(pipelineCases.pipelineId, pipelines.id))
+    .innerJoin(pipelineStages, eq(pipelineCases.stageId, pipelineStages.id))
+    .where(and(
+      eq(pipelineCases.companyId, input.companyId),
+      eq(pipelines.companyId, input.companyId),
+      eq(pipelineStages.kind, "review"),
+      isNull(pipelineCases.terminalKind),
+    ))
+    .orderBy(asc(pipelineCases.createdAt))
+    .limit(PIPELINE_ATTENTION_MAX_LIMIT * 2);
+
+  const reviews = reviewRows
+    .filter((row) => reviewStageAwaitsCaller((row.stage.config ?? {}) as Record<string, unknown>, input.caller))
+    .slice(0, limit)
+    .map((row) => {
+      const config = (row.stage.config ?? {}) as Record<string, unknown>;
+      return {
+        case: caseDisplay(row),
+        review: {
+          expectedVersion: row.case.version,
+          approveToStageKey: typeof config.approveToStageKey === "string" ? config.approveToStageKey : null,
+          rejectToStageKey: typeof config.rejectToStageKey === "string" ? config.rejectToStageKey : null,
+          requestChangesToStageKey: typeof config.requestChangesToStageKey === "string" ? config.requestChangesToStageKey : null,
+          requireRejectReason: config.requireRejectReason !== false,
+          reviewerKind: typeof config.reviewerKind === "string" ? config.reviewerKind : "human",
+        },
+      };
+    });
+
+  const driftRows = await db
+    .selectDistinctOn([pipelineCaseEvents.caseId], {
+      event: pipelineCaseEvents,
+      case: pipelineCases,
+      pipeline: pipelines,
+      stage: pipelineStages,
+    })
+    .from(pipelineCaseEvents)
+    .innerJoin(pipelineCases, eq(pipelineCaseEvents.caseId, pipelineCases.id))
+    .innerJoin(pipelines, eq(pipelineCases.pipelineId, pipelines.id))
+    .innerJoin(pipelineStages, eq(pipelineCases.stageId, pipelineStages.id))
+    .where(and(
+      eq(pipelineCaseEvents.companyId, input.companyId),
+      eq(pipelineCaseEvents.type, "upstream_drift"),
+      eq(pipelineCases.companyId, input.companyId),
+      isNull(pipelineCases.terminalKind),
+      sql`${pipelineCaseEvents.createdAt} > ${pipelineCases.updatedAt}`,
+    ))
+    .orderBy(asc(pipelineCaseEvents.caseId), desc(pipelineCaseEvents.createdAt))
+    .limit(limit);
+
+  const driftCaseIds = driftRows.map((row) => row.case.id);
+  const upstreamCaseIds = [...new Set(driftRows
+    .map((row) => (row.event.payload as Record<string, unknown>).upstreamCaseId)
+    .filter((value): value is string => typeof value === "string"))];
+
+  const [activeWorkByCase, workIssuesByCase, upstreamCases] = await Promise.all([
+    loadActiveWorkForCases(db, input.companyId, driftCaseIds),
+    loadOpenWorkIssuesForCases(db, input.companyId, driftCaseIds),
+    upstreamCaseIds.length === 0
+      ? Promise.resolve([] as Array<{ case: CaseRow; pipeline: PipelineRow }>)
+      : db
+        .select({ case: pipelineCases, pipeline: pipelines })
+        .from(pipelineCases)
+        .innerJoin(pipelines, eq(pipelineCases.pipelineId, pipelines.id))
+        .where(and(eq(pipelineCases.companyId, input.companyId), inArray(pipelineCases.id, upstreamCaseIds))),
+  ]);
+  const upstreamById = new Map(upstreamCases.map((row) => [row.case.id, row]));
+
+  const headsUp = driftRows.map((row) => {
+    const payload = row.event.payload as Record<string, unknown>;
+    const upstream = typeof payload.upstreamCaseId === "string" ? upstreamById.get(payload.upstreamCaseId) : undefined;
+    return {
+      case: caseDisplay(row),
+      drift: {
+        eventId: row.event.id,
+        createdAt: row.event.createdAt,
+        previousVersion: typeof payload.previousVersion === "number" ? payload.previousVersion : null,
+        version: typeof payload.version === "number" ? payload.version : null,
+        upstream: upstream
+          ? {
+            caseId: upstream.case.id,
+            caseKey: upstream.case.caseKey,
+            title: upstream.case.title,
+            pipelineId: upstream.pipeline.id,
+            pipelineName: upstream.pipeline.name,
+          }
+          : {
+            caseId: typeof payload.upstreamCaseId === "string" ? payload.upstreamCaseId : null,
+            caseKey: typeof payload.upstreamCaseKey === "string" ? payload.upstreamCaseKey : null,
+            title: null,
+            pipelineId: typeof payload.upstreamPipelineId === "string" ? payload.upstreamPipelineId : null,
+            pipelineName: null,
+          },
+      },
+      activeWork: activeWorkByCase.get(row.case.id) ?? null,
+      workIssue: workIssuesByCase.get(row.case.id) ?? null,
+    };
+  });
+
+  return {
+    suggestions,
+    reviews,
+    headsUp,
+    counts: { suggestions: suggestions.length, reviews: reviews.length, headsUp: headsUp.length },
+  };
+}
+
+export async function listCompanyCaseEvents(
+  db: Db,
+  input: { companyId: string; types?: string[]; limit?: number; offset?: number },
+) {
+  const limit = boundedLimit(input.limit, COMPANY_CASE_EVENTS_DEFAULT_LIMIT, COMPANY_CASE_EVENTS_MAX_LIMIT);
+  const offset = Math.max(0, Math.floor(input.offset ?? 0));
+  const fromStage = alias(pipelineStages, "from_stage");
+  const toStage = alias(pipelineStages, "to_stage");
+  const actorAgent = alias(agents, "actor_agent");
+
+  const rows = await db
+    .select({
+      event: pipelineCaseEvents,
+      case: {
+        id: pipelineCases.id,
+        caseKey: pipelineCases.caseKey,
+        title: pipelineCases.title,
+        terminalKind: pipelineCases.terminalKind,
+      },
+      pipeline: { id: pipelines.id, key: pipelines.key, name: pipelines.name },
+      fromStage: { id: fromStage.id, key: fromStage.key, name: fromStage.name, kind: fromStage.kind },
+      toStage: { id: toStage.id, key: toStage.key, name: toStage.name, kind: toStage.kind },
+      actorAgent: { id: actorAgent.id, name: actorAgent.name },
+    })
+    .from(pipelineCaseEvents)
+    .innerJoin(pipelineCases, eq(pipelineCaseEvents.caseId, pipelineCases.id))
+    .innerJoin(pipelines, eq(pipelineCases.pipelineId, pipelines.id))
+    .leftJoin(fromStage, eq(pipelineCaseEvents.fromStageId, fromStage.id))
+    .leftJoin(toStage, eq(pipelineCaseEvents.toStageId, toStage.id))
+    .leftJoin(actorAgent, eq(pipelineCaseEvents.actorAgentId, actorAgent.id))
+    .where(and(
+      eq(pipelineCaseEvents.companyId, input.companyId),
+      eq(pipelineCases.companyId, input.companyId),
+      input.types && input.types.length > 0 ? inArray(pipelineCaseEvents.type, input.types) : undefined,
+    ))
+    .orderBy(desc(pipelineCaseEvents.createdAt), desc(pipelineCaseEvents.id))
+    .limit(limit + 1)
+    .offset(offset);
+
+  const hasMore = rows.length > limit;
+  const items = (hasMore ? rows.slice(0, limit) : rows).map((row) => ({
+    ...row.event,
+    case: row.case,
+    pipeline: row.pipeline,
+    fromStage: row.fromStage?.id ? row.fromStage : null,
+    toStage: row.toStage?.id ? row.toStage : null,
+    actorAgent: row.actorAgent?.id ? row.actorAgent : null,
+  }));
+
+  return {
+    items,
+    pagination: { limit, offset, nextOffset: hasMore ? offset + limit : null, hasMore },
+  };
+}
+
+export type CaseChildrenRollup = { total: number; done: number; dropped: number; inMotion: number };
+
+type SubtreeRow = {
+  id: string;
+  parent_case_id: string | null;
+  pipeline_id: string;
+  stage_id: string;
+  case_key: string;
+  title: string;
+  terminal_kind: string | null;
+  created_at: string | Date;
+  updated_at: string | Date;
+  depth: number;
+};
+
+export type CaseChildNode = {
+  id: string;
+  caseKey: string;
+  title: string;
+  terminalKind: string | null;
+  createdAt: Date;
+  updatedAt: Date;
+  pipeline: { id: string; key: string; name: string };
+  stage: { id: string; key: string; name: string; kind: string };
+  rollup: CaseChildrenRollup;
+  childGroups: Array<{ pipeline: { id: string; key: string; name: string }; cases: CaseChildNode[] }>;
+};
+
+export async function getCaseChildrenTree(db: Db, companyId: string, caseId: string) {
+  const result = await db.execute(sql`
+    with recursive subtree as (
+      select id, parent_case_id, pipeline_id, stage_id, case_key, title, terminal_kind, created_at, updated_at, 0 as depth
+      from pipeline_cases
+      where company_id = ${companyId} and id = ${caseId}
+      union all
+      select child.id, child.parent_case_id, child.pipeline_id, child.stage_id, child.case_key, child.title,
+             child.terminal_kind, child.created_at, child.updated_at, parent.depth + 1
+      from pipeline_cases child
+      join subtree parent on child.parent_case_id = parent.id
+      where child.company_id = ${companyId} and parent.depth < ${CASE_CHILDREN_TREE_MAX_DEPTH}
+    )
+    select * from subtree limit ${CASE_CHILDREN_TREE_MAX_NODES + 1}
+  `);
+  const rows = Array.from(result) as SubtreeRow[];
+  if (rows.length === 0) throw notFound("Pipeline case not found");
+  const truncated = rows.length > CASE_CHILDREN_TREE_MAX_NODES;
+  const bounded = truncated ? rows.slice(0, CASE_CHILDREN_TREE_MAX_NODES) : rows;
+
+  const pipelineIds = [...new Set(bounded.map((row) => row.pipeline_id))];
+  const stageIds = [...new Set(bounded.map((row) => row.stage_id))];
+  const [pipelineRows, stageRows] = await Promise.all([
+    db.select({ id: pipelines.id, key: pipelines.key, name: pipelines.name })
+      .from(pipelines)
+      .where(and(eq(pipelines.companyId, companyId), inArray(pipelines.id, pipelineIds))),
+    db.select({ id: pipelineStages.id, key: pipelineStages.key, name: pipelineStages.name, kind: pipelineStages.kind })
+      .from(pipelineStages)
+      .where(inArray(pipelineStages.id, stageIds)),
+  ]);
+  const pipelineById = new Map(pipelineRows.map((row) => [row.id, row]));
+  const stageById = new Map(stageRows.map((row) => [row.id, row]));
+
+  const nodeById = new Map<string, CaseChildNode>();
+  const childRowsByParent = new Map<string, SubtreeRow[]>();
+  for (const row of bounded) {
+    if (row.id !== caseId && row.parent_case_id) {
+      const list = childRowsByParent.get(row.parent_case_id) ?? [];
+      list.push(row);
+      childRowsByParent.set(row.parent_case_id, list);
+    }
+  }
+
+  function buildNode(row: SubtreeRow): CaseChildNode {
+    const childRows = (childRowsByParent.get(row.id) ?? [])
+      .sort((a, b) => new Date(a.created_at).getTime() - new Date(b.created_at).getTime());
+    const children = childRows.map(buildNode);
+    const rollup: CaseChildrenRollup = { total: 0, done: 0, dropped: 0, inMotion: 0 };
+    for (const child of children) {
+      rollup.total += 1 + child.rollup.total;
+      rollup.done += (child.terminalKind === "done" ? 1 : 0) + child.rollup.done;
+      rollup.dropped += (child.terminalKind === "cancelled" ? 1 : 0) + child.rollup.dropped;
+      rollup.inMotion += (child.terminalKind === null ? 1 : 0) + child.rollup.inMotion;
+    }
+    const pipeline = pipelineById.get(row.pipeline_id) ?? { id: row.pipeline_id, key: "", name: "" };
+    const groups = new Map<string, { pipeline: { id: string; key: string; name: string }; cases: CaseChildNode[] }>();
+    for (const child of children) {
+      const group = groups.get(child.pipeline.id) ?? { pipeline: child.pipeline, cases: [] };
+      group.cases.push(child);
+      groups.set(child.pipeline.id, group);
+    }
+    const childGroups = [...groups.values()].sort((a, b) => {
+      if (a.pipeline.id === row.pipeline_id && b.pipeline.id !== row.pipeline_id) return -1;
+      if (b.pipeline.id === row.pipeline_id && a.pipeline.id !== row.pipeline_id) return 1;
+      return a.pipeline.name.localeCompare(b.pipeline.name);
+    });
+    const node: CaseChildNode = {
+      id: row.id,
+      caseKey: row.case_key,
+      title: row.title,
+      terminalKind: row.terminal_kind,
+      createdAt: new Date(row.created_at),
+      updatedAt: new Date(row.updated_at),
+      pipeline,
+      stage: stageById.get(row.stage_id) ?? { id: row.stage_id, key: "", name: "", kind: "" },
+      rollup,
+      childGroups,
+    };
+    nodeById.set(row.id, node);
+    return node;
+  }
+
+  const rootRow = bounded.find((row) => row.id === caseId);
+  if (!rootRow) throw notFound("Pipeline case not found");
+  const root = buildNode(rootRow);
+
+  return {
+    case: root,
+    rollup: root.rollup,
+    childGroups: root.childGroups,
+    truncated,
+    totalNodes: bounded.length,
+  };
+}
+
+export async function getDirectChildrenSummary(
+  db: Db,
+  companyId: string,
+  caseId: string,
+): Promise<CaseChildrenRollup> {
+  const [counts] = await db
+    .select({
+      total: sql<number>`count(*)::int`,
+      done: sql<number>`count(*) filter (where ${pipelineCases.terminalKind} = 'done')::int`,
+      dropped: sql<number>`count(*) filter (where ${pipelineCases.terminalKind} = 'cancelled')::int`,
+      inMotion: sql<number>`count(*) filter (where ${pipelineCases.terminalKind} is null)::int`,
+    })
+    .from(pipelineCases)
+    .where(and(eq(pipelineCases.companyId, companyId), eq(pipelineCases.parentCaseId, caseId)));
+  return counts ?? { total: 0, done: 0, dropped: 0, inMotion: 0 };
+}
+
+export async function loadActiveWorkForCases(
+  db: Db,
+  companyId: string,
+  caseIds: string[],
+): Promise<Map<string, ActiveWork | null>> {
+  const map = new Map<string, ActiveWork | null>(caseIds.map((id) => [id, null]));
+  if (caseIds.length === 0) return map;
+  const rows = await db
+    .select({
+      caseId: pipelineCaseIssueLinks.caseId,
+      issueId: issues.id,
+      issueIdentifier: issues.identifier,
+      issueTitle: issues.title,
+      agentId: issues.assigneeAgentId,
+      agentName: agents.name,
+      startedAt: issues.startedAt,
+      issueUpdatedAt: issues.updatedAt,
+    })
+    .from(pipelineCaseIssueLinks)
+    .innerJoin(issues, eq(pipelineCaseIssueLinks.issueId, issues.id))
+    .innerJoin(agents, eq(issues.assigneeAgentId, agents.id))
+    .where(and(
+      eq(pipelineCaseIssueLinks.companyId, companyId),
+      inArray(pipelineCaseIssueLinks.caseId, caseIds),
+      inArray(pipelineCaseIssueLinks.role, ["work", "automation"]),
+      eq(issues.companyId, companyId),
+      eq(issues.status, "in_progress"),
+      isNull(issues.hiddenAt),
+    ))
+    .orderBy(desc(issues.updatedAt));
+  for (const row of rows) {
+    if (map.get(row.caseId)) continue;
+    map.set(row.caseId, {
+      issueId: row.issueId,
+      issueIdentifier: row.issueIdentifier,
+      issueTitle: row.issueTitle,
+      agentId: row.agentId!,
+      agentName: row.agentName,
+      startedAt: row.startedAt ?? row.issueUpdatedAt,
+    });
+  }
+  return map;
+}
+
+async function loadOpenWorkIssuesForCases(db: Db, companyId: string, caseIds: string[]) {
+  const map = new Map<string, { issueId: string; issueIdentifier: string | null; title: string; status: string }>();
+  if (caseIds.length === 0) return map;
+  const rows = await db
+    .select({
+      caseId: pipelineCaseIssueLinks.caseId,
+      issueId: issues.id,
+      issueIdentifier: issues.identifier,
+      title: issues.title,
+      status: issues.status,
+    })
+    .from(pipelineCaseIssueLinks)
+    .innerJoin(issues, eq(pipelineCaseIssueLinks.issueId, issues.id))
+    .where(and(
+      eq(pipelineCaseIssueLinks.companyId, companyId),
+      inArray(pipelineCaseIssueLinks.caseId, caseIds),
+      eq(pipelineCaseIssueLinks.role, "work"),
+      eq(issues.companyId, companyId),
+      ne(issues.status, "done"),
+      ne(issues.status, "cancelled"),
+      isNull(issues.hiddenAt),
+    ))
+    .orderBy(desc(issues.updatedAt));
+  for (const row of rows) {
+    if (map.has(row.caseId)) continue;
+    map.set(row.caseId, {
+      issueId: row.issueId,
+      issueIdentifier: row.issueIdentifier,
+      title: row.title,
+      status: row.status,
+    });
+  }
+  return map;
+}
+
+export type PipelineConnections = { upstreamPipelineIds: string[]; downstreamPipelineIds: string[] };
+
+export async function loadPipelineConnections(
+  db: Db,
+  companyId: string,
+): Promise<Map<string, PipelineConnections>> {
+  const parentCase = alias(pipelineCases, "parent_case");
+  const rows = await db
+    .selectDistinct({
+      parentPipelineId: parentCase.pipelineId,
+      childPipelineId: pipelineCases.pipelineId,
+    })
+    .from(pipelineCases)
+    .innerJoin(parentCase, eq(pipelineCases.parentCaseId, parentCase.id))
+    .where(and(
+      eq(pipelineCases.companyId, companyId),
+      eq(parentCase.companyId, companyId),
+      ne(pipelineCases.pipelineId, parentCase.pipelineId),
+    ));
+  const map = new Map<string, PipelineConnections>();
+  const entry = (pipelineId: string) => {
+    let value = map.get(pipelineId);
+    if (!value) {
+      value = { upstreamPipelineIds: [], downstreamPipelineIds: [] };
+      map.set(pipelineId, value);
+    }
+    return value;
+  };
+  for (const row of rows) {
+    entry(row.childPipelineId).upstreamPipelineIds.push(row.parentPipelineId);
+    entry(row.parentPipelineId).downstreamPipelineIds.push(row.childPipelineId);
+  }
+  for (const value of map.values()) {
+    value.upstreamPipelineIds.sort();
+    value.downstreamPipelineIds.sort();
+  }
+  return map;
+}

server/src/services/pipelines.ts

@@ -617,26 +617,39 @@ async function notifyDependentWorkIssuesOfUpstreamContentChange(
     version: number;
   },
 ) {
-  const rows = await db
-    .selectDistinctOn([issues.id], { issueId: issues.id })
+  const dependents = await db
+    .select({ dependentCase: pipelineCases })
     .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),
+    ));
+
+  if (dependents.length === 0) return;
+
+  const dependentCaseIds = dependents.map((row) => row.dependentCase.id);
+  const linkRows = await db
+    .select({ caseId: pipelineCaseIssueLinks.caseId, issueId: issues.id })
+    .from(pipelineCaseIssueLinks)
+    .innerJoin(issues, eq(pipelineCaseIssueLinks.issueId, issues.id))
+    .where(and(
       eq(pipelineCaseIssueLinks.companyId, input.companyId),
+      inArray(pipelineCaseIssueLinks.caseId, dependentCaseIds),
       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 issueIdsByCase = new Map<string, string[]>();
+  for (const row of linkRows) {
+    const list = issueIdsByCase.get(row.caseId) ?? [];
+    list.push(row.issueId);
+    issueIdsByCase.set(row.caseId, list);
+  }
 
   const upstreamLink = buildCaseDeepLink({
     pipelineId: input.upstreamCase.pipelineId,
@@ -644,14 +657,36 @@ async function notifyDependentWorkIssuesOfUpstreamContentChange(
   });
   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({
+  const notifiedIssueIds = new Set<string>();
+  for (const { dependentCase } of dependents) {
+    const issueIds = issueIdsByCase.get(dependentCase.id) ?? [];
+    for (const issueId of issueIds) {
+      if (notifiedIssueIds.has(issueId)) continue;
+      notifiedIssueIds.add(issueId);
+      await db.insert(issueComments).values({
+        companyId: input.companyId,
+        issueId,
+        authorType: "system",
+        body,
+      });
+      await db.update(issues).set({ updatedAt: nowDate() }).where(eq(issues.id, issueId));
+    }
+    // The drift event intentionally does not bump the dependent case's
+    // updatedAt: "unresolved drift" is derived as event.createdAt > case.updatedAt.
+    await writeCaseEvent(db, {
       companyId: input.companyId,
-      issueId: row.issueId,
-      authorType: "system",
-      body,
+      caseId: dependentCase.id,
+      type: "upstream_drift",
+      actor: { type: "system" },
+      payload: {
+        upstreamCaseId: input.upstreamCase.id,
+        upstreamCaseKey: input.upstreamCase.caseKey,
+        upstreamPipelineId: input.upstreamCase.pipelineId,
+        previousVersion: input.previousVersion,
+        version: input.version,
+        notifiedIssueIds: issueIds,
+      },
     });
-    await db.update(issues).set({ updatedAt: nowDate() }).where(eq(issues.id, row.issueId));
   }
 }