Merge qualified sleep review fixes and prerequisites

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

* commit 'a72a0a6d653a9b338669c6aba9bbd7f0855f98a7':
  fix(retention): wake blocker retention after no-run workspace export
  Bound blocker workspace history reads by candidate workspace
  Validate eligibility before first attention retention materialization
  fix(retention): exclude finalized blockers before graph bounds
  test: hold lifecycle verification before retry identity assertions
  Exercise saved legacy input through its eligibility producer
  fix: cancel merge confirmation acceptance under lock
  Bound repeated external-object eligibility reconstruction

server/src/__tests__/attention-retention-work.test.ts

@@ -6,6 +6,8 @@ import { subscribeDatabaseWork, authUsers, companyMemberships, agents, approvals
   decisionRetention, decisions, costEvents, heartbeatRunEvents, heartbeatRuns, issueApprovals, issueRecoveryActions,
   issueRelations, issueThreadInteractions, issues, type Db } from "@paperclipai/db";
 import { attentionRetentionService, registerAttentionRetentionWork } from "../services/attention-retention.js";
+import { workspaceOperationService } from "../services/workspace-operations.js";
+import * as workspaceOperationLogs from "../services/workspace-operation-log-store.js";
 import { budgetService } from "../services/budgets.js";
 import { writeExecutionControl } from "../services/execution-control-notifications.js";
 import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js";
@@ -35,15 +37,22 @@ const delay = DEFAULT_DECISION_ARCHIVE_DAYS * 86_400_000;
   const errors: unknown[] = [];
   const statements: string[] = [];
   const parameterCounts: number[] = [];
+  let captureBlockerPlan = false;
+  let blockerPlanQuery: { query: string; parameters: unknown[] } | undefined;
   beforeAll(async () => {
     if (externalUrl) await applyPendingMigrations(externalUrl);
     else temporary = await startEmbeddedPostgresTestDatabase("attention-retention-");
     db = createDb(externalUrl ?? temporary!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 });
     db.$client.options.onnotice = () => {};
-    db.$client.options.debug = (_connection, query, parameters) => { statements.push(query); parameterCounts.push(parameters.length); };
+    db.$client.options.debug = (_connection, query, parameters) => {
+      statements.push(query); parameterCounts.push(parameters.length);
+      if (captureBlockerPlan && !blockerPlanQuery && query.includes("latest_blocker_operation") && query.includes("issue_relations"))
+        blockerPlanQuery = { query, parameters: [...parameters] };
+    };
   }, 30_000);
   beforeEach(async () => {
     companyId = randomUUID(); agentId = randomUUID(); draining = false; errors.length = 0;
+    captureBlockerPlan = false; blockerPlanQuery = undefined;
     await db.insert(companies).values({ id: companyId, name: "Retention", issuePrefix: `A${companyId.slice(0, 6)}` });
     await db.insert(authUsers).values({ id: "board", name: "Board", email: "board@retention.test", createdAt: new Date(), updatedAt: new Date() }).onConflictDoNothing();
     await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: "board", membershipRole: "owner", status: "active" });
@@ -428,6 +437,169 @@ const delay = DEFAULT_DECISION_ARCHIVE_DAYS * 86_400_000;
     expect(await db.select().from(decisionArchiveNotificationOutbox)).toHaveLength(0);
   });
 
+  it.each(["running", "run", "confirmation", "completed"] as const)("filters completed blocker history before bounded %s attention and archive checks", async kind => {
+    const root = await insertIssue("blocked"), workspaceId = randomUUID();
+    await db.execute(sql`insert into execution_workspaces (id, company_id, name, mode, strategy_type)
+      values (${workspaceId}::uuid, ${companyId}::uuid, 'Finalized shared history', 'shared', 'project_primary')`);
+    await db.execute(sql`insert into workspace_operations (company_id, execution_workspace_id, phase, status, started_at)
+      values (${companyId}::uuid, ${workspaceId}::uuid, 'workspace_finalize', 'succeeded', ${old().toISOString()}::timestamptz)`);
+    await db.execute(sql`with history as (
+      insert into issues (id, company_id, title, status, updated_at, execution_workspace_id)
+      select ('00000000-0000-4000-8000-' || lpad(to_hex(n), 12, '0'))::uuid,
+        ${companyId}::uuid, 'completed-blocker-' || n, 'done', ${old().toISOString()}::timestamptz, ${workspaceId}::uuid
+      from generate_series(1, 2100) n returning id)
+      insert into issue_relations (company_id, issue_id, related_issue_id, type)
+      select ${companyId}::uuid, id, ${root.id}::uuid, 'blocks' from history`);
+    if (kind !== "completed") {
+      const live = await insertIssue(kind === "running" ? "in_progress" : "todo", { id: "ffffffff-ffff-4fff-8fff-ffffffffffff", assigneeAgentId: agentId });
+      await db.insert(issueRelations).values({ companyId, issueId: live.id, relatedIssueId: root.id, type: "blocks" });
+      if (kind === "run") {
+        const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId, status: "running" }).returning();
+        await db.update(issues).set({ executionRunId: run.id }).where(eq(issues.id, live.id));
+      } else if (kind === "confirmation") await db.insert(issueThreadInteractions).values({ companyId, issueId: live.id,
+        kind: "request_confirmation", status: "pending", payload: { prompt: "Still waiting" } });
+    }
+    statements.length = 0; parameterCounts.length = 0;
+    captureBlockerPlan = kind === "completed";
+    const batch = await issueService(db).listBlockerAttentionBatch(companyId, [root]);
+    captureBlockerPlan = false;
+    if (kind === "completed") {
+      expect(blockerPlanQuery).toBeTruthy();
+      const explain = await db.$client.unsafe("EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) " + blockerPlanQuery!.query,
+        blockerPlanQuery!.parameters as never[]);
+      type Plan = { "Relation Name"?: string; "Actual Rows": number; "Actual Loops": number; "Rows Removed by Filter"?: number; Plans?: Plan[] };
+      const rootPlan = (explain[0]!["QUERY PLAN"] as Array<{ Plan: Plan }>)[0]!.Plan;
+      const visit = (node: Plan): void => {
+        if (node["Relation Name"] === "issues" && node["Actual Loops"] > 1) {
+          expect(node["Actual Rows"]).toBeLessThanOrEqual(1);
+          expect(node["Rows Removed by Filter"] ?? 0).toBe(0);
+        }
+        node.Plans?.forEach(visit);
+      };
+      visit(rootPlan);
+      process.stdout.write(JSON.stringify({ completedBlockerHistoryPlan: explain }) + "\n");
+    }
+    expect(batch.nodeCount).toBeLessThanOrEqual(BLOCKER_ATTENTION_MAX_NODES);
+    expect(batch.edgeCount).toBeLessThanOrEqual(BLOCKER_ATTENTION_MAX_NODES * 2);
+    const service = attentionRetentionService(db), page = await service.load({ companyId, sourceKind: "blocker_attention" });
+    const entry = page.rows!.find(row => row.rootId === root.id && row.variant === "terminal");
+    if (kind !== "completed") {
+      // Exercise the authoritative fresh check even if discovery correctly omits the live root.
+      await service.check(entry ?? { id: `blocker_attention:${root.id}:terminal`, companyId, sourceKind: "blocker_attention",
+        sourceId: root.id, rootId: root.id, issueId: root.id, variant: "terminal", activityAt: old().toISOString(), deadlineAt: Date.now(), dueAt: Date.now() });
+      expect(await state("blocker_attention", root.id)).toBeNull();
+      expect(await db.select().from(decisionArchiveNotificationOutbox)).toHaveLength(0);
+    } else {
+      expect(entry).toBeTruthy(); await service.check(entry!);
+      expect((await state("blocker_attention", root.id))?.archivedAt).toBeInstanceOf(Date);
+      expect(await db.select().from(decisionArchiveNotificationOutbox)).toHaveLength(1);
+    }
+    expect(batch.fetchedRows).toBeLessThanOrEqual(4);
+    expect(batch.deferredRootIds.size).toBe(0);
+    expect(Boolean(batch.attention.get(root.id)?.blockingTreeLive)).toBe(kind !== "completed");
+    expect(parameterCounts.length).toBeLessThanOrEqual(100);
+  });
+
+  it("does not archive an unknown truncated blocker tree during its fresh check", async () => {
+    const root = await insertIssue("blocked");
+    await db.execute(sql`with blockers as (
+      insert into issues (company_id, title, status, updated_at)
+      select ${companyId}::uuid, 'unresolved-' || n, 'backlog', ${old().toISOString()}::timestamptz
+      from generate_series(1, 2100) n returning id)
+      insert into issue_relations (company_id, issue_id, related_issue_id, type)
+      select ${companyId}::uuid, id, ${root.id}::uuid, 'blocks' from blockers`);
+    const service = attentionRetentionService(db), page = await service.load({ companyId, sourceKind: "blocker_attention" });
+    const entry = page.rows!.find(row => row.rootId === root.id && row.variant === "terminal")!;
+    expect(entry).toBeTruthy(); expect(await service.check(entry)).toBeNull();
+    expect(await db.select().from(decisionRetention)).toHaveLength(0);
+    expect(await db.select().from(decisionArchiveNotificationOutbox)).toHaveLength(0);
+    await start(); await quiet();
+  });
+
+  function simulateRecorderLogStorage() {
+    // The recorder lifecycle and SQL are real; only its log-storage boundary is simulated.
+    vi.spyOn(workspaceOperationLogs, "getWorkspaceOperationLogStore").mockReturnValue({
+      begin: async () => ({ store: "local_file", logRef: "retention-finalize-proof" }),
+      append: async () => {}, finalize: async () => ({ bytes: 0, compressed: false }), read: async () => ({ content: "" }),
+    });
+  }
+
+  it.each(["issue", "workspace"] as const)("publishes retention intent from ordinary no-run %s finalization", async scope => {
+    simulateRecorderLogStorage();
+    const source = await insertIssue(), workspaceId = randomUUID();
+    await db.execute(sql`insert into execution_workspaces (id, company_id, name, mode, strategy_type)
+      values (${workspaceId}::uuid, ${companyId}::uuid, 'Finalize producer', 'shared', 'project_primary')`);
+    await start(); await quiet();
+    const committedHints: string[] = [];
+    const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.attentionRetention, (event, hints) => {
+      if (event === "settled") committedHints.push(...hints ?? []);
+    });
+    try {
+      const result = await workspaceOperationService(db).createRecorder({ companyId,
+        executionWorkspaceId: scope === "workspace" ? workspaceId : undefined,
+        issueId: scope === "issue" ? source.id : undefined,
+      }).recordOperation({ phase: "workspace_finalize", run: async () => ({ status: "succeeded" }) });
+      expect(result.status).toBe("succeeded");
+      expect(committedHints.length).toBeGreaterThan(0);
+      expect(committedHints.map(hint => JSON.parse(hint).companyId).every(id => id === companyId)).toBe(true);
+      await quiet();
+      if (scope === "workspace") {
+        committedHints.length = 0;
+        await workspaceOperationService(db).createRecorder({ companyId })
+          .recordOperation({ phase: "workspace_finalize", run: async () => ({ status: "succeeded" }) });
+        expect(committedHints).toEqual([]); await quiet();
+      }
+    } finally { unsubscribe(); }
+  });
+
+  describe("no-run finalization with an incomplete blocker graph", () => {
+    let rootIds: string[], workspaceId: string, subjectId: string;
+    let loadedScopes: Array<Parameters<ReturnType<typeof attentionRetentionService>["load"]>[0]>;
+    beforeEach(async () => {
+      const began = performance.now();
+      simulateRecorderLogStorage();
+      const roots = await Promise.all([insertIssue("blocked"), insertIssue("blocked")]);
+      rootIds = roots.map(root => root.id); workspaceId = randomUUID();
+      await db.execute(sql`insert into execution_workspaces (id, company_id, name, mode, strategy_type)
+        values (${workspaceId}::uuid, ${companyId}::uuid, 'Pending shared export', 'shared', 'project_primary')`);
+      await workspaceOperationService(db).createRecorder({ companyId, executionWorkspaceId: workspaceId })
+        .recordOperation({ phase: "workspace_provision", run: async () => ({ status: "succeeded" }) });
+      await db.execute(sql`with history as (
+        insert into issues (company_id, title, status, updated_at, execution_workspace_id)
+        select ${companyId}::uuid, 'waiting-for-export-' || n, 'done', ${old().toISOString()}::timestamptz, ${workspaceId}::uuid
+        from generate_series(1, 2100) n returning id)
+        insert into issue_relations (company_id, issue_id, related_issue_id, type)
+        select ${companyId}::uuid, history.id, root.id, 'blocks' from history
+        cross join (values (${rootIds[0]}::uuid), (${rootIds[1]}::uuid)) root(id)`);
+      const [subject] = await db.select({ id: issues.id }).from(issues)
+        .where(and(eq(issues.companyId, companyId), eq(issues.status, "done"))).limit(1);
+      subjectId = subject!.id;
+      const review = await insertIssue("in_review", { assigneeUserId: "board" });
+      await decisionRetentionService(db).setKeep({ ...mutation("review", review.id), keep: true });
+      const service = attentionRetentionService(db), load = service.load.bind(service); loadedScopes = [];
+      vi.spyOn(service, "load").mockImplementation((scope, signal) => { loadedScopes.push(scope); return load(scope, signal); });
+      process.stdout.write(JSON.stringify({ incompleteFinalizePhase: "seeded", setupMs: performance.now() - began }) + "\n");
+      await start(service); await quiet();
+      for (const rootId of rootIds) expect(await state("blocker_attention", rootId)).toBeNull();
+      process.stdout.write(JSON.stringify({ incompleteFinalizeFixtureRows: 2100, incompleteRoots: 2, setupMs: performance.now() - began }) + "\n");
+    }, 30_000);
+    it("archives only after the ordinary workspace-only finalize commits, then settles quiet", async () => {
+      const finish = (database: Db) => workspaceOperationService(database).createRecorder({ companyId, executionWorkspaceId: workspaceId, issueId: subjectId })
+        .recordOperation({ phase: "workspace_finalize", run: async () => ({ status: "succeeded" }) });
+      await expect(db.transaction(async tx => {
+        await finish(tx as unknown as Db); throw new Error("rollback finalization");
+      })).rejects.toThrow("rollback finalization");
+      await settle(); for (const rootId of rootIds) expect(await state("blocker_attention", rootId)).toBeNull();
+      loadedScopes.length = 0;
+      expect((await finish(db)).status).toBe("succeeded");
+      for (const rootId of rootIds) await archived("blocker_attention", rootId);
+      await quiet();
+      expect(loadedScopes.length).toBeGreaterThan(0);
+      expect(loadedScopes.every(scope => scope?.companyId === companyId && scope.sourceKind === "blocker_attention")).toBe(true);
+      expect(await db.select().from(decisionArchiveNotificationOutbox)).toHaveLength(2);
+    });
+  });
+
   it("disarms finite deferred-root checks when large retained graphs have no archive obligation", async () => {
     await db.execute(sql`insert into issues (id, company_id, title, status, updated_at)
       select gen_random_uuid(), ${companyId}::uuid, 'held-root-' || n, 'blocked', ${old().toISOString()}::timestamptz

server/src/__tests__/blocker-workspace-history-performance.test.ts

@@ -0,0 +1,103 @@
+import { randomUUID } from "node:crypto";
+import { sql } from "drizzle-orm";
+import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest";
+import { applyPendingMigrations, companies, createDb, issues } from "@paperclipai/db";
+import { issueService } from "../services/issues.js";
+import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
+import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
+
+const external = resolveExternalTestDatabaseUrl("PAPERCLIP_BLOCKER_HISTORY_TEST_DATABASE_URL");
+const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport();
+type Plan = { "Node Type": string; "Relation Name"?: string; "Actual Rows": number; "Actual Loops": number;
+  "Rows Removed by Filter"?: number; "Rows Removed by Index Recheck"?: number; Plans?: Plan[] };
+
+describe.skipIf(!support.supported)("blocker workspace retained history cost", () => {
+  let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
+  let db: ReturnType<typeof createDb>;
+  let capture = false;
+  const queries: Array<{ query: string; parameters: unknown[] }> = [];
+  beforeAll(async () => {
+    if (external) await applyPendingMigrations(external);
+    else temporary = await startEmbeddedPostgresTestDatabase("blocker-history-");
+    db = createDb(external ?? temporary!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 });
+    db.$client.options.onnotice = () => {};
+    db.$client.options.debug = (_connection, query, parameters) => {
+      if (capture && query.includes("workspace_operations")) queries.push({ query, parameters: [...parameters] });
+    };
+  }, 30_000);
+  afterEach(async () => { capture = false; queries.length = 0; await db.execute(sql`truncate companies cascade`); });
+  afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); });
+
+  it.each(["finalized", "pending"] as const)("bounds operation-history work for 2100 blockers sharing 5000 %s operations", async kind => {
+    const companyId = randomUUID(), workspaceId = randomUUID(), unrelatedWorkspaceId = randomUUID();
+    await db.insert(companies).values({ id: companyId, name: "Shared history", issuePrefix: `SH${companyId.slice(0, 6)}` });
+    const [root] = await db.insert(issues).values({ companyId, title: "History dependent", status: "blocked" }).returning();
+    await db.execute(sql`insert into execution_workspaces (id, company_id, name, mode, strategy_type)
+      values (${workspaceId}::uuid, ${companyId}::uuid, 'Shared operation history', 'shared', 'project_primary'),
+        (${unrelatedWorkspaceId}::uuid, ${companyId}::uuid, 'Unrelated history', 'shared', 'project_primary')`);
+    await db.execute(sql`insert into workspace_operations (company_id, execution_workspace_id, phase, status, started_at)
+      select ${companyId}::uuid, ${unrelatedWorkspaceId}::uuid, 'worktree_prepare', 'succeeded',
+        '2025-01-01T00:00:00Z'::timestamptz + n * interval '1 second' from generate_series(1, 50000) n`);
+    await db.execute(sql`insert into workspace_operations (company_id, execution_workspace_id, phase, status, started_at)
+      select ${companyId}::uuid, ${workspaceId}::uuid,
+        case when n = 5000 and ${kind} = 'finalized' then 'workspace_finalize' else 'worktree_prepare' end,
+        'succeeded', '2026-01-01T00:00:00Z'::timestamptz + n * interval '1 second' from generate_series(1, 5000) n`);
+    await db.execute(sql`with history as (
+      insert into issues (company_id, title, status, execution_workspace_id)
+      select ${companyId}::uuid, 'done blocker ' || n, 'done', ${workspaceId}::uuid
+      from generate_series(1, 2100) n returning id)
+      insert into issue_relations (company_id, issue_id, related_issue_id, type)
+      select ${companyId}::uuid, id, ${root.id}::uuid, 'blocks' from history`);
+    await db.execute(sql`analyze workspace_operations`);
+    await db.execute(sql`analyze issues`);
+    await db.execute(sql`analyze issue_relations`);
+    capture = true;
+    const batch = await issueService(db).listBlockerAttentionBatch(companyId, [root]);
+    capture = false;
+    expect(queries.length).toBeGreaterThan(0);
+    expect(Math.max(...queries.map(statement => statement.parameters.length))).toBeLessThanOrEqual(1100);
+    let operationRowsExamined = 0;
+    const measurements: unknown[] = [];
+    for (const statement of queries) {
+      const result = await db.$client.unsafe("EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) " + statement.query,
+        statement.parameters as never[]);
+      const explain = (result[0]!["QUERY PLAN"] as Array<{ Plan: Plan }>)[0]!;
+      const visit = (node: Plan) => {
+        if (node["Relation Name"] === "workspace_operations") operationRowsExamined += node["Actual Loops"]
+          * (node["Actual Rows"] + (node["Rows Removed by Filter"] ?? 0) + (node["Rows Removed by Index Recheck"] ?? 0));
+        node.Plans?.forEach(visit);
+      };
+      visit(explain.Plan);
+      measurements.push({ ...statement, explain });
+    }
+    process.stdout.write(JSON.stringify({ sharedWorkspaceHistory: { kind, blockers: 2100, retainedOperations: 5000, unrelatedOperations: 50000,
+      operationRowsExamined, queryCount: queries.length, measurements } }) + "\n");
+    // The finite graph/pair pages may each read the 5,000 candidate rows once.
+    // Neither 2,100 blockers nor 50,000 unrelated operations may multiply that work.
+    expect(operationRowsExamined).toBeLessThanOrEqual(55_000);
+    expect(batch.deferredRootIds.has(root.id)).toBe(kind === "pending");
+    if (kind === "finalized") expect(batch.nodeCount).toBe(1);
+  }, 120_000);
+  it("preserves attributed precedence and the strict later-finalize boundary", async () => {
+    const companyId = randomUUID(), workspaceId = randomUUID();
+    await db.insert(companies).values({ id: companyId, name: "Finalize policy", issuePrefix: `FP${companyId.slice(0, 6)}` });
+    await db.execute(sql`insert into execution_workspaces (id, company_id, name, mode, strategy_type)
+      values (${workspaceId}::uuid, ${companyId}::uuid, 'Policy workspace', 'shared', 'project_primary')`);
+    const roots = await db.insert(issues).values(["older", "equal", "fallback"].map(title => ({ companyId, title, status: "blocked" }))).returning();
+    const blockers = await db.insert(issues).values(roots.map(root => ({ companyId, title: root.title + " blocker", status: "done", executionWorkspaceId: workspaceId }))).returning();
+    for (let index = 0; index < roots.length; index += 1) {
+      await db.execute(sql`insert into issue_relations (company_id, issue_id, related_issue_id, type)
+        values (${companyId}::uuid, ${blockers[index]!.id}::uuid, ${roots[index]!.id}::uuid, 'blocks')`);
+    }
+    await db.execute(sql`insert into workspace_operations (company_id, execution_workspace_id, issue_id, phase, status, started_at) values
+      (${companyId}::uuid, ${workspaceId}::uuid, ${blockers[0]!.id}::uuid, 'worktree_prepare', 'succeeded', '2026-01-01T00:00:01Z'),
+      (${companyId}::uuid, ${workspaceId}::uuid, ${blockers[1]!.id}::uuid, 'worktree_prepare', 'succeeded', '2026-01-01T00:00:02Z'),
+      (${companyId}::uuid, ${workspaceId}::uuid, null, 'workspace_finalize', 'succeeded', '2026-01-01T00:00:02Z'),
+      (${companyId}::uuid, ${workspaceId}::uuid, null, 'worktree_prepare', 'succeeded', '2026-01-01T00:00:03Z')`);
+    const service = issueService(db);
+    expect((await service.listBlockerAttentionBatch(companyId, [roots[0]!])).nodeCount).toBe(1);
+    expect((await service.listBlockerAttentionBatch(companyId, [roots[1]!])).nodeCount).toBe(2);
+    expect((await service.listBlockerAttentionBatch(companyId, [roots[2]!])).nodeCount).toBe(2);
+  });
+
+});

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/__tests__/external-object-refresh-work.test.ts

@@ -368,6 +368,56 @@ const active: ExternalObjectResolveResult = { ok: true, snapshot: {
     expect(provider).toHaveBeenCalledTimes(2);
   });
 
+  it("coalesces repeated company status writes without rebuilding every future object", async () => {
+    clock(); const svc = service(); const own = await fixture(), foreign = await fixture();
+    const future = new Date(Date.now() + 86_400_000);
+    for (let offset = 0; offset < 1100; offset += 250) {
+      const ids = Array.from({ length: Math.min(250, 1100 - offset) }, () => randomUUID());
+      await db.insert(externalObjects).values(ids.map(id => ({ id, companyId: own.companyId,
+        providerKey: "url", objectType: "link", externalId: id, nextRefreshAt: future })));
+      await db.insert(externalObjectMentions).values(ids.map(objectId => ({ companyId: own.companyId,
+        sourceIssueId: own.issueId, sourceKind: "description" as const, objectId })));
+    }
+    await svc.syncIssue(foreign.issueId);
+    await db.update(externalObjects).set({ nextRefreshAt: future });
+    await start(svc);
+    await settle(() => coordinator!.nextWakeAt() === future.getTime());
+    const reads: Array<{ query: string; params: unknown[] }> = [];
+    const debug = db.$client.options.debug;
+    db.$client.options.debug = (_connection, query, params) => {
+      if (/^select/i.test(query) && query.includes('from "external_objects"') && query.includes('inner join "companies"'))
+        reads.push({ query, params: [...params] });
+    };
+    try {
+      const companiesSvc = companyService(db);
+      await companiesSvc.update(own.companyId, { status: "paused" });
+      await settle(() => reads.length === 1 && coordinator!.nextWakeAt() === future.getTime()
+        && !databaseWorkPending(db, DELIVERY_QUEUES.externalObjectRefresh));
+      for (const status of ["active", "paused", "active", "paused", "active"] as const) {
+        await companiesSvc.update(own.companyId, { status });
+        await settle(() => !databaseWorkPending(db, DELIVERY_QUEUES.externalObjectRefresh)
+          && (coordinator!.nextWakeAt() ?? 0) > Date.now());
+        expect(reads).toHaveLength(1);
+      }
+      const rebuildAt = coordinator!.nextWakeAt()!;
+      expect(rebuildAt - Date.now()).toBeGreaterThan(20_000);
+      expect(rebuildAt - Date.now()).toBeLessThanOrEqual(30_000);
+      await vi.advanceTimersByTimeAsync(rebuildAt - Date.now());
+      await settle(() => coordinator!.nextWakeAt() === future.getTime() && reads.length === 6);
+      for (const read of reads) {
+        expect(read.query).toContain('"external_objects"."company_id" =');
+        expect(read.params).toContain(own.companyId);
+        expect(read.params).not.toContain(foreign.companyId);
+        expect(read.params.at(-1)).toBe(250);
+      }
+      const count = reads.length;
+      await vi.advanceTimersByTimeAsync(20 * 60_000);
+      expect(reads).toHaveLength(count); expect(provider).not.toHaveBeenCalled();
+      expect(coordinator!.nextWakeAt()).toBe(future.getTime());
+      expect(errors).not.toHaveBeenCalled();
+    } finally { db.$client.options.debug = debug; }
+  });
+
   it("scopes repeated issue deletions to cascaded source objects and cancels their last future mention", async () => {
     clock(); const svc = service();
     const own = await fixture(), foreign = await fixture();

server/src/__tests__/merged-pr-confirmation-work.test.ts

@@ -22,7 +22,7 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
   let coordinator: ReturnType<typeof createDeliveryWorkCoordinator> | undefined;
   let draining = false, standby = false;
   const errors: unknown[] = [];
-  const provider = vi.fn(async (_companyId: string, _reference: { number: number }): Promise<"open" | "merged" | "unknown"> => "merged");
+  const provider = vi.fn(async (_companyId: string, _reference: { number: number }, _signal?: AbortSignal): Promise<"open" | "merged" | "unknown"> => "merged");
   const wakeup = vi.fn(async () => ({ id: "simulated-durable-wake" }));
   beforeAll(async () => {
     if (!external) temporary = await startEmbeddedPostgresTestDatabase("merged-pr-work-");
@@ -248,6 +248,84 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
     expect(errors).toEqual([]);
   });
 
+  it("preserves manual acceptance without cancellation and rolls back a failed caller mutation", async () => {
+    const card = await create();
+    const issue = { id: issueId, companyId, projectId: null, goalId: null };
+    await expect(service().acceptInteraction(issue, card.id, {}, { userId: "board" }, {
+      afterResolveInTransaction: async () => { throw new Error("manual acceptance rollback"); },
+    })).rejects.toThrow("manual acceptance rollback");
+    expect((await stored(card.id)).status).toBe("pending");
+    const accepted = await service().acceptInteraction(issue, card.id, {}, { userId: "board" });
+    expect(accepted.interaction.status).toBe("accepted");
+    expect((await stored(card.id)).status).toBe("accepted");
+  });
+
+  it.each(["issue lock", "acceptance write"] as const)("keeps a stopped %s acceptance pending and accepts once after restart", async phase => {
+    vi.useRealTimers();
+    const card = await create();
+    let signal: AbortSignal | undefined;
+    provider.mockImplementation(async (_companyId, _reference, currentSignal) => { signal = currentSignal; return "merged"; });
+    let reserved: Awaited<ReturnType<typeof db.$client.reserve>> | undefined = await db.$client.reserve();
+    const [{ pid }] = await reserved<{ pid: number }[]>`select pg_backend_pid() as pid`;
+    if (phase === "issue lock") {
+      await reserved.unsafe("begin");
+      await reserved`select id from issues where id=${issueId} for update`;
+    } else {
+      await db.$client.unsafe(`create function maintenance_hold_merge_acceptance() returns trigger language plpgsql as $$begin
+        if new.id = '${card.id}'::uuid and new.status = 'accepted' then perform pg_advisory_xact_lock(930582); end if; return new; end$$`);
+      await db.$client.unsafe(`create trigger maintenance_hold_merge_acceptance before update on issue_thread_interactions for each row execute function maintenance_hold_merge_acceptance()`);
+      await reserved`select pg_advisory_lock(930582)`;
+    }
+    let stopping: Promise<void> | undefined;
+    try {
+      coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
+      const worker = registerMergeConfirmationWork(coordinator, service(), () => true);
+      let blocked: { query: string } | undefined;
+      for (let attempt = 0; attempt < 200; attempt++) {
+        [blocked] = await db.$client<{ query: string }[]>`select query from pg_stat_activity
+          where datname=current_database() and wait_event_type='Lock' and ${pid} = any(pg_blocking_pids(pid))`;
+        if (blocked) break;
+        await delay(10);
+      }
+      expect(blocked?.query).toContain(phase === "issue lock" ? '"issues"' : '"issue_thread_interactions"');
+      let stopped = false;
+      stopping = coordinator.stop().then(() => { stopped = true; });
+      await delay(20);
+      expect(signal?.aborted).toBe(true);
+      expect(stopped).toBe(false);
+      expect((await stored(card.id)).status).toBe("pending");
+      const releasedAt = performance.now();
+      if (phase === "issue lock") await reserved.unsafe("commit");
+      else await reserved`select pg_advisory_unlock(930582)`;
+      reserved.release(); reserved = undefined;
+      await stopping; await worker.ready; coordinator = undefined;
+      expect(performance.now() - releasedAt).toBeLessThan(1000);
+      expect((await stored(card.id)).status).toBe("pending");
+      expect(wakeup).not.toHaveBeenCalled();
+      expect(await db.select().from(activityLog).where(eq(activityLog.action, "issue.thread_interaction_accepted"))).toHaveLength(0);
+      expect(await db.select().from(chatActions)).toHaveLength(0);
+      await start();
+      expect((await stored(card.id)).status).toBe("accepted");
+      expect(wakeup).toHaveBeenCalledTimes(1);
+      expect(await db.select().from(activityLog).where(eq(activityLog.action, "issue.thread_interaction_accepted"))).toHaveLength(1);
+      vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
+      await quiet();
+      expect(wakeup).toHaveBeenCalledTimes(1); expect(errors).toEqual([]);
+    } finally {
+      if (reserved) {
+        if (phase === "issue lock") await reserved.unsafe("rollback");
+        else await reserved`select pg_advisory_unlock(930582)`;
+        reserved.release();
+      }
+      await stopping;
+      await coordinator?.stop(); coordinator = undefined;
+      if (phase === "acceptance write") {
+        await db.$client.unsafe("drop trigger if exists maintenance_hold_merge_acceptance on issue_thread_interactions");
+        await db.$client.unsafe("drop function if exists maintenance_hold_merge_acceptance()");
+      }
+    }
+  });
+
   it("awaits an in-flight observation at shutdown and reconstructs its unaccepted card on restart", async () => {
     let release!: () => void, observed!: () => void;
     const barrier = new Promise<void>(resolve => { release = resolve; });

server/src/__tests__/scoped-deadline-work.test.ts

@@ -37,3 +37,41 @@ it("preserves a subsecond replacement deadline when the owner does not require p
   expect(check).toHaveBeenCalledOnce();
   expect(onError).not.toHaveBeenCalled();
 });
+
+
+it("coalesces broad reconstruction without delaying an exact object hint or saved deadline", async () => {
+  vi.useFakeTimers();
+  const owner = installDatabaseWorkSignals({ execute: vi.fn(async () => []) });
+  coordinator = createDeliveryWorkCoordinator({ owner, canRun: () => true, onError: vi.fn() });
+  const now = Date.now();
+  let dueAt = now + 60_000;
+  const load = vi.fn(async () => [{ id: "object", companyId: "company", dueAt }]);
+  const check = vi.fn(async () => null);
+  const worker = registerScopedDeadlineWork(coordinator, {
+    queue: DELIVERY_QUEUES.externalObjectRefresh,
+    minimumPassIntervalMs: 0,
+    reconstructionIntervalMs: (scope: string | undefined) => scope === "company" ? 30_000 : 0,
+    decodeScope: hint => hint === "*" ? undefined : hint,
+    matches: () => true,
+    load,
+    check,
+  }, () => true);
+  await worker.ready;
+  worker.invalidate("company"); await vi.advanceTimersByTimeAsync(0);
+  expect(load).toHaveBeenCalledTimes(2);
+  worker.invalidate("company"); await vi.advanceTimersByTimeAsync(0);
+  expect(load).toHaveBeenCalledTimes(2);
+  expect(coordinator.nextWakeAt()).toBe(now + 30_000);
+  dueAt = now + 250;
+  worker.invalidate("object"); await vi.advanceTimersByTimeAsync(0);
+  expect(load).toHaveBeenCalledTimes(3);
+  expect(coordinator.nextWakeAt()).toBe(dueAt);
+  await vi.advanceTimersByTimeAsync(250);
+  expect(check).toHaveBeenCalledOnce();
+  load.mockResolvedValue([]);
+  await vi.advanceTimersByTimeAsync(29_750);
+  expect(load).toHaveBeenCalledTimes(4);
+  expect(coordinator.nextWakeAt()).toBeNull();
+  await vi.advanceTimersByTimeAsync(120_000);
+  expect(load).toHaveBeenCalledTimes(4); expect(check).toHaveBeenCalledOnce();
+});

server/src/modules/agent-lifecycle/__tests__/verification-retry.test.ts

@@ -19,8 +19,8 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
   let companyId: string;
   const workers: Array<{ stop(): Promise<void> }> = [];
   beforeAll(async () => {
-    if (external) { await applyPendingMigrations(external); db = createDb(external); }
-    else { database = await startEmbeddedPostgresTestDatabase("paperclip-verification-retry-"); db = createDb(database.connectionString); }
+    if (external) { await applyPendingMigrations(external); db = createDb(external, { maxConnections: 6 }); }
+    else { database = await startEmbeddedPostgresTestDatabase("paperclip-verification-retry-"); db = createDb(database.connectionString, { maxConnections: 6 }); }
   }, 90_000);
   afterAll(async () => { if (external) await closeRegisteredClients(external); else await database?.cleanup(); });
   beforeEach(async () => {
@@ -76,12 +76,22 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
 
   it("permits an explicit retry of the same operation and completes verification before readiness", async () => {
     const { agent, work, runHost } = await failed();
-    runHost.mockResolvedValue("complete");
-    await createAgentLifecycle(db).retry(agent.id);
-    expect((await current(agent.id)).lifecycleOperation).not.toHaveProperty("retryRequired");
-    expect((await current(agent.id)).lifecycleOperation).not.toHaveProperty("verificationFailures");
-    expect((await current(agent.id)).lifecycleOperation!.id).toBe(agent.lifecycleOperation!.id);
-    await work.process(agent.id);
+    let release!: () => void;
+    const verification = new Promise<void>(resolve => { release = resolve; });
+    runHost.mockImplementation(async () => { await verification; return "complete"; });
+    let processing: Promise<void> | undefined;
+    try {
+      await createAgentLifecycle(db).retry(agent.id);
+      processing = work.process(agent.id);
+      const retried = await current(agent.id);
+      expect(retried).toMatchObject({ lifecycleState: "verifying", status: "paused" });
+      expect(retried.lifecycleOperation).not.toHaveProperty("retryRequired");
+      expect(retried.lifecycleOperation).not.toHaveProperty("verificationFailures");
+      expect(retried.lifecycleOperation!.id).toBe(agent.lifecycleOperation!.id);
+    } finally {
+      release();
+      await processing;
+    }
     expect(await current(agent.id)).toMatchObject({ lifecycleState: "ready", status: "idle", lifecycleError: null });
   });
 

server/src/services/attention-retention.ts

@@ -235,7 +235,11 @@ export function attentionRetentionService(db: Db) {
           where company_id = ${candidate.companyId}::uuid and id = ${row.id}::uuid`) as unknown as Array<{ at: string; due: string }>)[0];
         activityAt = activity.at; dueAt = Number(activity.due);
       } else {
-        const attention = evidence ? evidence.blockers.get(row.id) : (await issueService(database).listBlockerAttention(candidate.companyId, [row])).get(row.id);
+        const fresh = evidence ? undefined : await issueService(database).listBlockerAttentionBatch(candidate.companyId, [row]);
+        // A bounded graph cannot establish absence of live work beyond its frontier.
+        // Keep unknown roots visible; a later source change can revalidate them.
+        if (fresh?.deferredRootIds.has(row.id)) return null;
+        const attention = evidence ? evidence.blockers.get(row.id) : fresh!.attention.get(row.id);
         if (!terminalBlockerRequiresAttention(attention)) return null;
         sourceId = attention.terminalBlockerIssueId ?? row.id;
         if (lock && sourceId !== candidate.sourceId) await database.execute(sql`select id from issues

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(),

server/src/services/external-objects.ts

@@ -1158,6 +1158,9 @@ export function externalObjectService(
       queue: DELIVERY_QUEUES.externalObjectRefresh,
       batchSize: 50,
       minimumPassIntervalMs: 0,
+      // Eligibility changes affect every object in the company. Coalesce repeated
+      // rebuilds while object/identity hints and saved deadlines remain immediate.
+      reconstructionIntervalMs: scope => scope && !scope.objectId && !scope.identity ? 30_000 : 0,
       decodeScope: hint => hint === "*" ? undefined : JSON.parse(hint) as RefreshScope,
       matches: (entry, scope) => !scope || (entry.companyId === scope.companyId
         && (!scope.objectId || entry.id === scope.objectId)

server/src/services/issue-thread-interactions.ts

@@ -2278,10 +2278,12 @@ export function issueThreadInteractionService(
     input: AcceptIssueThreadInteraction;
     actor: InteractionActor;
     mutationOptions?: InteractionResolutionMutationOptions;
+    signal?: AbortSignal;
   }): Promise<{
     interaction: IssueThreadInteraction;
     continuationIssue: IssueWakeTarget | null;
   }> {
+    args.signal?.throwIfAborted();
     const expired = await expireStaleRequestConfirmationTarget(db, {
       row: args.current,
       actor: args.actor,
@@ -2324,6 +2326,7 @@ export function issueThreadInteractionService(
         .for("update")
         .then((rows: IssueResolutionContext[]) => rows[0] ?? null);
 
+      args.signal?.throwIfAborted();
       if (!issueContext || issueContext.companyId !== args.issue.companyId) {
         throw notFound("Issue not found");
       }
@@ -2376,6 +2379,7 @@ export function issueThreadInteractionService(
           : undefined;
 
       if (lockedCurrent.kind === "request_confirmation") await notifyToolProjectionForInteractions(tx, lockedCurrent.companyId, [lockedCurrent.id]);
+      args.signal?.throwIfAborted();
       const [updated] = await tx
         .update(issueThreadInteractions)
         .set({
@@ -2547,6 +2551,7 @@ export function issueThreadInteractionService(
         });
       }
 
+      args.signal?.throwIfAborted();
       return {
         interaction: acceptedInteraction,
         continuationIssue,
@@ -3103,6 +3108,7 @@ export function issueThreadInteractionService(
           resolved = await acceptRequestConfirmation({
             issue: candidate.issue,
             current: candidate.interaction,
+            signal,
             input: {},
             actor: {
               systemId: "system:pr-merged",

server/src/services/issues.ts

@@ -2523,124 +2523,87 @@ export async function listUnfinalizedExecutionWorkspaceIds(
   return unfinalized;
 }
 
+/** Workspace-wide facts are shared by blockers, rather than rereading retained
+ * operation history for every issue pointing at the same workspace. The ordered
+ * OFFSET boundary keeps the rare finalize filter from turning into a company
+ * history scan; the existing workspace/started-at index can stream candidates. */
+function blockerWorkspaceFacts(companyId: string, workspaceScope: SQL) {
+  return sql`(
+    with candidate_workspaces as materialized (
+      select distinct workspace_id from (${workspaceScope}) scoped_workspaces(workspace_id)
+    ), workspace_facts as materialized (
+      select candidate.workspace_id, finalized.started_at as finalized_at,
+        legacy.phase, legacy.status, legacy.started_at
+      from candidate_workspaces candidate
+      left join lateral (
+        select started_at from (
+          select phase, status, started_at from workspace_operations
+          where company_id = ${companyId}::uuid
+            and execution_workspace_id = candidate.workspace_id
+          order by started_at desc offset 0
+        ) workspace_history
+        where phase = 'workspace_finalize' and status = 'succeeded'
+        order by started_at desc limit 1
+      ) finalized on true
+      left join lateral (
+        select phase, status, started_at from workspace_operations
+        where company_id = ${companyId}::uuid
+          and execution_workspace_id = candidate.workspace_id and issue_id is null
+        order by started_at desc limit 1
+      ) legacy on true
+    ) select * from workspace_facts
+  ) as blocker_workspace_facts`;
+}
+
+/** An issue's own operation takes precedence over the legacy NULL fallback.
+ * Only a strictly later successful finalize clears either barrier. */
+function pendingFinalizeBlockerCondition(companyId: string, issueId: SQL, workspaceId: SQL) {
+  return sql`${workspaceId} is not null and exists (
+    select 1 from (values (1)) seed(value)
+    left join lateral (
+      select phase, status, started_at from workspace_operations blocker_operation
+      where blocker_operation.company_id = ${companyId}::uuid
+        and blocker_operation.execution_workspace_id = ${workspaceId}
+        and blocker_operation.issue_id = ${issueId}
+      order by blocker_operation.started_at desc limit 1
+    ) latest_blocker_operation on true
+    where coalesce(latest_blocker_operation.started_at, blocker_workspace_facts.started_at) is not null
+      and not (coalesce(latest_blocker_operation.phase, blocker_workspace_facts.phase) = 'workspace_finalize'
+        and coalesce(latest_blocker_operation.status, blocker_workspace_facts.status) = 'succeeded')
+      and (blocker_workspace_facts.finalized_at is null
+        or blocker_workspace_facts.finalized_at <= coalesce(latest_blocker_operation.started_at, blocker_workspace_facts.started_at))
+  )`;
+}
+
 async function listPendingFinalizeBlockerIssueIds(
   dbOrTx: Pick<Db, "select">,
   companyId: string,
-  blockerWorkspacePairs: Array<{
-    blockerIssueId: string;
-    executionWorkspaceId: string;
-  }>,
+  blockerWorkspacePairs: Array<{ blockerIssueId: string; executionWorkspaceId: string }>,
   lockSources = false,
 ): Promise<Set<string>> {
   const pending = new Set<string>();
-  const blockerIssueIds = [
-    ...new Set(blockerWorkspacePairs.map((pair) => pair.blockerIssueId)),
-  ];
-  const executionWorkspaceIds = [
-    ...new Set(blockerWorkspacePairs.map((pair) => pair.executionWorkspaceId)),
-  ];
-  if (blockerIssueIds.length === 0 || executionWorkspaceIds.length === 0)
-    return pending;
-  const blockerWorkspaceKeys = new Set(
-    blockerWorkspacePairs.map(
-      (pair) => `${pair.blockerIssueId}:${pair.executionWorkspaceId}`,
-    ),
-  );
-
-  const rows: Array<{ issueId: string | null; executionWorkspaceId: string | null;
-    phase: string; status: string; startedAt: Date }> = [];
-  for (const chunk of chunkList(executionWorkspaceIds, ISSUE_LIST_RELATED_QUERY_CHUNK_SIZE)) {
-    if (lockSources) {
-
-
+  const pairs = [...new Map(blockerWorkspacePairs.map(pair => [`${pair.blockerIssueId}:${pair.executionWorkspaceId}`, pair])).values()];
+  if (lockSources) {
+    const workspaceIds = [...new Set(pairs.map(pair => pair.executionWorkspaceId))];
+    for (const chunk of chunkList(workspaceIds, ISSUE_LIST_RELATED_QUERY_CHUNK_SIZE)) {
       await dbOrTx.select({ id: executionWorkspaces.id }).from(executionWorkspaces)
         .where(and(eq(executionWorkspaces.companyId, companyId), inArray(executionWorkspaces.id, chunk)))
         .for("update", { noWait: true });
-    }
-    const query = dbOrTx
-      .select({
-        issueId: workspaceOperations.issueId,
-        executionWorkspaceId: workspaceOperations.executionWorkspaceId,
-        phase: workspaceOperations.phase,
-        status: workspaceOperations.status,
-        startedAt: workspaceOperations.startedAt,
-      })
-      .from(workspaceOperations)
-      .where(
-        and(
-          eq(workspaceOperations.companyId, companyId),
-          inArray(
-            workspaceOperations.executionWorkspaceId,
-            chunk,
-          ),
-        ),
-      );
-
-    rows.push(...await (lockSources ? query.for("share", { noWait: true }) : query));
-  }
-
-  const latestAttributedByBlockerWorkspace = new Map<
-    string,
-    { phase: string; status: string; startedAt: Date }
-  >();
-  const latestUnattributedByWorkspace = new Map<
-    string,
-    { phase: string; status: string; startedAt: Date }
-  >();
-  const latestSuccessfulFinalizeByWorkspace = new Map<string, Date>();
-  for (const row of rows) {
-    if (!row.executionWorkspaceId) continue;
-    if (row.phase === "workspace_finalize" && row.status === "succeeded") {
-      const current = latestSuccessfulFinalizeByWorkspace.get(
-        row.executionWorkspaceId,
-      );
-      if (!current || row.startedAt > current) {
-        latestSuccessfulFinalizeByWorkspace.set(
-          row.executionWorkspaceId,
-          row.startedAt,
-        );
-      }
-    }
-    if (row.issueId) {
-      const key = `${row.issueId}:${row.executionWorkspaceId}`;
-      if (!blockerWorkspaceKeys.has(key)) continue;
-      const current = latestAttributedByBlockerWorkspace.get(key);
-      if (!current || row.startedAt > current.startedAt) {
-        latestAttributedByBlockerWorkspace.set(key, {
-          phase: row.phase,
-          status: row.status,
-          startedAt: row.startedAt,
-        });
-      }
-      continue;
-    }
-
-    const current = latestUnattributedByWorkspace.get(row.executionWorkspaceId);
-    if (!current || row.startedAt > current.startedAt) {
-      latestUnattributedByWorkspace.set(row.executionWorkspaceId, {
-        phase: row.phase,
-        status: row.status,
-        startedAt: row.startedAt,
-      });
+      await dbOrTx.select({ id: workspaceOperations.id }).from(workspaceOperations)
+        .where(and(eq(workspaceOperations.companyId, companyId), inArray(workspaceOperations.executionWorkspaceId, chunk)))
+        .for("share", { noWait: true });
     }
   }
-
-  for (const pair of blockerWorkspacePairs) {
-    const latest =
-      latestAttributedByBlockerWorkspace.get(
-        `${pair.blockerIssueId}:${pair.executionWorkspaceId}`,
-      ) ?? latestUnattributedByWorkspace.get(pair.executionWorkspaceId);
-    if (!latest) continue; // no ops recorded -> nothing to finalize for this blocker
-    if (latest.phase === "workspace_finalize" && latest.status === "succeeded")
-      continue;
-    const laterSuccessfulFinalize = latestSuccessfulFinalizeByWorkspace.get(
-      pair.executionWorkspaceId,
-    );
-    if (laterSuccessfulFinalize && laterSuccessfulFinalize > latest.startedAt)
-      continue;
-    pending.add(pair.blockerIssueId);
+  // Each pair occurs in both the source and the workspace scope.
+  for (const chunk of chunkList(pairs, ISSUE_LIST_RELATED_QUERY_CHUNK_SIZE / 2)) {
+    const values = sql.join(chunk.map(pair => sql`(${pair.blockerIssueId}::uuid, ${pair.executionWorkspaceId}::uuid)`), sql`, `);
+    const rows = await dbOrTx.select({ id: sql<string>`blocker_source.issue_id` })
+      .from(sql`(values ${values}) as blocker_source(issue_id, workspace_id)`)
+      .leftJoin(blockerWorkspaceFacts(companyId, sql`select workspace_id from (values ${values}) workspace_scope(issue_id, workspace_id)`),
+        sql`blocker_workspace_facts.workspace_id = blocker_source.workspace_id`)
+      .where(pendingFinalizeBlockerCondition(companyId, sql`blocker_source.issue_id`, sql`blocker_source.workspace_id`));
+    for (const row of rows) pending.add(row.id);
   }
-
   return pending;
 }
 
@@ -3888,12 +3851,8 @@ async function readIssueBlockerAttentionBatch(
         BLOCKER_ATTENTION_MAX_NODES * 2 - edgeCount);
       if (remaining <= 0) break;
       const rowLimit = remaining + 1;
-      const explicitBlockerRowsPromise: Promise<
-        IssueBlockerAttentionQueryRow[]
-      > = dbOrTx
+      const explicitBlocker = dbOrTx
         .select({
-          issueId: issueRelations.relatedIssueId,
-          blockerIssueId: issues.id,
           id: issues.id,
           companyId: issues.companyId,
           parentId: issues.parentId,
@@ -3905,14 +3864,49 @@ async function readIssueBlockerAttentionBatch(
           assigneeAgentId: issues.assigneeAgentId,
           assigneeUserId: issues.assigneeUserId,
         })
+        // Resolve the globally unique relation target first. The company guard
+        // below remains authoritative, without a repeated company-index scan.
+        .from(issues).where(eq(issues.id, issueRelations.issueId))
+        .limit(1).as("explicit_blocker");
+      const workspaceScope = sql`select distinct candidate.execution_workspace_id
+        from issue_relations workspace_relation
+        inner join lateral (
+          select company_id, status, execution_workspace_id from issues
+          where id = workspace_relation.issue_id limit 1
+        ) candidate on true
+        where workspace_relation.company_id = ${companyId}::uuid
+          and workspace_relation.type = 'blocks'
+          and workspace_relation.related_issue_id in (${sql.join(chunk.map(id => sql`${id}::uuid`), sql`, `)})
+          and candidate.company_id = ${companyId}::uuid
+          and candidate.status = 'done' and candidate.execution_workspace_id is not null`;
+      const explicitBlockerRowsPromise: Promise<
+        IssueBlockerAttentionQueryRow[]
+      > = dbOrTx
+        .select({
+          issueId: issueRelations.relatedIssueId,
+          blockerIssueId: explicitBlocker.id,
+          id: explicitBlocker.id,
+          companyId: explicitBlocker.companyId,
+          parentId: explicitBlocker.parentId,
+          identifier: explicitBlocker.identifier,
+          title: explicitBlocker.title,
+          status: explicitBlocker.status,
+          executionRunId: explicitBlocker.executionRunId,
+          executionWorkspaceId: explicitBlocker.executionWorkspaceId,
+          assigneeAgentId: explicitBlocker.assigneeAgentId,
+          assigneeUserId: explicitBlocker.assigneeUserId,
+        })
         .from(issueRelations)
-        .innerJoin(issues, eq(issueRelations.issueId, issues.id))
+        .innerJoinLateral(explicitBlocker, sql`true`)
+        .leftJoin(blockerWorkspaceFacts(companyId, workspaceScope),
+          sql`blocker_workspace_facts.workspace_id = ${explicitBlocker.executionWorkspaceId}`)
         .where(
           and(
             eq(issueRelations.companyId, companyId),
             eq(issueRelations.type, "blocks"),
             inArray(issueRelations.relatedIssueId, chunk),
-            eq(issues.companyId, companyId),
+            eq(explicitBlocker.companyId, companyId),
+            or(ne(explicitBlocker.status, "done"), pendingFinalizeBlockerCondition(companyId, sql`${explicitBlocker.id}`, sql`${explicitBlocker.executionWorkspaceId}`)),
           ),
         ).limit(rowLimit);
       const childRowsPromise: Promise<IssueBlockerAttentionQueryRow[]> = dbOrTx

server/src/services/scoped-deadline-work.ts

@@ -12,7 +12,7 @@ export function registerScopedDeadlineWork<Entry extends ScheduledEntry, Scope>(
     queue: DeliveryQueue;
     batchSize?: number;
     minimumPassIntervalMs?: number;
-    reconstructionIntervalMs?: number;
+    reconstructionIntervalMs?: number | ((scope: Scope | undefined) => number);
     beforePass?(): Promise<void>;
     decodeScope(hint: string): Scope | undefined;
     matches(entry: Entry, scope?: Scope): boolean;
@@ -58,7 +58,7 @@ export function registerScopedDeadlineWork<Entry extends ScheduledEntry, Scope>(
     } },
     run: async signal => {
       if (!canRun() || Date.now() < earliest) return;
-      if ((options.reconstructionIntervalMs ?? 0) > 0) for (const [hint, until] of reconstructAfter) {
+      for (const [hint, until] of reconstructAfter) {
         if (until <= Date.now()) reconstructAfter.delete(hint);
       }
       const ready = [...dirty].filter(hint => Math.max(reconstructAfter.get(hint) ?? 0, scopeFailures.get(hint)?.retryAt ?? 0) <= Date.now());
@@ -94,8 +94,10 @@ export function registerScopedDeadlineWork<Entry extends ScheduledEntry, Scope>(
             entries.delete(entry.id); failures.delete(entry.id);
           }
           scans.delete(hint);
-          if ((options.reconstructionIntervalMs ?? 0) > 0) {
-            reconstructAfter.set(hint, Date.now() + options.reconstructionIntervalMs!);
+          const reconstructionInterval = typeof options.reconstructionIntervalMs === "function"
+            ? options.reconstructionIntervalMs(scope) : (options.reconstructionIntervalMs ?? 0);
+          if (reconstructionInterval > 0) {
+            reconstructAfter.set(hint, Date.now() + reconstructionInterval);
           }
           if ((versions.get(hint) ?? 0) !== scan.version) dirty.add(hint); else versions.delete(hint);
           if (!scope) scopeFailures.clear(); else scopeFailures.delete(hint);

server/src/services/workspace-operations.ts

@@ -1,3 +1,4 @@
+import { notifyAttentionRetentionWork, notifyIssueAttentionRetentionWork } from "./retention-work-notifications.js";
 import { notifyNativeResultCommit } from "./native-result-commit-notifications.js";
 import { notifyNativeWorkspaceFinalization } from "./native-workspace-finalization-notifications.js";
 import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js";
@@ -480,8 +481,14 @@ export function workspaceOperationService(db: Db) {
 
         async recordOperation(recordInput) {
           async function writeState<T>(write: (database: Db) => Promise<T>): Promise<T> {
-            if (recordInput.phase !== "workspace_finalize" || !input.heartbeatRunId) return write(db);
+            if (recordInput.phase !== "workspace_finalize" || (!input.heartbeatRunId && !executionWorkspaceId && !input.issueId)) return write(db);
             return db.transaction(async tx => {
+              if (!input.heartbeatRunId) {
+                // An export clears barriers for every issue sharing this workspace.
+                if (executionWorkspaceId) await notifyAttentionRetentionWork(tx, { companyId: input.companyId, sourceKind: "blocker_attention" });
+                else await notifyIssueAttentionRetentionWork(tx, { companyId: input.companyId, issueId: input.issueId ?? undefined });
+                return write(tx as unknown as Db);
+              }
               await notifyNativeResultCommit(tx, { companyId: input.companyId,
                 runId: input.heartbeatRunId!, issueId: input.issueId ?? undefined });
               await notifyNativeWorkspaceFinalization(tx, { companyId: input.companyId,