Merge qualified retention first-materialization guard

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

* commit 'b448320a5e5683956e83d5400cfc3a9c933fcc18':
  Validate eligibility before first attention retention materialization

server/src/__tests__/decision-retention-materialization-security.test.ts

@@ -0,0 +1,134 @@
+import { randomUUID } from "node:crypto";
+import { and, eq, inArray, sql } from "drizzle-orm";
+import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it } from "vitest";
+import { agents, authUsers, companies, companyMemberships, createDb, decisionRetention, issueRelations, issues } from "@paperclipai/db";
+import type { AttentionSourceKind } from "@paperclipai/shared";
+import { attentionService } from "../services/attention.js";
+import { decisionRetentionService } from "../services/decision-retention.js";
+import { issueService } from "../services/issues.js";
+import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
+import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
+
+const external = resolveExternalTestDatabaseUrl("PAPERCLIP_RETENTION_MATERIALIZATION_TEST_DATABASE_URL");
+const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport();
+(support.supported ? describe : describe.skip)("retention first-materialization security", () => {
+  let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
+  let db: ReturnType<typeof createDb>;
+  let companyId: string, agentId: string, userId: string;
+  const users: string[] = [];
+  beforeAll(async () => {
+    if (!external) temporary = await startEmbeddedPostgresTestDatabase("retention-security-");
+    db = createDb(external ?? temporary!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 });
+    db.$client.options.onnotice = () => {};
+  }, 30_000);
+  beforeEach(async () => {
+    companyId = randomUUID(); agentId = randomUUID(); userId = randomUUID(); users.push(userId);
+    await db.insert(authUsers).values({ id: userId, name: "Security operator", email: `${userId}@retention.test`, createdAt: new Date(), updatedAt: new Date() });
+    await db.insert(companies).values({ id: companyId, name: "Materialization", issuePrefix: `S${companyId.slice(0, 6)}`, defaultResponsibleUserId: userId });
+    await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: userId, membershipRole: "owner", status: "active" });
+    await db.insert(agents).values({ id: agentId, companyId, name: "Source agent", status: "idle", adapterType: "process" });
+  });
+  afterEach(async () => { await db.execute(sql`truncate companies cascade`); });
+  afterAll(async () => { if (db) { await db.delete(authUsers).where(inArray(authUsers.id, users)); await db.$client.end(); } await temporary?.cleanup(); });
+  const mutation = (sourceKind: AttentionSourceKind, sourceId: string) => ({ companyId, sourceKind, sourceId,
+    authActor: { type: "board" as const, source: "local_implicit" as const, userId, companyIds: [companyId], isInstanceAdmin: true },
+    actor: { actorType: "user" as const, actorId: userId, userId, agentId: null, runId: null, agentApiKeyId: null, responsibleUserId: userId } });
+  async function issue(values: Partial<typeof issues.$inferInsert> = {}) {
+    return (await db.insert(issues).values({ companyId, title: "Security source", status: "todo", createdByAgentId: agentId, ...values }).returning())[0]!;
+  }
+  async function visible(sourceKind: AttentionSourceKind, sourceId: string) {
+    const feed = await attentionService(db).list(companyId, { all: true, allowUnscopedAll: true, userId, excludeSources: ["approval", "join_request"] });
+    return feed.items.some(item => item.sourceKind === sourceKind && item.subject.id === sourceId);
+  }
+
+  it.each(["review", "blocker_attention"] as const)("cannot pre-archive a future %s before an ordinary transition makes it visible", async kind => {
+    const row = await issue({ assigneeUserId: userId });
+    let rejected: unknown;
+    try { await decisionRetentionService(db).archive(mutation(kind, row.id)); } catch (error) { rejected = error; }
+    await issueService(db).update(row.id, kind === "review"
+      ? { status: "in_review", actorUserId: userId }
+      : { status: "blocked", unblockDescriptor: { owner: "board", action: "Choose the next owner" }, actorUserId: userId });
+    expect(await visible(kind, row.id)).toBe(true);
+    expect(rejected).toMatchObject({ status: 404 });
+    expect((await decisionRetentionService(db).getState(companyId, kind, row.id))?.archivedAt).toBeNull();
+  });
+
+  it.each(["review", "blocker_attention", "productivity_review"] as const)("does not materialize Keep for an ineligible %s", async kind => {
+    const row = await issue();
+    await expect(decisionRetentionService(db).setKeep({ ...mutation(kind, row.id), keep: true })).rejects.toMatchObject({ status: 404 });
+    expect(await decisionRetentionService(db).getState(companyId, kind, row.id)).toBeNull();
+  });
+
+  it("allows Keep on an unseen eligible review and preserves existing manual actions after it leaves review", async () => {
+    const row = await issue({ status: "in_review", assigneeUserId: userId });
+    const service = decisionRetentionService(db), input = mutation("review", row.id);
+    expect(await service.getState(companyId, "review", row.id)).toBeNull();
+    expect((await service.setKeep({ ...input, keep: true })).keep).toBe(true);
+    await issueService(db).update(row.id, { status: "todo", actorUserId: userId });
+    expect((await service.archive(input)).archivedReason).toBe("manual");
+    expect((await service.setKeep({ ...input, keep: false })).keep).toBe(false);
+    expect((await service.revive(input)).archivedAt).toBeNull();
+  });
+
+  it("allows unseen current board-owned blocker attention", async () => {
+    const row = await issue({ status: "blocked", unblockDescriptor: { owner: "board", action: "Choose next owner" }, blockedTransitionAt: new Date() });
+    expect((await decisionRetentionService(db).setKeep({ ...mutation("blocker_attention", row.id), keep: true })).keep).toBe(true);
+  });
+
+  it("requires current agent error status for a new alert but preserves existing row actions", async () => {
+    const service = decisionRetentionService(db), input = mutation("agent_error_alert", agentId);
+    await expect(service.archive(input)).rejects.toMatchObject({ status: 404 });
+    expect(await service.getState(companyId, "agent_error_alert", agentId)).toBeNull();
+    await db.update(agents).set({ status: "error" }).where(eq(agents.id, agentId));
+    expect((await service.setKeep({ ...input, keep: true })).keep).toBe(true);
+    await db.update(agents).set({ status: "idle" }).where(eq(agents.id, agentId));
+    expect((await service.archive(input)).archivedAt).toBeInstanceOf(Date);
+    expect((await service.revive(input)).archivedAt).toBeNull();
+  });
+
+  it("preserves historical productivity and terminal-leaf rows without inventing a new source", async () => {
+    const row = await issue(); const service = decisionRetentionService(db);
+    await service.syncItems(companyId, ["productivity_review", "blocker_attention"].map(sourceKind => ({ sourceKind: sourceKind as AttentionSourceKind, subject: { id: row.id }, activityAt: row.updatedAt.toISOString() })));
+    for (const kind of ["productivity_review", "blocker_attention"] as const) {
+      const input = mutation(kind, row.id);
+      expect((await service.archive(input)).archivedAt).toBeInstanceOf(Date);
+      expect((await service.revive(input)).archivedAt).toBeNull();
+      expect((await service.setKeep({ ...input, keep: true })).keep).toBe(true);
+    }
+  });
+
+  it("requires ordinary feed materialization for an unseen TODO terminal leaf, then preserves its manual controls", async () => {
+    const root = await issue({ status: "blocked" });
+    const leaf = await issue({ status: "todo" });
+    await db.insert(issueRelations).values({ companyId, issueId: leaf.id, relatedIssueId: root.id, type: "blocks" });
+    const service = decisionRetentionService(db), input = mutation("blocker_attention", leaf.id);
+    await expect(service.setKeep({ ...input, keep: true })).rejects.toMatchObject({ status: 404 });
+    expect(await service.getState(companyId, "blocker_attention", leaf.id)).toBeNull();
+    const attention = await issueService(db).listBlockerAttention(companyId, [root]);
+    expect(attention.get(root.id)).toMatchObject({ terminalBlockerIssueId: leaf.id, state: "needs_attention", blockingTreeLive: false });
+    expect(await visible("blocker_attention", leaf.id)).toBe(true);
+    expect((await service.archive(input)).archivedAt).toBeInstanceOf(Date);
+    expect((await service.revive(input)).archivedAt).toBeNull();
+    expect((await service.setKeep({ ...input, keep: true })).keep).toBe(true);
+  });
+
+  it.each(["review", "blocker_attention"] as const)("does not pre-materialize hidden %s", async kind => {
+    const row = await issue({ hiddenAt: new Date(), status: kind === "review" ? "in_review" : "blocked", assigneeUserId: userId,
+      unblockDescriptor: { owner: "board", action: "Choose next owner" }, blockedTransitionAt: new Date() });
+    await expect(decisionRetentionService(db).archive(mutation(kind, row.id))).rejects.toMatchObject({ status: 404 });
+    expect(await decisionRetentionService(db).getState(companyId, kind, row.id)).toBeNull();
+  });
+
+  it.each(["review", "agent_error_alert"] as const)("keeps company and agent authorization isolated for %s", async kind => {
+    const foreign = randomUUID();
+    await db.insert(companies).values({ id: foreign, name: "Foreign", issuePrefix: "FOREIGN" });
+    const sourceId = kind === "review" ? (await issue({ companyId: foreign, status: "in_review", assigneeUserId: userId })).id
+      : (await db.insert(agents).values({ companyId: foreign, name: "Foreign error", status: "error" }).returning())[0]!.id;
+    const service = decisionRetentionService(db);
+    await expect(service.archive(mutation(kind, sourceId))).rejects.toMatchObject({ status: 404 });
+    const input = { ...mutation(kind, sourceId), companyId: foreign, authActor: { type: "agent" as const, agentId, companyId },
+      actor: { actorType: "agent" as const, actorId: agentId, userId: null, agentId, runId: null, agentApiKeyId: null, responsibleUserId: null } };
+    await expect(service.archive(input)).rejects.toMatchObject({ status: 404 });
+    expect(await db.select().from(decisionRetention).where(and(eq(decisionRetention.companyId, foreign), eq(decisionRetention.sourceId, sourceId)))).toHaveLength(0);
+  });
+});

server/src/services/decision-retention.ts

@@ -1,4 +1,8 @@
 import { isProspectiveBlockedTransition } from "./routable-blocked.js";
+import { requiresReviewAttention, terminalBlockerRequiresAttention, ATTENTION_AGENT_ERROR_STATUS } from "./attention-source-eligibility.js";
+import { attentionIssueCondition } from "./issue-visibility.js";
+import { issueService } from "./issues.js";
+import { parseIssueExecutionState } from "./issue-execution-policy.js";
 import { notifyRetentionWork } from "./retention-work-notifications.js";
 import { decisionArchiveDeliveryService, type ArchiveNotificationBatch } from "./decision-archive-delivery.js";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
@@ -184,17 +188,56 @@ export function decisionRetentionService(
     )).then((rows) => rows[0] ?? null);
   }
 
-  async function materializeSourceState(database: Db, input: { companyId: string; sourceKind: AttentionSourceKind; sourceId: string }) {
+  async function materializeSourceState(database: Db, input: { companyId: string; sourceKind: AttentionSourceKind; sourceId: string; actor: Pick<DecisionMutationActor, "userId"> }) {
     const tables: Record<AttentionSourceKind, string> = {
       approval: "approvals", join_request: "join_requests", decision: "decisions",
       issue_thread_interaction: "issue_thread_interactions", recovery_action: "issue_recovery_actions",
       failed_run: "heartbeat_runs", productivity_review: "issues", blocker_attention: "issues", review: "issues",
       budget_alert: "budget_incidents", agent_error_alert: "agents",
     };
-    if (input.sourceKind === "blocker_attention") {
+    const existingState = () => database.select({ id: decisionRetention.id }).from(decisionRetention).where(and(
+      eq(decisionRetention.companyId, input.companyId), eq(decisionRetention.sourceKind, input.sourceKind),
+      eq(decisionRetention.sourceId, input.sourceId),
+    )).then(rows => rows[0]);
+    if (input.sourceKind === "agent_error_alert") {
+      const [source] = await database.select().from(agents).where(and(eq(agents.companyId, input.companyId), eq(agents.id, input.sourceId))).for("update");
+      if (!source || (!await existingState() && source.status !== ATTENTION_AGENT_ERROR_STATUS)) throw notFound("Attention source not found");
+      await decisionRetentionService(database).syncItems(input.companyId, [{
+        sourceKind: input.sourceKind, subject: { id: input.sourceId }, activityAt: source.updatedAt.toISOString(),
+      }]);
+      return;
+    }
+    if (["review", "blocker_attention", "productivity_review"].includes(input.sourceKind)) {
       const [source] = await database.select().from(issues).where(and(eq(issues.companyId, input.companyId), eq(issues.id, input.sourceId))).for("update");
       if (!source) throw notFound("Attention source not found");
-      const activityAt = source.unblockDescriptor?.owner === "board" && isProspectiveBlockedTransition(source)
+      if (!await existingState()) {
+        const [visible] = await database.select({ id: issues.id }).from(issues).where(and(
+          eq(issues.companyId, input.companyId), eq(issues.id, source.id), attentionIssueCondition(false),
+        ));
+        let eligible = false;
+        if (visible && input.sourceKind === "review" && source.status === "in_review") {
+          const [pendingApproval] = await database.select({ id: approvals.id }).from(issueApprovals)
+            .innerJoin(approvals, eq(approvals.id, issueApprovals.approvalId)).where(and(
+              eq(issueApprovals.companyId, input.companyId), eq(approvals.companyId, input.companyId),
+              eq(issueApprovals.issueId, source.id), eq(approvals.status, "pending"),
+            )).limit(1);
+          const state = parseIssueExecutionState(source.executionState);
+          const attention = await issueService(database).listReviewAttention(input.companyId, [source]);
+          eligible = requiresReviewAttention({
+            hasHumanParticipant: state?.status === "pending" && state.currentParticipant?.type === "user",
+            hasUserAssignee: Boolean(source.assigneeUserId), hasPendingApproval: Boolean(pendingApproval),
+            stalled: attention.get(source.id)?.state === "stalled",
+          });
+        } else if (visible && input.sourceKind === "blocker_attention" && source.status === "blocked") {
+          const owner = source.unblockDescriptor?.owner;
+          const ownsDescriptor = owner === "board" || (owner && "userId" in owner && owner.userId === input.actor.userId);
+          const attention = (await issueService(database).listBlockerAttention(input.companyId, [source])).get(source.id);
+          eligible = Boolean(ownsDescriptor && isProspectiveBlockedTransition(source))
+            || (terminalBlockerRequiresAttention(attention) && (attention.terminalBlockerIssueId ?? source.id) === source.id);
+        }
+        if (!eligible) throw notFound("Attention source not found");
+      }
+      const activityAt = input.sourceKind === "blocker_attention" && source.unblockDescriptor?.owner === "board" && isProspectiveBlockedTransition(source)
         ? source.blockedTransitionAt : source.updatedAt;
       await decisionRetentionService(database).syncItems(input.companyId, [{
         sourceKind: input.sourceKind, subject: { id: input.sourceId }, activityAt: activityAt.toISOString(),