Merge qualified sleep review fixes and prerequisites

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

* commit '2d59dd1ba339f9ba55faaa8ae5763f4e7d8e5fb1':
  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

doc/plans/2026-10-30-attention-retention-sleep.md

@@ -50,3 +50,11 @@ Review the source parity fixtures and producer hints together. The intended outc
 Issue producer invalidation has its own finite discovery budget. Reverse parent/blocker traversal returns at most 101 distinct UUIDs in one recursive query, with each relation adjacency capped at 101 unique edges before materialization. At most 100 affected issues produce one sorted canonical batch hint; an overflow sentinel produces one company hint and the existing complete paged reconstruction. UNION deduplication terminates cycles. A PostgreSQL EXPLAIN ANALYZE on 2,100 reverse edges proves the outer limit stops at 101 CTE rows while the indexed relation branch returns only 100 rows. Repeated identical writes coalesce by canonical key. The shared scheduler is unchanged: a broader company hint does not cancel already pending narrower scopes; those remain finite batches of at most 100 IDs, processed at the existing 25-scope pass limit. This permits redundant bounded reconstruction instead of discarding invalidations or introducing cursor/subsumption policy across other owners. A 151-node cyclic chain proves overflow still registers the far ancestor's exact archive deadline and settles quietly afterward.
 
 The ordinary feed retains conservative `needs_attention` entries for roots outside the shared graph budget. The batch finalization visits every input root; a root without explored edges points attention at itself. A real PostgreSQL regression with 2,001 blocked roots verifies the nonempty deferred set, bounded graph size, ordinary issue-list metadata, and all 2,001 attention-feed subject IDs. This is a consumer parity proof; it does not increase graph traversal or change archive eligibility.
+
+## Completed blocker history and incomplete graph checks
+
+Explicit blocker queries apply the dependency owner's pending-finalization predicate before the graph row limit. A done blocker with no operation debt is excluded; attributed operations take precedence over legacy unattributed workspace operations, and a later successful export clears the barrier. The readiness helper and bounded attention query share this predicate, including the existing optional workspace/operation locks. Each relation resolves its globally unique issue identity before tenant and eligibility checks, avoiding repeated company-index scans.
+
+A fresh archive check consumes the private graph completeness result. An unexpanded frontier cannot establish absence of live work, so that check disarms without archiving. Supported issue, relation, run and workspace changes reactivate the root; restart reconstructs it. Completed-only history remains fully classified and can archive normally. No incomplete-root polling is added.
+
+The ordinary workspace-operation recorder also signals no-run finalization in its owning transaction. Export with a workspace identity invalidates that company's blocker source, including every root sharing the exported workspace; only a known issue without a workspace uses the existing reverse-issue hint. This rare source change uses bounded company/source paging and does not rebuild unrelated retention kinds. Run-backed finalization retains its existing native-status fanout. This lets a parked incomplete tree reactivate when export clears its done blockers, without an incomplete-root retry timer. Recorder rollback leaves the operation barrier and attention state unchanged.

doc/plans/2026-10-30-deferred-execution-sleep.md

@@ -32,3 +32,23 @@ Project budget admission locks and re-reads the company-scoped project before pu
 Native receipt checks share an opaque dispatcher created by the existing queue owner for exactly one bounded worker pass. Its canonical company/agent/issue/review scope map is the same map used by the manual dispatcher; the native worker explicitly admits at most100due entries per pass, matching the existing default batch while retaining company fairness. The next pass creates a fresh map. This restores same-pass grouping without introducing global or cross-invocation deduplication, changing claim authority, or consuming later notices. Each receipt still re-reads its own moved claim/error deadline before dispatch.
 
 Actual PostgreSQL controls reconstruct two saved native intents into one dispatch and shared run with real process stdout, cover the100-intent boundary, and keep another issue and company separate. A real issue-row lock holds the first admission while a separate transaction commits a new notice whose requestedAt is explicitly older than admission. The fixed current batch cannot consume that new identity; a fresh pass creates a distinct run, preserving the late notice even though its ordering metadata is old. Once all executions and stored deadlines settle,25virtual minutes produce zero driver SQL. The canonical dispatch body, including parsing, claims, deterministic receipt recovery and error handling, is unchanged apart from moving its map into the owner-created batch closure.
+
+
+### Browser fixture producer boundary
+
+The legacy recovery browser fixture used to insert a deferred request through a
+separate SQL connection after server startup, then wait for the removed poll.
+That is neither saved startup history nor a supported in-process producer. The
+automatic browser case now seeds while the company is paused, verifies the saved
+request and sole failed source run, and reactivates through the ordinary company
+API. No Retry or duplicate message is sent. It requires exactly one new run,
+one persisted agent response, the original queued comment in the process prompt,
+and the response after page reload. Startup reconstruction remains covered by
+the PostgreSQL case that persists history before worker registration.
+
+The corrected automatic path passes three consecutive built-UI/browser runs
+(2.1 minutes total). All six legacy continuation browser paths pass together
+(1.6 minutes), including task/inbox Retry, Retry after reload, ordinary message,
+queued Retry and automatic reactivation. The fixture also reads the configured
+external PostgreSQL connection when present, so local proof can reuse a dedicated
+test database instead of starting another database process.

doc/plans/2026-10-30-merged-pr-confirmations-sleep.md

@@ -104,6 +104,20 @@ The full existing 17-test file passes with these failure controls. Production
 configuration validation remains unchanged. Current-head required CI remains
 the completion gate.
 
+## Cancellation during acceptance
+
+The registered worker passes its optional cancellation signal into the canonical
+confirmation acceptance transaction. It rechecks after the issue lock, before
+mutation and before commit. Cancellation while an awaited write is blocked rolls
+back the interaction, audit and transactional delivery effects. Manual callers
+without a signal retain their existing acceptance and rollback behavior.
+
+A PostgreSQL lock wait itself remains unabortable by this signal: stop waits for
+the query to unwind. Once the lock is released, cancelled acceptance leaves the
+card pending for restart. The existing HTTP request timeout and prompt stop abort
+remain unchanged. Actual lock and write-trigger controls verify pending state,
+no post-stop wake, and exactly one acceptance/audit/wake on restart.
+
 ## Existing limitations outside this migration
 
 Direct non-chat continuation still uses the existing postcommit `wakeup` call.

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__/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/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/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

@@ -2522,124 +2522,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;
 }
 
@@ -3887,12 +3850,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,
@@ -3904,14 +3863,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/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,

tests/e2e/legacy-failure-continuation.spec.ts

@@ -11,14 +11,15 @@ async function json(response: APIResponse) {
   return response.json();
 }
 
-for (const action of ["task_retry", "retry_after_reload", "inbox_retry", "message", "queued_retry", "automatic_message"] as const) {
+for (const action of ["task_retry", "retry_after_reload", "inbox_retry", "message", "queued_retry", "automatic_message_on_reactivation"] as const) {
   test(`legacy startup hold: ${action} reaches a new agent response`, async ({ page, request }) => {
     test.setTimeout(120_000);
     const root = await mkdtemp(path.join(os.tmpdir(), "legacy-recovery-browser-"));
     const config = JSON.parse(await readFile(process.env.PAPERCLIP_E2E_SERVER_CONFIG!, "utf8"));
     // Use the running test server's actual port, including fallback allocation.
-    const pid = await readFile(path.join(config.database.embeddedPostgresDataDir, "postmaster.pid"), "utf8");
-    const url = `postgres://paperclip:paperclip@127.0.0.1:${pid.split("\n")[3]}/paperclip`;
+    const url = config.database.mode === "postgres" ? config.database.connectionString
+      : `postgres://paperclip:paperclip@127.0.0.1:${(await readFile(
+        path.join(config.database.embeddedPostgresDataDir, "postmaster.pid"), "utf8")).split("\n")[3]}/paperclip`;
     const db = createDb(url);
     const company = await json(await request.post("/api/companies", { data: { name: `Legacy recovery ${action} ${Date.now()}` } }));
     try {
@@ -37,8 +38,11 @@ for (const action of ["task_retry", "retry_after_reload", "inbox_retry", "messag
         title: "Continue after startup failure", description: "Answer the pending follow-up once.",
         status: "backlog", assigneeAgentId: agent.id,
       } }));
+      if (action === "automatic_message_on_reactivation") {
+        await json(await request.patch(`/api/companies/${company.id}`, { data: { status: "paused" } }));
+      }
       const sourceRunId = randomUUID();
-      // Seed the historical incident, then exercise all recovery through the UI.
+      // Seed the historical incident, then exercise ordinary recovery entry points.
       // No adapter.invoke or new dispatch identity exists on this pre-upgrade run.
       await db.insert(heartbeatRuns).values({ id: sourceRunId, companyId: company.id, agentId: agent.id,
         status: "failed", runtimeMode: "legacy", processPid: action === "queued_retry" ? process.pid : 999999999,
@@ -58,7 +62,7 @@ for (const action of ["task_retry", "retry_after_reload", "inbox_retry", "messag
         await db.insert(companyMemberships).values({ companyId: company.id, principalType: "user",
           principalId: "original-board", membershipRole: "operator", status: "active" });
       }
-      if (action === "queued_retry" || action === "automatic_message") {
+      if (action === "queued_retry" || action === "automatic_message_on_reactivation") {
         const commentId = randomUUID();
         // Reproduce a real user comment saved while the failed run was active,
         // including queues originally created by a system wake.
@@ -104,8 +108,19 @@ for (const action of ["task_retry", "retry_after_reload", "inbox_retry", "messag
           retry.click(),
         ]);
         expect(response.ok(), `${response.status()} ${await response.text()}`).toBe(true);
-      } else if (action === "automatic_message") {
-        // No Retry, duplicate message, or status change: the saved input runs.
+      } else if (action === "automatic_message_on_reactivation") {
+        // This fixture writes history through a separate SQL connection after boot.
+        // Reactivation is the supported producer that makes saved work eligible;
+        // startup reconstruction itself is covered before registration in the PG suite.
+        const pending = await db.select().from(agentWakeupRequests).where(and(
+          eq(agentWakeupRequests.companyId, company.id), eq(agentWakeupRequests.agentId, agent.id),
+          eq(agentWakeupRequests.status, "deferred_issue_execution"),
+        ));
+        expect(pending).toHaveLength(1);
+        const before = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, company.id));
+        expect(before).toHaveLength(1);
+        await json(await request.patch(`/api/companies/${company.id}`, { data: { status: "active" } }));
+        // No Retry or duplicate message: the existing saved input runs once.
       } else if (action === "message") {
         await page.getByRole("textbox", { name: "editable markdown" }).fill("Please continue the pending follow-up.");
         await page.getByRole("button", { name: "Send", exact: true }).click();
@@ -133,7 +148,7 @@ for (const action of ["task_retry", "retry_after_reload", "inbox_retry", "messag
       expect(runs.filter(run => run.id !== sourceRunId)).toHaveLength(1);
       expect(runs.find(run => run.id === sourceRunId)).toMatchObject({ status: "failed", resultJson: null });
       const prompts = await readFile(path.join(root, "prompts"), "utf8");
-      if (action === "queued_retry" || action === "automatic_message") {
+      if (action === "queued_retry" || action === "automatic_message_on_reactivation") {
         expect(prompts).toContain("Approved");
         await expect(page.getByRole("button", { name: "Interrupt", exact: true })).toHaveCount(0);
       }