fix(server): bound retention reconstruction and preserve producer transactions

Batch review and confirmation eligibility per company, scope decision and budget hints, and keep child blocker relation intents inside their owning transaction. Cover ordinary child, budget and execution producers plus full candidate-page query bounds; update startup/import fixtures for source-driven retention.

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

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

@@ -25,7 +25,7 @@ The feed and worker share confirmation collapse, newer-run probes, issue visibil
 
 ## Existing mutation boundaries reactivate work
 
-Committed hints identify a company, source kind, and source ID where those values are known. Retention's own writes use exact source hints; blocker retention changes invalidate the company/source because multiple blocked roots can project one terminal subject. Reconstruction pages and due batches remain bounded.
+Committed hints identify a company, source kind, and source ID where those values are known. Retention's own writes use exact source hints; blocker retention changes invalidate the company/source because multiple blocked roots can project one terminal subject. Reconstruction pages and due batches remain bounded. Review attention and interaction invokability are computed in company batches per candidate page; a full 100-candidate review page uses at most 20 statements and 100 confirmations use at most six. Ordinary execution, decision and budget mutations pass their affected company; exact decision sources are included.
 
 Producer coverage includes decision creation and changes; interaction creation, resolution, expiry and supersession; issue creation, updates, blocker relations, checkout, adoption and release; agent hire, configuration, lifecycle, error clearing and deletion; budget incidents and company pause/resume; exhausted-run events and retry-run creation; native status, review, finalization and reconstruction; external conversation transitions; and imported blocker relations. Existing service transactions retain outer commit ownership. Savepoint release does not publish work before outer commit.
 
@@ -33,7 +33,7 @@ The writer contract follows the existing delivery workers: serving processes use
 
 ## Real PostgreSQL evidence proves quiet and reactivation
 
-The new acceptance suite exercises all eight sources, including terminal blocker identity and date, newest confirmations, independent quality gates, unhealthy reporting chains, conversation visibility, maintained reviews and pending approvals, newer-run suppression, ordinary task and interaction producers, ordinary exhaustion-event ingress, and ordinary error clearing.
+The new acceptance suite exercises all eight sources, including terminal blocker identity and date, newest confirmations, independent quality gates, unhealthy reporting chains, conversation visibility, maintained reviews and pending approvals, newer-run suppression, ordinary task and interaction producers, ordinary exhaustion-event ingress, and ordinary error clearing. Additional ordinary producer controls cover child creation with a parent-blocking relation, budget incident creation and resolution, and repeated company-scoped execution-control writes.
 
 It also checks keep/revive, moved and cancelled deadlines, company activation and isolation, retained terminal history, restart, nested outer commit and rollback, first writes during drain, retry across drain, atomic archive/outbox rollback, and notifications during reconstruction. A 225-item backlog uses three startup pages and source-specific self-invalidations. SQL is counted at the PostgreSQL driver boundary.
 

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

@@ -2,9 +2,12 @@ import { randomUUID } from "node:crypto";
 import { and, eq, sql } from "drizzle-orm";
 import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
 import { authUsers, companyMemberships, agents, approvals, budgetIncidents, budgetPolicies, companies, createDb, applyPendingMigrations, decisionArchiveNotificationOutbox,
-  decisionRetention, decisions, heartbeatRunEvents, heartbeatRuns, issueApprovals, issueRecoveryActions,
+  decisionRetention, decisions, costEvents, heartbeatRunEvents, heartbeatRuns, issueApprovals, issueRecoveryActions,
   issueRelations, issueThreadInteractions, issues, type Db } from "@paperclipai/db";
 import { attentionRetentionService, registerAttentionRetentionWork } from "../services/attention-retention.js";
+import { budgetService } from "../services/budgets.js";
+import { writeExecutionControl } from "../services/execution-control-notifications.js";
+import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js";
 import { attentionService } from "../services/attention.js";
 import { companyService } from "../services/companies.js";
 import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
@@ -135,6 +138,76 @@ const delay = DEFAULT_DECISION_ARCHIVE_DAYS * 86_400_000;
     await coordinator!.stop(); coordinator = undefined; await start(); await quiet();
   });
 
+  it("registers the ordinary createChild blocker path inside its relation transaction", async () => {
+    const parent = await issueService(db).create(companyId, { title: "Parent", status: "in_progress", assigneeAgentId: agentId, createdByAgentId: agentId });
+    await start(); await quiet();
+    const { issue: child, parentBlockerAdded } = await issueService(db).createChild(parent.id, {
+      title: "Child needs review", status: "in_review", assigneeUserId: "board", createdByAgentId: agentId,
+      blockParentUntilDone: true,
+    });
+    expect(parentBlockerAdded).toBe(true);
+    expect(await db.select().from(issueRelations).where(and(eq(issueRelations.issueId, child.id),
+      eq(issueRelations.relatedIssueId, parent.id)))).toHaveLength(1);
+    await settle(); expect(coordinator!.nextWakeAt()).not.toBeNull();
+    await vi.advanceTimersByTimeAsync(delay + 2000); await archived("review", child.id); await quiet();
+    expect(await db.select().from(decisionArchiveNotificationOutbox).where(eq(decisionArchiveNotificationOutbox.sourceId, child.id))).toHaveLength(1);
+  });
+
+  it("batches a full page of maintained reviews and confirmations by company", async () => {
+    const rows = await db.insert(issues).values(Array.from({ length: 100 }, () => ({ companyId,
+      title: randomUUID(), status: "in_review", assigneeAgentId: agentId, updatedAt: old() }))).returning();
+    await db.insert(issueThreadInteractions).values(rows.map(row => ({ companyId, issueId: row.id,
+      kind: "request_confirmation", addresseeAgentId: agentId, payload: { version: 1, prompt: "Agent verdict" } })));
+    statements.length = 0;
+    const page = await attentionRetentionService(db).load({ companyId, sourceKind: "review" });
+    expect(page.rows).toEqual([]); expect(page.complete).toBe(false);
+    expect(statements.length).toBeLessThanOrEqual(20);
+    expect(statements.filter(query => query.includes('from "agents"')).length).toBeLessThanOrEqual(1);
+    await db.update(agents).set({ status: "paused" }).where(eq(agents.id, agentId));
+    statements.length = 0;
+    const interactions = await attentionRetentionService(db).load({ companyId, sourceKind: "issue_thread_interaction" });
+    expect(interactions.rows).toHaveLength(100); expect(interactions.complete).toBe(false);
+    expect(statements.length).toBeLessThanOrEqual(6);
+    expect(statements.filter(query => query.includes('from "agents"'))).toHaveLength(1);
+  });
+
+  it("scopes ordinary execution-control writes to their company", async () => {
+    const otherCompany = randomUUID();
+    await db.insert(companies).values({ id: otherCompany, name: "Other", issuePrefix: "OTHER" });
+    await db.insert(issues).values(Array.from({ length: 125 }, () => ({ companyId: otherCompany,
+      title: randomUUID(), status: "in_review", assigneeUserId: "board", updatedAt: old() })));
+    const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId, status: "running" }).returning();
+    const service = attentionRetentionService(db), load = vi.spyOn(service, "load");
+    await start(service); await quiet(); load.mockClear();
+    for (let count = 0; count < 3; count++) {
+      await writeExecutionControl(db, DELIVERY_QUEUES.executionFinalization,
+        tx => tx.update(heartbeatRuns).set({ updatedAt: new Date() }).where(eq(heartbeatRuns.id, run.id)), { companyId });
+      await quiet();
+    }
+    expect(load).toHaveBeenCalled();
+    expect(load.mock.calls.every(([scope]) => scope?.companyId === companyId)).toBe(true);
+  });
+
+  it("registers and cancels budget deadlines through ordinary policy and incident producers", async () => {
+    const service = budgetService(db);
+    await service.upsertPolicy(companyId, { scopeType: "agent", scopeId: agentId, amount: 100,
+      warnPercent: 85, hardStopEnabled: false, notifyEnabled: true }, "board");
+    const retention = attentionRetentionService(db), load = vi.spyOn(retention, "load");
+    await start(retention); await quiet(); load.mockClear();
+    const [event] = await db.insert(costEvents).values({ companyId, agentId, provider: "openai", biller: "openai",
+      billingType: "metered_api", model: "fixture", inputTokens: 100, outputTokens: 20,
+      costCents: 90, occurredAt: new Date() }).returning();
+    await service.evaluateCostEvent(event);
+    await settle(); expect(coordinator!.nextWakeAt()).not.toBeNull();
+    const [incident] = await db.select().from(budgetIncidents).where(eq(budgetIncidents.companyId, companyId));
+    expect(incident).toMatchObject({ status: "open", thresholdType: "soft" });
+    expect(await state("budget_alert", incident.id)).toBeNull();
+    await service.upsertPolicy(companyId, { scopeType: "agent", scopeId: agentId, amount: 200 }, "board");
+    await quiet();
+    expect((await db.select().from(budgetIncidents).where(eq(budgetIncidents.id, incident.id)))[0].status).toBe("resolved");
+    expect(load.mock.calls.every(([scope]) => scope?.companyId === companyId)).toBe(true);
+  });
+
   it("matches the existing feed across all eight remaining source kinds and exclusion variants", async () => {
     const origin = await insertIssue();
     const review = await insertIssue("in_review", { assigneeUserId: "board" });

server/src/__tests__/company-portability.test.ts

@@ -4184,6 +4184,7 @@ describe("company portability", () => {
       }),
       // eslint-disable-next-line @typescript-eslint/no-explicit-any
     } as any;
+    db.transaction = async (write: (tx: typeof db) => Promise<unknown>) => write(db);
     return { db, insertedRelationValues, monitorUpdates };
   }
 

server/src/__tests__/decision-effect-recovery-work.test.ts

@@ -73,7 +73,7 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes
   const claim = (id: string, updatedAt = new Date(), owner = db) => writeDecisionChange(owner, async tx => tx.update(decisions).set({
     status: "decided", executionStatus: "running", chosenOptionId: "yes", decidedByUserId: userId,
     updatedAt, inputValues: {}, metadata: { continuationPending: true },
-  }).where(eq(decisions.id, id)));
+  }).where(eq(decisions.id, id)), { companyId });
   async function start() {
     coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining, canReconcile: () => true,
       onError: error => errors.push(error) });
@@ -467,7 +467,7 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes
     } else {
       await writeDecisionChange(db, async tx => tx.update(decisions).set({ status: "decided", executionStatus: "running",
         chosenOptionId: "archive", decidedByUserId: userId, updatedAt: new Date(Date.now() - 120_000),
-        metadata: { ...row.metadata, continuationPending: true } }).where(eq(decisions.id, row.id)));
+        metadata: { ...row.metadata, continuationPending: true } }).where(eq(decisions.id, row.id)), { companyId });
     }
     await start(); await settled(row.id, committed ? "failed" : "succeeded"); await quiet();
     const [retained] = await db.select().from(decisionRetention);

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

@@ -793,7 +793,7 @@ describePg("decisionService", () => {
       expect((await service().get(id))?.metadata.continuationPending).toBe(false);
     }
     const markReady = (id: string, source = db) => writeDecisionChange(source, async tx => tx.update(decisions)
-      .set({ status: "cancelled", metadata: { continuationPending: true } }).where(eq(decisions.id, id)));
+      .set({ status: "cancelled", metadata: { continuationPending: true } }).where(eq(decisions.id, id)), { companyId });
 
     it("coalesces sequential tenant writes without dropping pending recovery", async () => {
       const row = await createCommentDecision();

server/src/__tests__/server-startup-feedback-export.test.ts

@@ -1,3 +1,4 @@
+import { attentionRetentionService } from "../services/attention-retention.js";
 import { nativeHarnessBackupExpirationService } from "../services/native-runtime/native-harness-backup-expiration.js";
 import { skillVersionRetentionService } from "../services/skill-version-retention.js";
 import { approvalQueueSeedingService } from "../services/approval-queue-seeding-work.js";
@@ -377,6 +378,11 @@ vi.mock("../realtime/live-events-ws.js", () => ({
   setupLiveEventsWebSocketServer: vi.fn(),
 }));
 
+vi.mock("../services/attention-retention.js", async importOriginal => ({
+  ...await importOriginal<typeof import("../services/attention-retention.js")>(),
+  attentionRetentionService: vi.fn(() => ({ load: vi.fn(async () => ({ rows: [], complete: true })), check: vi.fn() })),
+}));
+
 vi.mock("../services/join-request-retention.js", async importOriginal => ({
   ...await importOriginal<typeof import("../services/join-request-retention.js")>(),
   joinRequestRetentionService: vi.fn(() => ({ sweep: vi.fn(async () => ({ archived: 0 })), next: vi.fn(async () => null) })),
@@ -974,7 +980,7 @@ describe("startServer feedback export wiring", () => {
     }
   });
 
-  it.each([false, true])("registers archive delivery separately from retention sweeps (scheduler %s)", async enabled => {
+  it.each([false, true])("registers source retention separately from archive delivery without tick scans (scheduler %s)", async enabled => {
     loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled, heartbeatSchedulerIntervalMs: 30000 }));
     let tick: (() => void) | undefined;
     const timer = vi.spyOn(globalThis, "setInterval").mockImplementation(((callback: () => void, interval: number) => {
@@ -986,12 +992,17 @@ describe("startServer feedback export wiring", () => {
       const registration = deliveryWorkRegisterMock.mock.calls.find(([queue]) => queue === DELIVERY_QUEUES.decisionArchiveDelivery);
       if (!enabled) {
         expect(registration).toBeUndefined();
+        expect(attentionRetentionService).not.toHaveBeenCalled();
         expect(decisionRetentionService).not.toHaveBeenCalled();
         expect(tick).toBeDefined(); tick!(); await settleStartupHousekeeping();
         expect(decisionRetentionService).not.toHaveBeenCalled();
         return;
       }
       expect(registration).toBeDefined();
+      expect(deliveryWorkRegisterMock.mock.calls.some(([queue]) => queue === DELIVERY_QUEUES.attentionRetention)).toBe(true);
+      expect(attentionRetentionService).toHaveBeenCalledOnce();
+      const source = vi.mocked(attentionRetentionService).mock.results[0]!.value;
+      expect(source.load).toHaveBeenCalledOnce();
       const retention = vi.mocked(decisionRetentionService).mock.results[0]!.value;
       expect(retention.deliverNotifications).toHaveBeenCalledOnce();
       expect(retention.nextArchiveNotificationAt).toHaveBeenCalledOnce();
@@ -999,9 +1010,10 @@ describe("startServer feedback export wiring", () => {
       const companyReads = () => vi.mocked(db.select).mock.results.filter(({ value }) =>
         vi.mocked(value.from).mock.calls.some(([table]) => table === companies)).length;
       const before = companyReads();
-      expect(before).toBeGreaterThan(0); expect(tick).toBeDefined();
+      expect(before).toBe(0); expect(tick).toBeDefined();
       tick!(); await settleStartupHousekeeping();
-      expect(companyReads()).toBeGreaterThan(before);
+      expect(companyReads()).toBe(before);
+      expect(source.load).toHaveBeenCalledOnce();
       expect(retention.deliverNotifications).toHaveBeenCalledOnce();
       await registration![1].run(new AbortController().signal);
       expect(retention.deliverNotifications).toHaveBeenCalledTimes(2);

server/src/services/attention-retention.ts

@@ -1,13 +1,14 @@
-import { and, eq, isNull, sql, type SQL } from "drizzle-orm";
+import { and, eq, inArray, isNull, sql, type SQL } from "drizzle-orm";
 import { agents, decisionRetention, decisions, heartbeatRuns,
   issueApprovals, issueRecoveryActions, issueThreadInteractions, issues, approvals, type Db } from "@paperclipai/db";
 import type { AttentionSourceKind } from "@paperclipai/shared";
 import { evaluateAgentInvokability } from "./agent-invokability.js";
-import { collapsePendingConfirmationsToNewest, listFailedRunsWithNewerRun, readRunIssueId, requiresReviewAttention, terminalBlockerRequiresAttention } from "./attention-source-eligibility.js";
+import { HUMAN_RECOVERY_OWNER_TYPES, ATTENTION_BUDGET_WARNING_PERCENT, ATTENTION_AGENT_ERROR_STATUS, collapsePendingConfirmationsToNewest, listFailedRunsWithNewerRun, readRunIssueId, requiresReviewAttention, terminalBlockerRequiresAttention } from "./attention-source-eligibility.js";
 import { decisionRetentionService, DEFAULT_DECISION_ARCHIVE_DAYS } from "./decision-retention.js";
 import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js";
 import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
 import { attentionIssueCondition } from "./issue-visibility.js";
+import { RETRY_EXHAUSTION_MESSAGE_PREFIX } from "./heartbeat-run-events.js";
 import { issueService } from "./issues.js";
 import { parseIssueExecutionState } from "./issue-execution-policy.js";
 import { isProspectiveBlockedTransition } from "./routable-blocked.js";
@@ -34,7 +35,7 @@ const sources: Record<Kind, SQL> = {
   issue_thread_interaction: sql`select 'issue_thread_interaction', i.id::text, i.company_id, i.updated_at
     from issue_thread_interactions i where i.status = 'pending' and i.addressee_user_id is null`,
   recovery_action: sql`select 'recovery_action', r.id::text, r.company_id, r.updated_at
-    from issue_recovery_actions r where r.status in ('active', 'escalated') and r.owner_type in ('user', 'board')`,
+    from issue_recovery_actions r where r.status in ('active', 'escalated') and r.owner_type in (${sql.join(HUMAN_RECOVERY_OWNER_TYPES.map(owner => sql`${owner}`), sql`, `)})`,
   blocker_attention: sql`select 'blocker_attention', i.id::text, i.company_id, i.updated_at
     from issues i where i.status = 'blocked' and i.hidden_at is null and i.harness_kind is null`,
   review: sql`select 'review', i.id::text, i.company_id, i.updated_at
@@ -42,11 +43,11 @@ const sources: Record<Kind, SQL> = {
   failed_run: sql`select 'failed_run', r.id::text, r.company_id, coalesce(r.finished_at, r.updated_at, r.created_at)
     from heartbeat_runs r where r.status in ('failed', 'timed_out') and exists (
       select 1 from heartbeat_run_events e where e.run_id = r.id and e.company_id = r.company_id
-        and e.event_type = 'lifecycle' and e.message like 'Bounded retry exhausted%')`,
+        and e.event_type = 'lifecycle' and e.message like ${`${RETRY_EXHAUSTION_MESSAGE_PREFIX}%`})`,
   budget_alert: sql`select 'budget_alert', b.id::text, b.company_id, b.updated_at from budget_incidents b
     where b.status = 'open' and (b.threshold_type = 'hard' or (b.amount_limit > 0
-      and round(b.amount_observed::numeric / b.amount_limit * 10000) / 100 >= 85))`,
-  agent_error_alert: sql`select 'agent_error_alert', a.id::text, a.company_id, a.updated_at from agents a where a.status = 'error'`,
+      and round(b.amount_observed::numeric / b.amount_limit * 10000) / 100 >= ${ATTENTION_BUDGET_WARNING_PERCENT}))`,
+  agent_error_alert: sql`select 'agent_error_alert', a.id::text, a.company_id, a.updated_at from agents a where a.status = ${ATTENTION_AGENT_ERROR_STATUS}`,
 };
 
 export function attentionRetentionService(db: Db) {
@@ -74,7 +75,69 @@ export function attentionRetentionService(db: Db) {
     return row;
   }
 
-  async function eligible(database: Db, candidate: Candidate, variant?: "owner" | "terminal", lock = false): Promise<Entry | null> {
+  async function interactionEligibility(database: Db, candidates: Candidate[]) {
+    const result = new Map<string, boolean>();
+    const byCompany = new Map<string, string[]>();
+    for (const candidate of candidates) {
+      if (candidate.sourceKind !== "issue_thread_interaction") continue;
+      const ids = byCompany.get(candidate.companyId) ?? [];
+      ids.push(candidate.sourceId);
+      byCompany.set(candidate.companyId, ids);
+    }
+    for (const [companyId, ids] of byCompany) {
+      const rows = await database.select().from(issueThreadInteractions).where(and(
+        eq(issueThreadInteractions.companyId, companyId), inArray(issueThreadInteractions.id, ids),
+        eq(issueThreadInteractions.status, "pending"), isNull(issueThreadInteractions.addresseeUserId)));
+      if (!rows.length) continue;
+      const visible = await database.select({ id: issues.id }).from(issues).where(and(eq(issues.companyId, companyId),
+        inArray(issues.id, rows.map(row => row.issueId)), attentionIssueCondition(true)));
+      const visibleIds = new Set(visible.map(row => row.id));
+      const roster = await database.select().from(agents).where(eq(agents.companyId, companyId));
+      const boardAddressed = (row: typeof issueThreadInteractions.$inferSelect) => !row.addresseeAgentId
+        || !evaluateAgentInvokability(roster.find(agent => agent.id === row.addresseeAgentId), roster).invokable;
+      const confirmationIssueIds = rows.filter(row => row.kind === "request_confirmation").map(row => row.issueId);
+      const siblings = confirmationIssueIds.length ? await database.select().from(issueThreadInteractions).where(and(
+        eq(issueThreadInteractions.companyId, companyId), inArray(issueThreadInteractions.issueId, confirmationIssueIds),
+        eq(issueThreadInteractions.status, "pending"), eq(issueThreadInteractions.kind, "request_confirmation"),
+        isNull(issueThreadInteractions.addresseeUserId))) : [];
+      const newest = new Set(collapsePendingConfirmationsToNewest(siblings.filter(boardAddressed)).map(row => row.id));
+      for (const row of rows) result.set(row.id, visibleIds.has(row.issueId) && boardAddressed(row)
+        && (row.kind !== "request_confirmation" || newest.has(row.id)));
+    }
+    return result;
+  }
+
+  async function reviewEligibility(database: Db, candidates: Candidate[]) {
+    const result = new Map<string, boolean>();
+    const byCompany = new Map<string, string[]>();
+    for (const candidate of candidates) {
+      if (candidate.sourceKind !== "review") continue;
+      const ids = byCompany.get(candidate.companyId) ?? [];
+      ids.push(candidate.sourceId);
+      byCompany.set(candidate.companyId, ids);
+    }
+    for (const [companyId, ids] of byCompany) {
+      const rows = await database.select().from(issues).where(and(eq(issues.companyId, companyId),
+        inArray(issues.id, ids), attentionIssueCondition(false)));
+      const pendingApprovals = await database.select({ issueId: issueApprovals.issueId }).from(issueApprovals)
+        .innerJoin(approvals, eq(approvals.id, issueApprovals.approvalId)).where(and(
+          eq(issueApprovals.companyId, companyId), eq(approvals.companyId, companyId),
+          inArray(issueApprovals.issueId, ids), eq(approvals.status, "pending")));
+      const approved = new Set(pendingApprovals.map(row => row.issueId));
+      const stalled = await issueService(database).listReviewAttention(companyId, rows);
+      for (const row of rows) {
+        const state = parseIssueExecutionState(row.executionState);
+        result.set(row.id, requiresReviewAttention({
+          hasHumanParticipant: state?.status === "pending" && state.currentParticipant?.type === "user",
+          hasUserAssignee: Boolean(row.assigneeUserId), hasPendingApproval: approved.has(row.id),
+          stalled: stalled.get(row.id)?.state === "stalled",
+        }));
+      }
+    }
+    return result;
+  }
+
+  async function eligible(database: Db, candidate: Candidate, variant?: "owner" | "terminal", lock = false, reviewAttention?: Map<string, boolean>, interactionAttention?: Map<string, boolean>): Promise<Entry | null> {
     let sourceId = candidate.sourceId;
     let activityAt = candidate.activityAt;
     let dueAt = candidate.deadlineAt;
@@ -83,38 +146,14 @@ export function attentionRetentionService(db: Db) {
         .where(and(eq(decisions.companyId, candidate.companyId), eq(decisions.id, sourceId), eq(decisions.status, "open")));
       if (!row || row.originIssueId && !await visibleIssue(database, candidate.companyId, row.originIssueId)) return null;
     } else if (candidate.sourceKind === "issue_thread_interaction") {
-      const [row] = await database.select().from(issueThreadInteractions).where(and(
-        eq(issueThreadInteractions.companyId, candidate.companyId), eq(issueThreadInteractions.id, sourceId),
-        eq(issueThreadInteractions.status, "pending"), isNull(issueThreadInteractions.addresseeUserId)));
-      if (!row || !await visibleIssue(database, candidate.companyId, row.issueId, true)) return null;
-      const roster = row.addresseeAgentId ? await database.select().from(agents).where(eq(agents.companyId, candidate.companyId)) : [];
-      if (row.addresseeAgentId && evaluateAgentInvokability(roster.find(agent => agent.id === row.addresseeAgentId), roster).invokable) return null;
-      if (row.kind === "request_confirmation") {
-        const siblings = await database.select().from(issueThreadInteractions).where(and(
-          eq(issueThreadInteractions.companyId, candidate.companyId), eq(issueThreadInteractions.issueId, row.issueId),
-          eq(issueThreadInteractions.status, "pending"), eq(issueThreadInteractions.kind, "request_confirmation"),
-          isNull(issueThreadInteractions.addresseeUserId)));
-        const companyAgents = roster.length ? roster : await database.select().from(agents).where(eq(agents.companyId, candidate.companyId));
-        const board = siblings.filter(sibling => !sibling.addresseeAgentId
-          || !evaluateAgentInvokability(companyAgents.find(agent => agent.id === sibling.addresseeAgentId), companyAgents).invokable);
-        if (!collapsePendingConfirmationsToNewest(board).some(sibling => sibling.id === sourceId)) return null;
-      }
+      const attention = interactionAttention ?? await interactionEligibility(database, [candidate]);
+      if (!attention.get(sourceId)) return null;
     } else if (candidate.sourceKind === "recovery_action") {
       const [row] = await database.select().from(issueRecoveryActions).where(and(eq(issueRecoveryActions.companyId, candidate.companyId), eq(issueRecoveryActions.id, sourceId)));
       if (!row || !await visibleIssue(database, candidate.companyId, row.sourceIssueId)) return null;
     } else if (candidate.sourceKind === "review") {
-      const row = await visibleIssue(database, candidate.companyId, sourceId);
-      if (!row) return null;
-      const state = parseIssueExecutionState(row.executionState);
-      const [approval] = await database.select({ id: approvals.id }).from(issueApprovals).innerJoin(approvals, eq(approvals.id, issueApprovals.approvalId))
-        .where(and(eq(issueApprovals.companyId, candidate.companyId), eq(approvals.companyId, candidate.companyId),
-          eq(issueApprovals.issueId, sourceId), eq(approvals.status, "pending"))).limit(1);
-      if (!requiresReviewAttention({ hasHumanParticipant: state?.status === "pending" && state.currentParticipant?.type === "user",
-        hasUserAssignee: Boolean(row.assigneeUserId), hasPendingApproval: Boolean(approval), stalled: false })) {
-        const attention = await issueService(database).listReviewAttention(candidate.companyId, [row]);
-        if (!requiresReviewAttention({ hasHumanParticipant: false, hasUserAssignee: false, hasPendingApproval: false,
-          stalled: attention.get(sourceId)?.state === "stalled" })) return null;
-      }
+      const attention = reviewAttention ?? await reviewEligibility(database, [candidate]);
+      if (!attention.get(sourceId)) return null;
     } else if (candidate.sourceKind === "blocker_attention") {
       const row = await visibleIssue(database, candidate.companyId, sourceId);
       if (!row) return null;
@@ -157,10 +196,12 @@ export function attentionRetentionService(db: Db) {
     const key = JSON.stringify(scope ?? {});
     const rows = await candidates(db, scope, cursors.get(key));
     const entries: Entry[] = [];
+    const reviews = await reviewEligibility(db, rows);
+    const interactions = await interactionEligibility(db, rows);
     for (const row of rows) {
       if (signal?.aborted) return { rows: null, complete: true };
       for (const variant of row.sourceKind === "blocker_attention" ? ["owner", "terminal"] as const : [undefined]) {
-        const entry = await eligible(db, row, variant);
+        const entry = await eligible(db, row, variant, false, reviews, interactions);
         if (entry) entries.push(entry);
       }
     }

server/src/services/attention-source-eligibility.ts

@@ -1,6 +1,10 @@
 import { and, eq, gt, inArray, sql, type SQL } from "drizzle-orm";
 import { heartbeatRuns, type Db } from "@paperclipai/db";
 
+export const HUMAN_RECOVERY_OWNER_TYPES = ["user", "board"] as const;
+export const ATTENTION_BUDGET_WARNING_PERCENT = 85;
+export const ATTENTION_AGENT_ERROR_STATUS = "error";
+
 export function collapsePendingConfirmationsToNewest<T extends {
   id: string;
   issueId: string;

server/src/services/attention.ts

@@ -1,4 +1,4 @@
-import { collapsePendingConfirmationsToNewest, readRunIssueId, listFailedRunsWithNewerRun, requiresReviewAttention, terminalBlockerRequiresAttention } from "./attention-source-eligibility.js";
+import { HUMAN_RECOVERY_OWNER_TYPES, ATTENTION_BUDGET_WARNING_PERCENT, ATTENTION_AGENT_ERROR_STATUS, collapsePendingConfirmationsToNewest, readRunIssueId, listFailedRunsWithNewerRun, requiresReviewAttention, terminalBlockerRequiresAttention } from "./attention-source-eligibility.js";
 import { approvalExecutionIssueCondition, firstApprovalIssueId } from "./approval-visibility.js";
 import { and, asc, desc, eq, inArray, isNull, notInArray, or, sql, type SQL } from "drizzle-orm";
 import type { Db } from "@paperclipai/db";
@@ -108,7 +108,6 @@ const SOURCE_RANK: Record<AttentionSourceKind, number> = {
 
 const PENDING_INTERACTION_STATUSES = ["pending"] as const;
 const OPEN_RECOVERY_STATUSES = ["active", "escalated"] as const;
-const HUMAN_RECOVERY_OWNER_TYPES = ["user", "board"] as const;
 const DETAIL_EXCERPT_LENGTH = 160;
 const DETAIL_IMAGE_LIMIT = 3;
 const OPEN_DECISION_DEFAULT_LIMIT = 500;
@@ -1762,7 +1761,7 @@ export function attentionService(db: Db, serviceOptions: AttentionServiceOptions
       const budgetOverview = await budgetService(db).overview(companyId);
       for (const incident of budgetOverview.activeIncidents) {
         const observedPercent = budgetObservedPercent(incident.amountObserved, incident.amountLimit);
-        if (incident.thresholdType !== "hard" && observedPercent < 85) continue;
+        if (incident.thresholdType !== "hard" && observedPercent < ATTENTION_BUDGET_WARNING_PERCENT) continue;
         const dedupKey = `budget:${incident.policyId}:${toIso(incident.windowStart)}:${incident.thresholdType}`;
         add(createItem({
           companyId,
@@ -1825,7 +1824,7 @@ export function attentionService(db: Db, serviceOptions: AttentionServiceOptions
           updatedAt: agents.updatedAt,
         })
         .from(agents)
-        .where(and(eq(agents.companyId, companyId), eq(agents.status, "error")))
+        .where(and(eq(agents.companyId, companyId), eq(agents.status, ATTENTION_AGENT_ERROR_STATUS)))
         .orderBy(desc(agents.updatedAt), desc(agents.id));
 
       for (const agent of erroredAgents) {

server/src/services/budgets.ts

@@ -505,8 +505,8 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica
     return incident ? { incident, created: true } : null;
   }
 
-  async function resolveOpenSoftIncidents(policyId: string) {
-    await notifyAttentionRetentionWork(db, { sourceKind: "budget_alert" });
+  async function resolveOpenSoftIncidents(policyId: string, companyId: string) {
+    await notifyAttentionRetentionWork(db, { companyId, sourceKind: "budget_alert" });
     await db
       .update(budgetIncidents)
       .set({
@@ -525,11 +525,12 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica
 
   async function resolveOpenIncidentsForPolicy(
     policyId: string,
+    companyId: string,
     approvalStatus: "approved" | "rejected" | null,
     decidedByUserId: string | null,
     thresholdType?: BudgetThresholdType,
   ) {
-    await notifyAttentionRetentionWork(db, { sourceKind: "budget_alert" });
+    await notifyAttentionRetentionWork(db, { companyId, sourceKind: "budget_alert" });
     const condition = and(eq(budgetIncidents.policyId, policyId), eq(budgetIncidents.status, "open"),
       thresholdType ? eq(budgetIncidents.thresholdType, thresholdType) : undefined);
     const openRows = await db
@@ -618,9 +619,9 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica
       if (policy.isActive && policy.hardStopEnabled && policy.amount > 0 && observed.recoveringRunCount > 0) recoveringPolicy ??= policy;
       // A closed UTC window, inactive policy, or budget raise releases only
       // budget-owned pauses. Manual pauses remain operator decisions.
-      await resolveOpenIncidentsForPolicy(policy.id, "approved", "budget_service", "hard");
+      await resolveOpenIncidentsForPolicy(policy.id, policy.companyId, "approved", "budget_service", "hard");
       if (!policy.isActive || !policy.notifyEnabled || compareCents(observed.totalExact, Math.ceil(policy.amount * policy.warnPercent / 100)) < 0) {
-        await resolveOpenIncidentsForPolicy(policy.id, "approved", "budget_service", "soft");
+        await resolveOpenIncidentsForPolicy(policy.id, policy.companyId, "approved", "budget_service", "soft");
       }
     }
     // Agent status has its own admission gate and error code. Preserve a
@@ -677,7 +678,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica
       }
 
       if (observedBlocks(policy, observed)) {
-        await resolveOpenSoftIncidents(policy.id);
+        await resolveOpenSoftIncidents(policy.id, policy.companyId);
         const hardIncident = await createIncidentIfNeeded(policy, "hard", observed);
         await pauseAndCancelScopeForBudget(policy);
         if (hardIncident?.created) {
@@ -847,11 +848,11 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica
         const observed = await computeObservedSpend(db, row);
         if (!observedBlocks(row, observed)) {
           await resumeScopeFromBudget(row);
-          await resolveOpenIncidentsForPolicy(row.id, "approved", actorUserId ?? "budget_service", "hard");
+          await resolveOpenIncidentsForPolicy(row.id, row.companyId, "approved", actorUserId ?? "budget_service", "hard");
           if (row.notifyEnabled && compareCents(observed.totalExact, Math.ceil(row.amount * row.warnPercent / 100)) >= 0) {
             await createIncidentIfNeeded(row, "soft", observed);
           } else {
-            await resolveOpenIncidentsForPolicy(row.id, null, null, "soft");
+            await resolveOpenIncidentsForPolicy(row.id, row.companyId, null, null, "soft");
           }
         } else {
           const softThreshold = Math.ceil((row.amount * row.warnPercent) / 100);
@@ -859,14 +860,14 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica
             await createIncidentIfNeeded(row, "soft", observed);
           }
           if (observedBlocks(row, observed)) {
-            await resolveOpenSoftIncidents(row.id);
+            await resolveOpenSoftIncidents(row.id, row.companyId);
             await createIncidentIfNeeded(row, "hard", observed);
             await pauseAndCancelScopeForBudget(row);
           }
         }
       } else {
         await resumeScopeFromBudget(row);
-        await resolveOpenIncidentsForPolicy(row.id, "approved", actorUserId ?? "budget_service");
+        await resolveOpenIncidentsForPolicy(row.id, row.companyId, "approved", actorUserId ?? "budget_service");
       }
 
       await recordActivity({

server/src/services/decision-expiry.ts

@@ -23,7 +23,7 @@ export function decisionExpiryService(db: Db, deliverContinuation: (id: string)
         agentId: row.originAgentId, runId: row.originRunId, action: "decision.expired", entityType: "decision",
         entityId: row.id, details: { expiredReason: reason } }, publications);
       return row;
-    });
+    }, { companyId: decision.companyId, sourceId: decision.id });
     if (!updated) return 0;
     for (const publication of publications) publishActivity(publication);
     await deliverContinuation(updated.id);

server/src/services/decision-work-notifications.ts

@@ -2,20 +2,22 @@ import { notifyAttentionRetentionWork } from "./retention-work-notifications.js"
 import type { Db } from "@paperclipai/db";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
 
-export async function notifyDecisionCreated(transaction: object) {
-  await notifyAttentionRetentionWork(transaction, { sourceKind: "decision" });
+type DecisionWorkScope = { companyId: string; sourceId?: string };
+
+export async function notifyDecisionCreated(transaction: object, scope: DecisionWorkScope) {
+  await notifyAttentionRetentionWork(transaction, { ...scope, sourceKind: "decision" });
   return notifyDeliveryWork(transaction, DELIVERY_QUEUES.decisionExpiry);
 }
 
-export async function notifyDecisionChange(transaction: object) {
-  await notifyAttentionRetentionWork(transaction, { sourceKind: "decision" });
+export async function notifyDecisionChange(transaction: object, scope: DecisionWorkScope) {
+  await notifyAttentionRetentionWork(transaction, { ...scope, sourceKind: "decision" });
   await notifyDeliveryWork(transaction, DELIVERY_QUEUES.decisionContinuation);
   await notifyDeliveryWork(transaction, DELIVERY_QUEUES.decisionExpiry);
 }
 
-export function writeDecisionChange<T>(db: Db, write: (transaction: Db) => Promise<T>): Promise<T> {
+export function writeDecisionChange<T>(db: Db, write: (transaction: Db) => Promise<T>, scope: DecisionWorkScope): Promise<T> {
   return db.transaction(async transaction => {
-    await notifyDecisionChange(transaction);
+    await notifyDecisionChange(transaction, scope);
     await notifyDeliveryWork(transaction, DELIVERY_QUEUES.decisionEffectRecovery);
     return write(transaction as unknown as Db);
   });

server/src/services/decisions.ts

@@ -239,7 +239,7 @@ export function decisionService(db: Db, options: DecisionServiceOptions) {
       ...(input.additionalTargetSnapshots ?? {}),
     };
     const id = randomUUID();
-    await notifyDecisionCreated(dbOrTx);
+    await notifyDecisionCreated(dbOrTx, { companyId: input.companyId, sourceId: id });
     await notifyDecisionTargetWork(dbOrTx, `decision:${id}`);
     const [created] = await dbOrTx.insert(decisions).values({ id, companyId: input.companyId, bundleId: input.bundleId ?? null,
       originAgentId: input.agentId, originIssueId: provenance.issueId, originRunId: input.runId, ruleKey: input.ruleKey ?? null,
@@ -581,7 +581,8 @@ export function decisionService(db: Db, options: DecisionServiceOptions) {
     return attempt;
   }
 
-  async function settleInvalidEffectOutcome(id: string) {
+  async function settleInvalidEffectOutcome(decision: typeof decisions.$inferSelect) {
+    const id = decision.id;
     const publications: ActivityPublication[] = [];
     await writeDecisionChange(db, async tx => {
       const [current] = await tx.select().from(decisions).where(eq(decisions.id, id)).for("update");
@@ -601,14 +602,14 @@ export function decisionService(db: Db, options: DecisionServiceOptions) {
         action: "decision.recovery_failed", entityType: "decision", entityId: id,
         details: { reason, executionStatus, chosenOptionId: current.chosenOptionId, decidedByUserId: current.decidedByUserId },
       }, publications);
-    });
+    }, { companyId: decision.companyId, sourceId: id });
     for (const publication of publications) publishActivity(publication);
     return outcome(id);
   }
 
   async function executeDecisionEffects(decision: typeof decisions.$inferSelect, userActor: AuthorizationActor) {
     const option = decision.options.find((item) => item.id === decision.chosenOptionId);
-    if (!option || !decision.decidedByUserId) return settleInvalidEffectOutcome(decision.id);
+    if (!option || !decision.decidedByUserId) return settleInvalidEffectOutcome(decision);
     const run = await db.select({ responsibleUserId: heartbeatRuns.responsibleUserId }).from(heartbeatRuns).where(eq(heartbeatRuns.id, decision.originRunId)).then((rows) => rows[0] ?? null);
     const metadata = decision.metadata as Record<string, unknown>;
     if (metadata.kind === "attention_archive_proposal") {
@@ -618,7 +619,7 @@ export function decisionService(db: Db, options: DecisionServiceOptions) {
         targetSnapshots: decision.targetSnapshots as Record<string, Snapshot>,
       }), decision.signedSpec)) {
         await writeDecisionChange(db, async tx => tx.update(decisions).set({ executionStatus: "failed", updatedAt: new Date(), metadata: { ...metadata, archiveProposalError: "invalid_signature", ...(decision.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } })
-          .where(and(eq(decisions.id, decision.id), eq(decisions.executionStatus, "running"))));
+          .where(and(eq(decisions.id, decision.id), eq(decisions.executionStatus, "running"))), { companyId: decision.companyId, sourceId: decision.id });
         return outcome(decision.id);
       }
       if (option.id === "archive") {
@@ -629,7 +630,7 @@ export function decisionService(db: Db, options: DecisionServiceOptions) {
         const manifestHash = manifest ? hashAttentionArchiveManifest(manifest) : null;
         if (!manifest || manifestHash !== metadata.manifestHash) {
           await writeDecisionChange(db, async tx => tx.update(decisions).set({ executionStatus: "failed", updatedAt: new Date(), metadata: { ...metadata, archiveProposalError: "manifest_mismatch", ...(decision.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } })
-            .where(and(eq(decisions.id, decision.id), eq(decisions.executionStatus, "running"))));
+            .where(and(eq(decisions.id, decision.id), eq(decisions.executionStatus, "running"))), { companyId: decision.companyId, sourceId: decision.id });
           return outcome(decision.id);
         }
         const originActor: AuthorizationActor = {
@@ -651,7 +652,7 @@ export function decisionService(db: Db, options: DecisionServiceOptions) {
         } catch (error) {
           const message = error instanceof Error ? error.message : "archive_proposal_failed";
           await writeDecisionChange(db, async tx => tx.update(decisions).set({ executionStatus: "failed", updatedAt: new Date(), metadata: { ...metadata, archiveProposalError: message, ...(decision.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } })
-            .where(and(eq(decisions.id, decision.id), eq(decisions.executionStatus, "running"))));
+            .where(and(eq(decisions.id, decision.id), eq(decisions.executionStatus, "running"))), { companyId: decision.companyId, sourceId: decision.id });
           return outcome(decision.id);
         }
       }
@@ -659,7 +660,7 @@ export function decisionService(db: Db, options: DecisionServiceOptions) {
         executionStatus: "succeeded",
         updatedAt: new Date(),
         metadata: { ...metadata, ...(decision.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) },
-      }).where(and(eq(decisions.id, decision.id), eq(decisions.executionStatus, "running"))));
+      }).where(and(eq(decisions.id, decision.id), eq(decisions.executionStatus, "running"))), { companyId: decision.companyId, sourceId: decision.id });
       return outcome(decision.id);
     }
     for (let index = 0; index < option.effects.length; index += 1) await executeEffect(decision, option.effects[index]!, index, userActor, decision.decidedByUserId, run?.responsibleUserId ?? null);
@@ -667,7 +668,7 @@ export function decisionService(db: Db, options: DecisionServiceOptions) {
     const successful = rows.filter((row) => row.status === "executed").length;
     const status = rows.every((row) => row.status === "executed") ? "succeeded" : successful ? "partial" : "failed";
     await writeDecisionChange(db, async tx => tx.update(decisions).set({ executionStatus: status, updatedAt: new Date(), metadata: { ...decision.metadata,
-      ...(decision.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } }).where(and(eq(decisions.id, decision.id), eq(decisions.executionStatus, "running"))));
+      ...(decision.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } }).where(and(eq(decisions.id, decision.id), eq(decisions.executionStatus, "running"))), { companyId: decision.companyId, sourceId: decision.id });
     return outcome(decision.id);
   }
 
@@ -706,7 +707,7 @@ export function decisionService(db: Db, options: DecisionServiceOptions) {
         decideIdempotencyKey: input.idempotencyKey ?? null,
         ...(current.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}),
         ...(input.dismissed ? { dismissed: true, dismissReason: input.dismissReason ?? null } : {}) } })
-      .where(and(eq(decisions.id, current.id), eq(decisions.status, "open"))).returning());
+      .where(and(eq(decisions.id, current.id), eq(decisions.status, "open"))).returning(), { companyId: current.companyId, sourceId: current.id });
     if (!claimed) throw conflict("decision_already_resolved", { code: "decision_already_resolved" });
     const run = await db.select({ responsibleUserId: heartbeatRuns.responsibleUserId }).from(heartbeatRuns).where(eq(heartbeatRuns.id, claimed.originRunId)).then((rows) => rows[0] ?? null);
     await logActivity(db, { companyId: claimed.companyId, actorType: "system", actorId: "decision-executor", agentId: claimed.originAgentId, runId: claimed.originRunId,
@@ -724,7 +725,7 @@ export function decisionService(db: Db, options: DecisionServiceOptions) {
       return (await get(id))!;
     }
     const [updated] = await writeDecisionChange(db, async tx => tx.update(decisions).set({ status: "cancelled", updatedAt: new Date(), metadata: { ...current.metadata,
-      ...(current.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } }).where(and(eq(decisions.id, id), eq(decisions.status, "open"))).returning());
+      ...(current.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } }).where(and(eq(decisions.id, id), eq(decisions.status, "open"))).returning(), { companyId: current.companyId, sourceId: id });
     if (!updated) throw conflict("decision_already_resolved", { code: "decision_already_resolved" });
     await logActivity(db, { companyId: updated.companyId, actorType: actor.actorType, actorId: actor.actorId, runId: actor.runId, action: "decision.cancelled", entityType: "decision", entityId: id });
     await continuations.deliver(updated.id);
@@ -738,7 +739,7 @@ export function decisionService(db: Db, options: DecisionServiceOptions) {
     if (empty) return decide({ id, optionId: empty.id, decidedByUserId: userId, userActor, dismissed: true, dismissReason: reason });
     const [updated] = await writeDecisionChange(db, async tx => tx.update(decisions).set({ status: "decided", executionStatus: "succeeded", chosenOptionId: "dismissed", decidedByUserId: userId,
       decidedAt: new Date(), updatedAt: new Date(), metadata: { ...current.metadata, dismissed: true, dismissReason: reason ?? null,
-        ...(current.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } }).where(and(eq(decisions.id, id), eq(decisions.status, "open"))).returning());
+        ...(current.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } }).where(and(eq(decisions.id, id), eq(decisions.status, "open"))).returning(), { companyId: current.companyId, sourceId: id });
     if (!updated) throw conflict("decision_already_resolved", { code: "decision_already_resolved" });
     await logActivity(db, { companyId: updated.companyId, actorType: "system", actorId: "decision-executor", agentId: updated.originAgentId,
       runId: updated.originRunId, responsibleUserIdOverride: userId, action: "decision.dismissed", entityType: "decision", entityId: updated.id,

server/src/services/heartbeat-run-events.ts

@@ -10,6 +10,8 @@ import type { Db } from "@paperclipai/db";
 import { heartbeatRunEvents, heartbeatRuns } from "@paperclipai/db";
 import { nativeSha256 } from "./native-runtime/canonical.js";
 
+export const RETRY_EXHAUSTION_MESSAGE_PREFIX = "Bounded retry exhausted";
+
 export interface AppendHeartbeatRunEventInput {
   companyId: string;
   runId: string;
@@ -75,7 +77,7 @@ export async function appendHeartbeatRunEvent(
   db: Db,
   input: AppendHeartbeatRunEventInput,
 ): Promise<AppendHeartbeatRunEventResult> {
-  const projectsRetention = input.eventType === "lifecycle" && Boolean(input.message?.startsWith("Bounded retry exhausted"));
+  const projectsRetention = input.eventType === "lifecycle" && Boolean(input.message?.startsWith(RETRY_EXHAUSTION_MESSAGE_PREFIX));
   const projectsReassessment = readNativeEventEvidenceOutcome(input.payload?.prpEvent) !== null;
   const projectsProgress = isSafeNativeChatProgressEventType(input.eventType);
   const projectsResponse = isNativeResponseEvidence(input.eventType, input.payload?.prpEvent);
@@ -131,7 +133,7 @@ export async function appendHeartbeatRunEvent(
           eq(heartbeatRunEvents.runId, input.runId),
           eq(heartbeatRunEvents.agentId, input.agentId),
           eq(heartbeatRunEvents.eventType, "lifecycle"),
-          sql`${heartbeatRunEvents.message} like 'Bounded retry exhausted%'`,
+          sql`${heartbeatRunEvents.message} like ${`${RETRY_EXHAUSTION_MESSAGE_PREFIX}%`}`,
           sql`${heartbeatRunEvents.payload} @> ${JSON.stringify(input.retryExhaustion)}::jsonb`,
         ))
         .limit(1)

server/src/services/issues.ts

@@ -7598,7 +7598,7 @@ export function issueService(db: Db) {
     companyId: string,
     blockedByIssueIds: string[],
     actor: { agentId?: string | null; userId?: string | null } = {},
-    dbOrTx: any = db,
+    dbOrTx: any,
   ) {
     const deduped = [...new Set(blockedByIssueIds)];
     if (deduped.some((candidate) => candidate === issueId)) {
@@ -9676,27 +9676,32 @@ export function issueService(db: Db) {
       });
 
       if (blockParentUntilDone) {
-        const existingBlockers = await db
-          .select({ blockerIssueId: issueRelations.issueId })
-          .from(issueRelations)
-          .where(
-            and(
-              eq(issueRelations.companyId, parent.companyId),
-              eq(issueRelations.relatedIssueId, parent.id),
-              eq(issueRelations.type, "blocks"),
-            ),
+        await db.transaction(async tx => {
+          await tx.select({ id: issues.id }).from(issues).where(and(
+            eq(issues.companyId, parent.companyId), eq(issues.id, parent.id))).for("update");
+          const existingBlockers = await tx
+            .select({ blockerIssueId: issueRelations.issueId })
+            .from(issueRelations)
+            .where(
+              and(
+                eq(issueRelations.companyId, parent.companyId),
+                eq(issueRelations.relatedIssueId, parent.id),
+                eq(issueRelations.type, "blocks"),
+              ),
+            );
+          await syncBlockedByIssueIds(
+            parent.id,
+            parent.companyId,
+            [
+              ...new Set([
+                ...existingBlockers.map((row) => row.blockerIssueId),
+                child.id,
+              ]),
+            ],
+            { agentId: actorAgentId ?? null, userId: actorUserId ?? null },
+            tx,
           );
-        await syncBlockedByIssueIds(
-          parent.id,
-          parent.companyId,
-          [
-            ...new Set([
-              ...existingBlockers.map((row) => row.blockerIssueId),
-              child.id,
-            ]),
-          ],
-          { agentId: actorAgentId ?? null, userId: actorUserId ?? null },
-        );
+        });
         [child] = await withIssueRelationSummaries(
           parent.companyId,
           [child],