perf(server): archive attention from source deadlines

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

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

@@ -0,0 +1,48 @@
+# Archive attention sources without repeated feed scans
+
+Attention retention now reconstructs source deadlines at startup and after committed source changes. It no longer rebuilds every active company's full feed on each heartbeat tick. The archive age remains 90 days. Keep and manual archive state remain authoritative, and automatic archive still commits its notification outbox atomically.
+
+This document records source ownership, producer coverage, and the evidence needed to review the replacement. It makes no claim about host suspension or external wake alarms.
+
+## Existing source rows own eligibility and deadlines
+
+`attention-retention.ts` selects at most 100 unfinished source candidates per reconstruction page. It excludes inactive companies and kept or archived retention states. The scoped coordinator retains eligible deadlines, retries concrete failures, and cancels entries that leave attention. It does not schedule another scan after settlement unless a producer changes a relevant source.
+
+| Source | Eligible durable state | Activity owner |
+| --- | --- | --- |
+| Decision | Open decision with a visible execution issue | Decision updated time |
+| Interaction | Pending board-visible interaction; newest ordinary confirmation or independent quality gate | Interaction updated time |
+| Recovery action | Active or escalated action owned by a user or board, with a visible source issue | Recovery updated time |
+| Blocker attention | Board unblock descriptor or non-live terminal blocker selected by the issue owner | Blocked transition or terminal issue updated time |
+| Review | Human participant, user assignee, pending approval, or stalled review selected by the issue owner | Issue updated time |
+| Failed run | Failed or timed-out run with an exhaustion event, eligible agent and issue, and no newer matching run | Finished time, falling back to updated or created time |
+| Budget alert | Open hard incident, or soft incident meeting the existing 85 percent rounded threshold | Incident updated time |
+| Agent error | Agent in error status | Agent updated time |
+
+The feed and worker share confirmation collapse, newer-run probes, issue visibility, review eligibility, and terminal-blocker predicates. Issue graph and agent invokability logic remain with their existing owners. Blocker entries retain both the root used to reconstruct the graph and the terminal subject that owns retention state. PostgreSQL deadlines round submillisecond timestamps upward. The worker locks the source and terminal subject before archival.
+
+`productivity_review` generates no feed items and owes no retention work. Queue retention-day overrides control shelf presentation, which is computed on reads. They do not change the fixed automatic archive age. Approval retention, join-request retention, and archive notification delivery keep their existing workers.
+
+## 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.
+
+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.
+
+The writer contract follows the existing delivery workers: serving processes use the transaction-aware service boundaries, and each serving process registers its workers. Independent raw SQL writers must supply equivalent ingress or restart reconstruction. Restores reconstruct work at startup. The change adds no permanent listener, independent timer, table, or Cloud wake mechanism. Existing sleep-safety gates remain in force.
+
+## 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.
+
+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.
+
+The old aggregate path issued 12 statements per tick in each of three configured-idle samples. The replacement issued zero statements during 300 seconds after settlement. One transaction-status read after an injected rollback is classified as uncertainty reconciliation; the following 300-second drain window must remain quiet.
+
+The focused suite supports `PAPERCLIP_ATTENTION_RETENTION_TEST_DATABASE_URL`, validates its dedicated `_test` database name, applies pending migrations, and prewarms a six-connection pool before advancing fake time. Providers are not involved in retention selection; the existing archive outbox delivery tests verify the separate controlled notification boundary. Fake time proves scheduling and SQL quiet, not database or host suspension.
+
+## Review and operational limits remain explicit
+
+Local embedded PostgreSQL could not start on the test host because system semaphores were exhausted. The accepted runs use isolated PostgreSQL 17 databases. A mixed consumer run also exposed fixture contamination from a lifecycle suite that retains native finalization rows; isolated reruns are required and preserve the failed run in the review evidence. Broad Linux CI remains the required full-suite gate.
+
+Review the source parity fixtures and producer hints together. The intended outcome is unchanged archive state and notifications with no recurring retention SQL when the sources remain unchanged. No merge or deployment is part of this change.

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

@@ -0,0 +1,339 @@
+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,
+  issueRelations, issueThreadInteractions, issues, type Db } from "@paperclipai/db";
+import { attentionRetentionService, registerAttentionRetentionWork } from "../services/attention-retention.js";
+import { attentionService } from "../services/attention.js";
+import { companyService } from "../services/companies.js";
+import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
+import { decisionRetentionService, DEFAULT_DECISION_ARCHIVE_DAYS } from "../services/decision-retention.js";
+import { appendHeartbeatRunEvent } from "../services/heartbeat-run-events.js";
+import { issueThreadInteractionService } from "../services/issue-thread-interactions.js";
+import { createAgentLifecycle } from "../services/agent-lifecycle.js";
+import { issueService } from "../services/issues.js";
+import { notifyAttentionRetentionWork } from "../services/retention-work-notifications.js";
+import { idleWorkSnapshot } from "../services/task-admission.js";
+import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
+
+import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
+
+const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_ATTENTION_RETENTION_TEST_DATABASE_URL");
+const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTestSupport();
+const delay = DEFAULT_DECISION_ARCHIVE_DAYS * 86_400_000;
+(support.supported ? describe : describe.skip)("remaining attention retention", () => {
+  let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
+  let db: ReturnType<typeof createDb>;
+  let companyId: string, agentId: string, draining = false;
+  let coordinator: ReturnType<typeof createDeliveryWorkCoordinator> | undefined;
+  const errors: unknown[] = [];
+  const statements: string[] = [];
+  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) => statements.push(query);
+  }, 30_000);
+  beforeEach(async () => {
+    companyId = randomUUID(); agentId = randomUUID(); draining = false; errors.length = 0;
+    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" });
+    await db.insert(agents).values({ id: agentId, companyId, name: "Origin", role: "engineer", status: "active" });
+    await Promise.all(Array.from({ length: 6 }, () => db.execute(sql`select pg_sleep(0.01)`)));
+    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: true });
+    vi.setSystemTime(new Date("2026-11-01T12:00:00Z"));
+  });
+  afterEach(async () => {
+    await coordinator?.stop(); coordinator = undefined;
+    vi.useRealTimers(); vi.restoreAllMocks();
+    await db.transaction(async tx => {
+      await tx.execute(sql`set local client_min_messages = warning`);
+      await tx.execute(sql`drop trigger if exists attention_retention_fail on decision_retention`);
+      await tx.execute(sql`drop function if exists attention_retention_fail()`);
+      await tx.execute(sql`truncate companies cascade`);
+    });
+    expect(errors).toEqual([]);
+  });
+  afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); });
+  const old = () => new Date(Date.now() - delay - 600_000);
+  async function insertIssue(status = "todo", values: Partial<typeof issues.$inferInsert> = {}, database: Db = db) {
+    const [row] = await database.insert(issues).values({ companyId, title: randomUUID(), status,
+      createdByAgentId: agentId, createdAt: old(), updatedAt: old(), ...values }).returning();
+    return row;
+  }
+  async function change<T>(write: (tx: Db) => PromiseLike<T>, scope = { companyId }) {
+    return db.transaction(async tx => { await notifyAttentionRetentionWork(tx, scope); return write(tx as unknown as Db); });
+  }
+  const state = (kind: string, id: string) => db.select().from(decisionRetention).where(and(
+    eq(decisionRetention.companyId, companyId), eq(decisionRetention.sourceKind, kind), eq(decisionRetention.sourceId, id)))
+    .then(rows => rows[0] ?? null);
+  async function settle() {
+    await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(idleWorkSnapshot().active).toBe(0); }, { timeout: 15_000 });
+  }
+  async function quiet() {
+    await settle(); expect(coordinator!.nextWakeAt()).toBeNull();
+    const select = vi.spyOn(db, "select"), execute = vi.spyOn(db, "execute"), transaction = vi.spyOn(db, "transaction");
+    statements.length = 0;
+    await vi.advanceTimersByTimeAsync(300_000);
+    expect(statements).toEqual([]);
+    expect(select).not.toHaveBeenCalled(); expect(execute).not.toHaveBeenCalled(); expect(transaction).not.toHaveBeenCalled();
+    select.mockRestore(); execute.mockRestore(); transaction.mockRestore();
+  }
+  async function start(service = attentionRetentionService(db)) {
+    coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining, canReconcile: () => true, onError: error => errors.push(error) });
+    await registerAttentionRetentionWork(coordinator, service, () => !draining).ready;
+    return service;
+  }
+  async function archived(kind: string, id: string) {
+    await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect((await state(kind, id))?.archivedAt).toBeInstanceOf(Date); }, { timeout: 15_000 });
+  }
+  async function loadAll(service = attentionRetentionService(db)) {
+    const entries = [];
+    for (;;) { const page = await service.load({ companyId }); entries.push(...page.rows ?? []); if (page.complete) return entries; }
+  }
+  const mutation = (sourceKind: "review", sourceId: string) => ({ companyId, sourceKind, sourceId,
+    authActor: { type: "board" as const, source: "local_implicit" as const, userId: "board", companyIds: [companyId], isInstanceAdmin: true },
+    actor: { actorType: "user" as const, actorId: "board", userId: "board", agentId: null, runId: null, agentApiKeyId: null, responsibleUserId: "board" } });
+
+  it("records the aggregate baseline SQL across old ticks and eliminates it after settlement", async () => {
+    await insertIssue("done");
+    const ticks: number[] = [];
+    for (let tick = 0; tick < 3; tick++) {
+      statements.length = 0;
+      const feed = await attentionService(db).list(companyId, { includeDismissed: true, excludeSources: ["approval", "join_request"], all: true, allowUnscopedAll: true });
+      await decisionRetentionService(db).autoArchive({ companyId, items: feed.items });
+      ticks.push(statements.length);
+      await vi.advanceTimersByTimeAsync(30_000);
+    }
+    expect(ticks.every(count => count > 0)).toBe(true);
+    process.stdout.write(JSON.stringify({ baselineStatementsPer30sTick: ticks, treatmentIdle300s: 0 }) + "\n");
+    await start(); await quiet();
+  });
+
+  it("has no recurring SQL when empty or when terminal source history remains", async () => {
+    await insertIssue("done"); await insertIssue("cancelled");
+    await start(); await quiet();
+  });
+
+  it("wakes from the ordinary task producer, waits until due, archives with an outbox, and settles", async () => {
+    await start(); await quiet();
+    const row = await issueService(db).create(companyId, { title: "Review a delivered result", status: "in_review", assigneeUserId: "board", createdByAgentId: agentId });
+    await settle();
+    const [saved] = await db.select({ at: sql<number>`ceil(extract(epoch from ${issues.updatedAt}) * 1000)`.mapWith(Number) }).from(issues).where(eq(issues.id, row.id));
+    const due = saved.at + delay;
+    expect(coordinator!.nextWakeAt()).toBe(due); expect(await state("review", row.id)).toBeNull();
+    const select = vi.spyOn(db, "select"), execute = vi.spyOn(db, "execute");
+    await vi.advanceTimersByTimeAsync(due - Date.now() - 1000);
+    expect(select).not.toHaveBeenCalled(); expect(execute).not.toHaveBeenCalled(); select.mockRestore(); execute.mockRestore();
+    await vi.advanceTimersByTimeAsync(2000); await archived("review", row.id); await quiet();
+    const outbox = await db.select().from(decisionArchiveNotificationOutbox);
+    expect(outbox).toHaveLength(1); expect(outbox[0]).toMatchObject({ companyId, sourceKind: "review", sourceId: row.id, originAgentId: agentId, originIssueId: row.id });
+    expect((await db.select().from(issues).where(eq(issues.id, row.id)))[0].status).toBe("in_review");
+    await coordinator!.stop(); coordinator = undefined; await start(); await quiet();
+  });
+
+  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" });
+    const blocker = await insertIssue("blocked", { unblockDescriptor: { owner: "board", action: "Choose a dependency" }, blockedTransitionAt: old() });
+    await insertIssue("in_review", { hiddenAt: old(), assigneeUserId: "board" });
+    const hidden = await insertIssue("todo", { hiddenAt: old() });
+    const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId, status: "failed", contextSnapshot: { issueId: origin.id }, createdAt: old(), updatedAt: old(), finishedAt: old() }).returning();
+    await db.insert(heartbeatRunEvents).values({ companyId, agentId, runId: run.id, seq: 1, eventType: "lifecycle", message: "Bounded retry exhausted: fixture" });
+    await db.insert(decisions).values({ companyId, originAgentId: agentId, originIssueId: origin.id, originRunId: run.id,
+      title: "Decision", body: "Choose", options: [], signedSpec: "fixture", targetSnapshots: {}, expiresAt: new Date(Date.now() + 86_400_000), updatedAt: old() });
+    await db.insert(issueThreadInteractions).values([
+      { companyId, issueId: origin.id, kind: "ask_user_questions", payload: { questions: [] }, updatedAt: old() },
+      { companyId, issueId: origin.id, kind: "ask_user_questions", payload: { questions: [] }, addresseeAgentId: agentId, updatedAt: old() },
+      { companyId, issueId: origin.id, kind: "ask_user_questions", payload: { questions: [] }, addresseeUserId: "board", updatedAt: old() },
+      { companyId, issueId: hidden.id, kind: "ask_user_questions", payload: { questions: [] }, updatedAt: old() },
+    ]);
+    await db.insert(issueRecoveryActions).values({ companyId, sourceIssueId: origin.id, kind: "execution", ownerType: "board", cause: "fixture", fingerprint: "fixture", nextAction: "Choose", updatedAt: old() });
+    const [policy] = await db.insert(budgetPolicies).values({ companyId, scopeType: "company", scopeId: companyId, windowKind: "calendar_month_utc", amount: 100 }).returning();
+    await db.insert(budgetIncidents).values({ companyId, policyId: policy.id, scopeType: "company", scopeId: companyId,
+      metric: "billed_cents", windowKind: "calendar_month_utc", windowStart: old(), windowEnd: new Date(), thresholdType: "hard", amountLimit: 100, amountObserved: 100, updatedAt: old() });
+    await db.insert(agents).values({ companyId, name: "Error", role: "engineer", status: "error", updatedAt: old() });
+    const feed = await attentionService(db).list(companyId, { all: true, allowUnscopedAll: true, includeDismissed: true, excludeSources: ["approval", "join_request"] });
+    const expected = [...new Set(feed.items.map(item => `${item.sourceKind}:${item.subject.id}`))].sort();
+    const actual = [...new Set((await loadAll()).map(item => `${item.sourceKind}:${item.sourceId}`))].sort();
+    expect(new Set(feed.items.map(item => item.sourceKind)).size).toBe(8); expect(actual).toEqual(expected);
+    expect(actual).toContain(`review:${review.id}`); expect(actual).toContain(`blocker_attention:${blocker.id}`);
+    await start(); for (const item of feed.items) await archived(item.sourceKind, item.subject.id); await quiet();
+  });
+
+  it("keeps unseen sources, cancels their deadline, and reactivates when unkept or revived", async () => {
+    const row = await insertIssue("in_review", { assigneeUserId: "board" });
+    await decisionRetentionService(db).setKeep({ ...mutation("review", row.id), keep: true });
+    await start(); await quiet(); expect((await state("review", row.id))?.archivedAt).toBeNull();
+    await decisionRetentionService(db).setKeep({ ...mutation("review", row.id), keep: false });
+    await archived("review", row.id); await quiet();
+    await decisionRetentionService(db).revive(mutation("review", row.id)); await archived("review", row.id); await quiet();
+  });
+
+  it("moves deadlines on source activity and cancels when the source leaves attention", async () => {
+    const row = await insertIssue("in_review", { assigneeUserId: "board", updatedAt: new Date() });
+    await start(); await settle(); const first = coordinator!.nextWakeAt()!;
+    await change(tx => tx.update(issues).set({ updatedAt: new Date(Date.now() - delay + 600_000) }).where(eq(issues.id, row.id)));
+    await settle(); expect(coordinator!.nextWakeAt()).toBeLessThan(first);
+    await issueService(db).update(row.id, { status: "todo" }); await quiet(); expect(await state("review", row.id)).toBeNull();
+  });
+
+  it("preserves outer transaction ownership and first writes during drain", async () => {
+    await start(); await quiet(); draining = true;
+    let id: string | undefined;
+    await db.transaction(async outer => {
+      await outer.transaction(async nested => {
+        await notifyAttentionRetentionWork(nested, { companyId });
+        id = (await insertIssue("in_review", { assigneeUserId: "board" }, nested as unknown as Db)).id;
+      });
+      await vi.advanceTimersByTimeAsync(60_000); expect(await state("review", id!)).toBeNull();
+    });
+    await vi.advanceTimersByTimeAsync(120_000); expect(await state("review", id!)).toBeNull();
+    draining = false; await vi.advanceTimersByTimeAsync(30_000); await archived("review", id!); await quiet();
+    await expect(db.transaction(async outer => {
+      await outer.transaction(async nested => { await notifyAttentionRetentionWork(nested, { companyId }); await insertIssue("in_review", { assigneeUserId: "board" }, nested as unknown as Db); });
+      throw new Error("rollback");
+    })).rejects.toThrow("rollback");
+    await quiet(); expect(await db.select().from(decisionRetention)).toHaveLength(1);
+  });
+
+  it("isolates companies, catches up at activation, and bounds retained-history pages", async () => {
+    const other = randomUUID(); await db.insert(companies).values({ id: other, name: "Paused", issuePrefix: "RP", status: "paused" });
+    const row = await insertIssue("in_review", { companyId: other, assigneeUserId: "board" });
+    await db.insert(issues).values(Array.from({ length: 1000 }, () => ({ companyId, title: randomUUID(), status: "done", updatedAt: old() })));
+    const service = attentionRetentionService(db); const page = await service.load({ companyId });
+    expect(page).toEqual({ rows: [], complete: true });
+    await start(service); await quiet(); expect(await db.select().from(decisionRetention)).toHaveLength(0);
+    await companyService(db).update(other, { status: "active" });
+    await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(await db.select().from(decisionRetention)).toHaveLength(1); });
+    expect((await db.select().from(decisionRetention))[0]).toMatchObject({ companyId: other, sourceId: row.id }); await quiet();
+  });
+  it("preserves terminal blocker source identity and its own activity deadline", async () => {
+    const root = await insertIssue("blocked", { unblockDescriptor: { owner: "board", action: "Pick an owner" }, blockedTransitionAt: old() });
+    const terminal = await insertIssue("backlog", { assigneeAgentId: agentId, updatedAt: new Date(old().getTime() + 60_000) });
+    await db.insert(issueRelations).values({ companyId, issueId: terminal.id, relatedIssueId: root.id, type: "blocks" });
+    const feed = await attentionService(db).list(companyId, { all: true, allowUnscopedAll: true });
+    const entries = await loadAll();
+    const expected = feed.items.filter(item => item.sourceKind === "blocker_attention")
+      .map(item => `${item.subject.id}:${new Date(item.activityAt).toISOString()}`).sort();
+    expect(entries.filter(item => item.sourceKind === "blocker_attention")
+      .map(item => `${item.sourceId}:${new Date(item.activityAt).toISOString()}`).sort()).toEqual(expected);
+    expect(expected).toContain(`${terminal.id}:${terminal.updatedAt.toISOString()}`);
+    await start(); await archived("blocker_attention", terminal.id); await archived("blocker_attention", root.id); await quiet();
+    expect((await state("blocker_attention", terminal.id))?.sourceActivityAt).toEqual(terminal.updatedAt);
+    expect((await state("blocker_attention", root.id))?.sourceActivityAt).toEqual(root.blockedTransitionAt);
+  });
+
+  it("matches newest confirmations, independent gates, org invokability, and conversation visibility", async () => {
+    const origin = await insertIssue();
+    const conversation = await insertIssue("in_review", { conversationAgentId: agentId, conversationUserId: "board", conversationState: "waiting", assigneeAgentId: agentId });
+    const hidden = await insertIssue("todo", { hiddenAt: old() });
+    const [paused, manager, unhealthy] = await db.insert(agents).values([
+      { companyId, name: "Paused", role: "engineer", status: "paused" },
+      { companyId, name: "Terminated", role: "manager", status: "terminated" },
+      { companyId, name: "Unhealthy", role: "engineer", status: "active" },
+    ]).returning();
+    await db.update(agents).set({ reportsTo: manager.id }).where(eq(agents.id, unhealthy.id));
+    const common = { companyId, issueId: origin.id, kind: "request_confirmation", payload: { prompt: "Approve?" }, createdAt: old(), updatedAt: old() };
+    const [older, newest, independent] = await db.insert(issueThreadInteractions).values([
+      common, { ...common, createdAt: new Date(old().getTime() + 1000) },
+      { ...common, payload: { prompt: "Independent", qualityReview: { artifactKey: "copy" } } },
+      { ...common, createdAt: new Date(), addresseeUserId: "board" },
+      { ...common, addresseeAgentId: agentId },
+      { ...common, issueId: hidden.id },
+      { ...common, issueId: conversation.id, kind: "ask_user_questions", payload: { questions: [] } },
+      { ...common, issueId: conversation.id, addresseeAgentId: paused.id },
+      { ...common, issueId: conversation.id, addresseeAgentId: unhealthy.id, kind: "ask_user_questions", payload: { questions: [] } },
+    ] as typeof issueThreadInteractions.$inferInsert[]).returning();
+    const feed = await attentionService(db).list(companyId, { all: true, allowUnscopedAll: true });
+    const expected = [...new Set(feed.items.map(item => `${item.sourceKind}:${item.subject.id}`))].sort();
+    expect([...new Set((await loadAll()).map(item => `${item.sourceKind}:${item.sourceId}`))].sort()).toEqual(expected);
+    expect(expected).not.toContain(`issue_thread_interaction:${older.id}`);
+    expect(expected).toContain(`issue_thread_interaction:${newest.id}`); expect(expected).toContain(`issue_thread_interaction:${independent.id}`);
+    await start(); await archived("issue_thread_interaction", newest.id); await archived("issue_thread_interaction", independent.id); await quiet();
+    await change(tx => tx.update(issueThreadInteractions).set({ status: "cancelled" }).where(eq(issueThreadInteractions.id, newest.id)));
+    await archived("issue_thread_interaction", older.id); await quiet();
+  });
+
+  it("excludes maintained reviews until a pending approval creates human attention", async () => {
+    const row = await insertIssue("in_review", { assigneeAgentId: agentId });
+    const [interaction] = await db.insert(issueThreadInteractions).values({ companyId, issueId: row.id,
+      kind: "request_confirmation", addresseeAgentId: agentId, payload: { prompt: "Agent verdict" } }).returning();
+    expect((await loadAll()).some(entry => entry.sourceKind === "review")).toBe(false);
+    const [approval] = await db.insert(approvals).values({ companyId, type: "hire_agent", payload: {}, status: "pending" }).returning();
+    await db.insert(issueApprovals).values({ companyId, issueId: row.id, approvalId: approval.id });
+    const feed = await attentionService(db).list(companyId, { all: true, allowUnscopedAll: true });
+    expect(feed.items.some(item => item.sourceKind === "review" && item.subject.id === row.id)).toBe(true);
+    await start(); await archived("review", row.id); await quiet();
+    expect(await state("issue_thread_interaction", interaction.id)).toBeNull();
+  });
+
+  it("reactivates an exhausted failed run through the ordinary durable event producer", async () => {
+    const origin = await insertIssue();
+    const [run, newer] = await db.insert(heartbeatRuns).values([
+      { companyId, agentId, status: "failed", contextSnapshot: { issueId: origin.id }, createdAt: old(), finishedAt: old() },
+      { companyId, agentId, status: "succeeded", contextSnapshot: { issueId: origin.id }, createdAt: new Date(old().getTime() + 1000), finishedAt: old() },
+    ]).returning();
+    await start(); await quiet();
+    await appendHeartbeatRunEvent(db, { companyId, agentId, runId: run.id, eventType: "lifecycle", message: "Bounded retry exhausted after 3 attempts" });
+    await quiet(); expect(await state("failed_run", run.id)).toBeNull();
+    await change(tx => tx.delete(heartbeatRuns).where(eq(heartbeatRuns.id, newer.id)));
+    await archived("failed_run", run.id); await quiet();
+  });
+
+  it("cancels an agent error deadline through ordinary clearError", async () => {
+    await db.update(agents).set({ status: "error", updatedAt: new Date() }).where(eq(agents.id, agentId));
+    await start(); await settle(); expect(coordinator!.nextWakeAt()).not.toBeNull();
+    await createAgentLifecycle(db).clearError(agentId); await quiet();
+    expect(await state("agent_error_alert", agentId)).toBeNull();
+  });
+
+  it("drains a paged backlog with source-scoped self notifications and leaves completed history quiet", async () => {
+    await db.insert(issues).values(Array.from({ length: 225 }, () => ({ companyId, title: randomUUID(), status: "in_review", assigneeUserId: "board", updatedAt: old() })));
+    const service = attentionRetentionService(db);
+    const load = vi.spyOn(service, "load");
+    await start(service);
+    await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(await db.select().from(decisionRetention)).toHaveLength(225); }, { timeout: 30_000 });
+    await quiet();
+    const scopes = load.mock.calls.map(([scope]) => scope);
+    expect(scopes.filter(scope => !scope)).toHaveLength(3);
+    expect(scopes.filter(scope => scope && !scope.sourceId)).toHaveLength(0);
+    expect(scopes.filter(scope => scope?.sourceId).length).toBeLessThanOrEqual(225);
+    expect(new Set((await db.select().from(decisionRetention)).map(row => row.sourceId)).size).toBe(225);
+  });
+
+  it("retries a failed atomic archive across drain and preserves notifications during a scan", async () => {
+    const row = await insertIssue("in_review", { assigneeUserId: "board" });
+    await db.execute(sql`create function attention_retention_fail() returns trigger language plpgsql as $$ begin raise exception 'retention injected'; end $$`);
+    await db.execute(sql`create trigger attention_retention_fail before insert on decision_retention for each row execute function attention_retention_fail()`);
+    await start(); expect(errors).toHaveLength(1); errors.length = 0;
+    expect(await state("review", row.id)).toBeNull(); expect(await db.select().from(decisionArchiveNotificationOutbox)).toHaveLength(0);
+    draining = true; await db.execute(sql`drop trigger attention_retention_fail on decision_retention`);
+    statements.length = 0; await vi.advanceTimersByTimeAsync(120_000);
+    expect(statements.length).toBeLessThanOrEqual(1);
+    expect(statements.every(query => query.includes("pg_xact_status"))).toBe(true);
+    statements.length = 0; await vi.advanceTimersByTimeAsync(300_000); expect(statements).toEqual([]);
+    draining = false; await vi.advanceTimersByTimeAsync(30_000); await archived("review", row.id); await quiet();
+    await coordinator!.stop(); coordinator = undefined;
+    let release!: () => void; let scanned!: () => void;
+    const gate = new Promise<void>(resolve => { release = resolve; });
+    const entered = new Promise<void>(resolve => { scanned = resolve; });
+    const service = attentionRetentionService(db), original = service.load;
+    vi.spyOn(service, "load").mockImplementationOnce(async (...args) => { const result = await original(...args); scanned(); await gate; return result; });
+    const starting = start(service); await entered;
+    const next = await change(async tx => insertIssue("in_review", { assigneeUserId: "board" }, tx));
+    release(); await starting; await archived("review", next.id); await quiet();
+  });
+
+  it("registers and cancels a deadline through ordinary interaction creation and withdrawal", async () => {
+    const origin = await insertIssue(); await start(); await quiet();
+    const service = issueThreadInteractionService(db);
+    const interaction = await service.create(origin, { kind: "request_confirmation", payload: { version: 1, prompt: "Approve the result?" } }, { userId: "board" });
+    await settle(); expect(coordinator!.nextWakeAt()).not.toBeNull(); expect(await state("issue_thread_interaction", interaction.id)).toBeNull();
+    await service.withdrawInteraction(origin, interaction.id, { reason: "No longer needed" }, { userId: "board" }); await quiet();
+  });
+
+});

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

@@ -62,7 +62,7 @@ describeEmbeddedPostgres("attention service", () => {
 
   beforeAll(async () => {
     if (!externalUrl) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-attention-service-");
-    db = createDb(externalUrl ?? tempDb!.connectionString);
+    db = createDb(externalUrl ?? tempDb!.connectionString, { maxConnections: 6 });
   }, 30_000);
 
   afterEach(async () => {

server/src/index.ts

@@ -1,3 +1,4 @@
+import { attentionRetentionService, registerAttentionRetentionWork } from "./services/attention-retention.js";
 import { registerProviderTraceRetentionWork } from "./services/provider-trace-retention-work.js";
 import { registerRunEvidenceMigrationWork } from "./services/run-evidence-migration-work.js";
 import { registerSkillVersionRetentionWork } from "./services/skill-version-retention-work.js";
@@ -127,7 +128,6 @@ import { cloudActorHeaderSourceFromHeaders, resolveCloudTenantActor } from "./mi
 import {
   feedbackService,
   applyManagedEnvironments,
-  attentionService,
   backfillPrincipalAccessCompatibility,
   backfillLegacyToolOAuthTokens,
   bootstrapExecutionPolicyFromEnv,
@@ -1926,23 +1926,6 @@ async function startServerWithDatabaseTeardown(
     if (config.heartbeatSchedulerEnabled) await heartbeat.registerStoppedInstructionCollectionWork(deliveryWork,
       () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready;
 
-    const runRetentionSweep = async () => {
-      const activeCompanies = await db.select({ id: companies.id }).from(companies).where(eq(companies.status, "active"));
-      let archived = 0;
-      for (const company of activeCompanies) {
-        // Cursor pagination rebuilds the whole feed for every page; one
-        // unscoped all-items build keeps this sweep at a single feed build
-        // per company per tick.
-        const page = await attentionService(db as any).list(company.id, {
-          includeDismissed: true,
-          excludeSources: ["approval", "join_request"],
-          all: true,
-          allowUnscopedAll: true,
-        });
-        archived += await retentionExecutor.autoArchive({ companyId: company.id, items: page.items });
-      }
-      return { archived };
-    };
     // Startup housekeeping: provider and connection sweeps that every timer tick
     // repeats and that neither serving requests nor run recovery depends on.
     // Readiness does not wait for it. It is tracked as scheduler work, so idle
@@ -1985,7 +1968,8 @@ async function startServerWithDatabaseTeardown(
         () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready);
       await step("join_request_retention", () => registerJoinRequestRetentionWork(deliveryWork, joinRequestRetentionService(db as any),
         () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready);
-      await step("retention_sweep", runRetentionSweep);
+      await step("attention_retention", () => registerAttentionRetentionWork(deliveryWork, attentionRetentionService(db as any),
+        () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready);
     };
 
     const createHeartbeatSchedulerTick = (scheduled: ReturnType<typeof registerScheduledWork>) => () => {
@@ -1995,9 +1979,6 @@ async function startServerWithDatabaseTeardown(
       trackHeartbeatSchedulerWork((async () => {
         if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return;
         scheduleRoutineStatusReconcile();
-        trackHeartbeatSchedulerWork(runRetentionSweep().catch((err: unknown) => {
-          logger.error({ err }, "decision retention sweep failed");
-        }));
         const sweptRuntimeStatuses = heartbeat.sweepExpiredRuntimeStatuses();
         if (sweptRuntimeStatuses > 0) {
           logger.info(

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

@@ -40,7 +40,7 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
   let companyId: string;
   const workers: ReturnType<typeof configureAgentLifecycle>[] = [];
   beforeAll(async () => {
-    if (external) { await applyPendingMigrations(external); db = createDb(external); }
+    if (external) { await applyPendingMigrations(external); db = createDb(external, { maxConnections: 6 }); }
     else { database = await startEmbeddedPostgresTestDatabase("paperclip-agent-lifecycle-"); db = createDb(database.connectionString); }
   }, 90_000);
   afterAll(async () => {

server/src/modules/agent-lifecycle/adapters/records.ts

@@ -80,18 +80,19 @@ export function agentRecords(db: Db, effects: LifecycleEffects) {
         throw conflict("Only agents in error status can have their error cleared");
       }
 
-      const updated = await db
-        .update(agents)
-        .set({
+      const updated = await effects.transaction(db, existing.companyId, async tx => {
+        await effects.beforeStatusChange(tx, existing.companyId, id, "error", "idle");
+        return tx.update(agents).set({
           status: "idle",
           pauseReason: null,
           pausedAt: null,
           errorReason: null,
           updatedAt: new Date(),
         })
-        .where(and(eq(agents.id, id), eq(agents.status, "error"), eq(agents.lifecycleState, "ready")))
-        .returning()
-        .then((rows) => rows[0] ?? null);
+          .where(and(eq(agents.id, id), eq(agents.status, "error"), eq(agents.lifecycleState, "ready")))
+          .returning()
+          .then((rows) => rows[0] ?? null);
+      });
 
       if (!updated) {
         throw conflict("Only agents in error status can have their error cleared");

server/src/services/accounting-integrity.ts

@@ -1,3 +1,4 @@
+import { notifyAttentionRetentionWork } from "./retention-work-notifications.js";
 import { and, eq, sql } from "drizzle-orm";
 import { agents, agentRuntimeState, companies, heartbeatRuns, type Db } from "@paperclipai/db";
 import { normalizeCents, usdToCents, type AccountingFinding, type AccountingHealth, type AccountingInspection } from "@paperclipai/shared";
@@ -75,6 +76,8 @@ export function accountingIntegrityService(db: Db, hooks: BudgetServiceHooks = {
       if (!reason.trim()) throw conflict("A repair reason is required");
       const before = await inspect(tx, companyId);
       if (before.fingerprint !== fingerprint) throw conflict("Accounting changed since inspection; inspect again before repairing");
+      if (before.findings.some(f => f.repairable && f.kind === "agent_projection"))
+        await notifyAttentionRetentionWork(tx, { companyId, sourceKind: "agent_error_alert" });
       for (const finding of before.findings.filter(f => f.repairable)) {
         if (finding.kind === "company_projection") await tx.update(companies).set({ spentMonthlyCents: sql`${finding.expected.cents}::numeric`, updatedAt: new Date() }).where(eq(companies.id, companyId));
         if (finding.kind === "agent_projection") await tx.update(agents).set({ spentMonthlyCents: sql`${finding.expected.cents}::numeric`, updatedAt: new Date() }).where(and(eq(agents.id, finding.entityId), eq(agents.companyId, companyId)));

server/src/services/agent-configuration.ts

@@ -1,3 +1,4 @@
+import { notifyAttentionRetentionWork } from "./retention-work-notifications.js";
 import { notifySkillVersionPinChange } from "./skill-version-retention-notifications.js";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
 import { notifyExternalConversationSettlement } from "./external-conversation-settlement-notifications.js";
@@ -16,6 +17,7 @@ import { clearPrimaryAgent } from "./primary-agent.js";
 import { budgetServiceInTransaction } from "./budgets.js";
 
 export async function prepareAgentConfiguration(db: Db, existing: AgentConfigurationRecord, data: AgentConfigurationPatch, options?: UpdateAgentOptions): Promise<AgentConfigurationPatch> {
+  await notifyAttentionRetentionWork(db, { companyId: existing.companyId });
   const id = existing.id;
   const { ensureManager, assertNoCycle, assertCompanyShortnameAvailable, assertBuiltInAgentMetadataMutationAllowed } = agentRecordQueries(db);
   if (existing.status === "pending_approval" && !options?.allowPendingApprovalConfigUpdate) {

server/src/services/agent-deletion.ts

@@ -1,3 +1,4 @@
+import { notifyAttentionRetentionWork } from "./retention-work-notifications.js";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
 import { notifyExternalConversationSettlement } from "./external-conversation-settlement-notifications.js";
 import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js";
@@ -34,6 +35,7 @@ export async function deleteAgentDependencies(tx: Db, companyId: string, id: str
   if (classifierHold) throw conflict("Wait for active prompt injection checks or resolve their unknown charges in Costs before deleting this agent", {
     code: "agent_prompt_injection_accounting_pending",
   });
+  await notifyAttentionRetentionWork(tx, { companyId });
   await notifyExternalConversationSettlement(tx, { companyId });
   await issueThreadInteractionService(tx).cancelPendingForDeletedAddressee(companyId, id);
   await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);

server/src/services/agent-hiring.ts

@@ -1,3 +1,4 @@
+import { notifyAttentionRetentionWork } from "./retention-work-notifications.js";
 import { notifySkillVersionRetention, notifySkillVersionPinChange } from "./skill-version-retention-notifications.js";
 import { agentAppearanceSchema, randomAgentAppearance } from "@paperclipai/shared";
 import { normalizePaperclipRunnerAdapterConfig } from "@paperclipai/adapter-utils/server-utils";
@@ -16,6 +17,7 @@ import { initializePrimaryAgent } from "./primary-agent.js";
 import { resolveAgentRunnerTaskDriverForCompany, resolveNewAgentRunnerForCompany } from "./agent-runner-selection.js";
 
 export async function prepareAgentHire(db: Db, companyId: string, input: CreateAgentData, options?: CreateAgentOptions): Promise<CreateAgentData> {
+  await notifyAttentionRetentionWork(db, { companyId });
   const { runner, ...data } = input;
   const selection = options?.runnerResolved
     ? { adapterType: data.adapterType ?? "process", adapterConfig: isPlainRecord(data.adapterConfig) ? data.adapterConfig : {} }

server/src/services/agent-lifecycle.ts

@@ -1,3 +1,4 @@
+import { notifyAttentionRetentionWork } from "./retention-work-notifications.js";
 import { notifyNativeReviewResponse } from "./native-review-response-notifications.js";
 import { isAgentStatusInvokable } from "@paperclipai/shared";
 import { notifyNativeResponsePresentation } from "./native-response-presentation-notifications.js";
@@ -38,6 +39,7 @@ export function createAgentLifecycleEffects(_hooks: BudgetServiceHooks = {}): li
     },
     notifyWork: tx => notifyDeliveryWork(tx, DELIVERY_QUEUES.agentLifecycle),
     async beforeStatusChange(tx, companyId, agentId, before, after) {
+      await notifyAttentionRetentionWork(tx, { companyId });
       if (before !== after) await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
       if (!isAgentStatusInvokable(before) && isAgentStatusInvokable(after)) {
         await notifyNativeResponsePresentation(tx, { companyId, agentId });

server/src/services/agent-project-access.ts

@@ -1,3 +1,4 @@
+import { notifyAttentionRetentionWork } from "./retention-work-notifications.js";
 import { and, eq, inArray } from "drizzle-orm";
 import { agents, projects, type Db } from "@paperclipai/db";
 import { trustAuthorizationPolicySchema, type AgentProjectAccess } from "@paperclipai/shared";
@@ -55,6 +56,7 @@ export async function addAgentProjectAccess(db: Db, actor: AuthorizationActor, c
     const added = requested.filter(id => !previous.includes(id));
     const next = [...previous, ...added];
     if (added.length) {
+      await notifyAttentionRetentionWork(tx, { companyId, sourceKind: "agent_error_alert" });
       await tx.update(agents).set({ permissions: { ...agent.permissions, authorizationPolicy: {
         ...policy, trustBoundary: { ...boundary, mode: "low_trust_review", companyId, projectIds: next },
       } }, updatedAt: new Date() }).where(and(eq(agents.id, agentId), eq(agents.companyId, companyId)));

server/src/services/agents.ts

@@ -1,3 +1,4 @@
+import { notifyAttentionRetentionWork } from "./retention-work-notifications.js";
 import { type BudgetServiceHooks } from "./budgets.js";
 
 import { and, desc, eq, inArray } from "drizzle-orm";
@@ -52,7 +53,9 @@ export function agentService(db: Db, budgetHooks: BudgetServiceHooks = {}) {
         });
       }
 
-      const updated = await db
+      const updated = await db.transaction(async tx => {
+        await notifyAttentionRetentionWork(tx, { companyId: existing.companyId });
+        return tx
         .update(agents)
         .set({
           permissions: normalizeAgentPermissions({ ...existing.permissions, ...permissions }),
@@ -61,6 +64,7 @@ export function agentService(db: Db, budgetHooks: BudgetServiceHooks = {}) {
         .where(eq(agents.id, id))
         .returning()
         .then((rows) => rows[0] ?? null);
+      });
 
       return updated ? getById(updated.id) : null;
     },

server/src/services/attention-retention.ts

@@ -0,0 +1,204 @@
+import { and, eq, 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 { 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 { issueService } from "./issues.js";
+import { parseIssueExecutionState } from "./issue-execution-policy.js";
+import { isProspectiveBlockedTransition } from "./routable-blocked.js";
+import { registerScopedDeadlineWork } from "./scoped-deadline-work.js";
+
+const archiveDelay = DEFAULT_DECISION_ARCHIVE_DAYS * 86_400_000;
+const kinds = ["decision", "issue_thread_interaction", "recovery_action", "blocker_attention", "review",
+  "failed_run", "budget_alert", "agent_error_alert"] as const;
+type Kind = typeof kinds[number];
+type Scope = { companyId?: string; sourceKind?: AttentionSourceKind; sourceId?: string };
+type Candidate = { sourceKind: Kind; sourceId: string; companyId: string; activityAt: string; deadlineAt: number };
+type Entry = Candidate & { id: string; dueAt: number; rootId: string; variant?: "owner" | "terminal" };
+type Cursor = { sourceKind: string; sourceId: string };
+
+const sourceTables: Record<Kind, SQL> = {
+  decision: sql`decisions`, issue_thread_interaction: sql`issue_thread_interactions`,
+  recovery_action: sql`issue_recovery_actions`, blocker_attention: sql`issues`, review: sql`issues`,
+  failed_run: sql`heartbeat_runs`, budget_alert: sql`budget_incidents`, agent_error_alert: sql`agents`,
+};
+
+const sources: Record<Kind, SQL> = {
+  decision: sql`select 'decision' as source_kind, d.id::text as source_id, d.company_id, d.updated_at as activity_at
+    from decisions d where d.status = 'open'`,
+  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')`,
+  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
+    from issues i where i.status = 'in_review' and i.hidden_at is null and i.harness_kind is null`,
+  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%')`,
+  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'`,
+};
+
+export function attentionRetentionService(db: Db) {
+  const cursors = new Map<string, Cursor>();
+  async function candidates(database: Db, scope: Scope = {}, cursor?: Cursor, sourceId?: string): Promise<Candidate[]> {
+    const selected = kinds.filter(kind => !scope.sourceKind || scope.sourceKind === kind);
+    if (!selected.length) return [];
+    const rows = await database.execute(sql`with source_candidates(source_kind, source_id, company_id, activity_at) as (${sql.join(selected.map(kind => sources[kind]), sql` union all `)})
+      select s.source_kind as "sourceKind", s.source_id as "sourceId", s.company_id as "companyId",
+        s.activity_at::text as "activityAt", ceil(extract(epoch from s.activity_at) * 1000 + ${archiveDelay}) as "deadlineAt"
+      from source_candidates s inner join companies c on c.id = s.company_id
+      left join decision_retention r on r.company_id = s.company_id and r.source_kind = s.source_kind and r.source_id = s.source_id
+      where c.status = 'active'
+        and (${scope.companyId ?? null}::uuid is null or s.company_id = ${scope.companyId ?? null}::uuid)
+        and (${sourceId ?? scope.sourceId ?? null}::text is null or s.source_id = ${sourceId ?? scope.sourceId ?? null})
+        and (s.source_kind = 'blocker_attention' or ((r.id is null or r.keep = false) and r.archived_at is null))
+        ${cursor ? sql`and (s.source_kind, s.source_id) > (${cursor.sourceKind}, ${cursor.sourceId})` : sql``}
+      order by s.source_kind, s.source_id limit 100`) as unknown as Candidate[];
+    return rows.map(row => ({ ...row, deadlineAt: Number(row.deadlineAt) }));
+  }
+
+  async function visibleIssue(database: Db, companyId: string, issueId: string, conversations = false) {
+    const [row] = await database.select().from(issues).where(and(eq(issues.companyId, companyId), eq(issues.id, issueId),
+      attentionIssueCondition(conversations)));
+    return row;
+  }
+
+  async function eligible(database: Db, candidate: Candidate, variant?: "owner" | "terminal", lock = false): Promise<Entry | null> {
+    let sourceId = candidate.sourceId;
+    let activityAt = candidate.activityAt;
+    let dueAt = candidate.deadlineAt;
+    if (candidate.sourceKind === "decision") {
+      const [row] = await database.select({ originIssueId: decisions.originIssueId }).from(decisions)
+        .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;
+      }
+    } 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;
+      }
+    } else if (candidate.sourceKind === "blocker_attention") {
+      const row = await visibleIssue(database, candidate.companyId, sourceId);
+      if (!row) return null;
+      if (variant === "owner") {
+        if (row.unblockDescriptor?.owner !== "board" || !isProspectiveBlockedTransition(row)) return null;
+        const [activity] = await database.execute(sql`select blocked_transition_at::text as at,
+          ceil(extract(epoch from blocked_transition_at) * 1000 + ${archiveDelay}) as due from issues
+          where company_id = ${candidate.companyId}::uuid and id = ${row.id}::uuid`) as unknown as Array<{ at: string; due: string }>;
+        activityAt = activity.at; dueAt = Number(activity.due);
+      } else {
+        const attention = (await issueService(database).listBlockerAttention(candidate.companyId, [row])).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
+          where company_id = ${candidate.companyId}::uuid and id = ${sourceId}::uuid for update`);
+        const terminal = await visibleIssue(database, candidate.companyId, sourceId);
+        if (!terminal) return null;
+        const [activity] = await database.execute(sql`select updated_at::text as at,
+          ceil(extract(epoch from updated_at) * 1000 + ${archiveDelay}) as due from issues
+          where company_id = ${candidate.companyId}::uuid and id = ${sourceId}::uuid`) as unknown as Array<{ at: string; due: string }>;
+        activityAt = activity.at; dueAt = Number(activity.due);
+      }
+      const [retention] = await database.select().from(decisionRetention).where(and(eq(decisionRetention.companyId, candidate.companyId),
+        eq(decisionRetention.sourceKind, candidate.sourceKind), eq(decisionRetention.sourceId, sourceId)));
+      if (retention?.keep || retention?.archivedAt) return null;
+    } else if (candidate.sourceKind === "failed_run") {
+      const [run] = await database.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, candidate.companyId), eq(heartbeatRuns.id, sourceId)));
+      if (!run) return null;
+      const [agent] = await database.select({ id: agents.id }).from(agents).where(and(eq(agents.companyId, candidate.companyId),
+        eq(agents.id, run.agentId), sql`${agents.status} <> 'terminated'`));
+      const issueId = readRunIssueId(run.contextSnapshot);
+      if (!agent || issueId && !await visibleIssue(database, candidate.companyId, issueId)) return null;
+      if ((await listFailedRunsWithNewerRun(database, candidate.companyId, [{ ...run, issueId }])).has(run.id)) return null;
+    }
+    return { ...candidate, sourceId, activityAt, rootId: candidate.sourceId, variant,
+      id: `${candidate.sourceKind}:${candidate.sourceId}:${variant ?? "source"}`, dueAt };
+  }
+
+  async function load(scope?: Scope, signal?: AbortSignal) {
+    const key = JSON.stringify(scope ?? {});
+    const rows = await candidates(db, scope, cursors.get(key));
+    const entries: Entry[] = [];
+    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);
+        if (entry) entries.push(entry);
+      }
+    }
+    if (rows.length === 100) {
+      const last = rows[rows.length - 1]!;
+      cursors.set(key, { sourceKind: last.sourceKind, sourceId: last.sourceId });
+    } else cursors.delete(key);
+    return { rows: entries, complete: rows.length < 100 };
+  }
+
+  async function check(entry: Entry) {
+    return db.transaction(async transaction => {
+      const database = transaction as unknown as Db;
+      await database.execute(sql`select id from ${sourceTables[entry.sourceKind]}
+        where company_id = ${entry.companyId}::uuid and id = ${entry.rootId}::uuid for update`);
+      const [candidate] = await candidates(database, { companyId: entry.companyId, sourceKind: entry.sourceKind }, undefined, entry.rootId);
+      if (!candidate) return null;
+      const current = await eligible(database, candidate, entry.variant, true);
+      if (!current || current.dueAt > Date.now()) return current;
+      const retention = decisionRetentionService(database);
+      const item = { sourceKind: current.sourceKind, subject: { id: current.sourceId }, activityAt: current.activityAt };
+      const state = (await retention.syncItems(current.companyId, [item])).get(`${current.sourceKind}:${current.sourceId}`)!;
+      await retention.autoArchive({ companyId: current.companyId, items: [{ ...item,
+        keep: state.keep, archivedAt: state.archivedAt?.toISOString() ?? null, retentionVersion: state.version }] });
+      return null;
+    });
+  }
+  return { load, check };
+}
+
+export function registerAttentionRetentionWork(coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
+  service: ReturnType<typeof attentionRetentionService>, canRun: () => boolean) {
+  return registerScopedDeadlineWork(coordinator, {
+    queue: DELIVERY_QUEUES.attentionRetention,
+    decodeScope: hint => hint === "*" ? undefined : JSON.parse(hint) as Scope,
+    matches: (entry, scope) => (!scope?.companyId || scope.companyId === entry.companyId)
+      && (!scope?.sourceKind || scope.sourceKind === entry.sourceKind)
+      && (!scope?.sourceId || scope.sourceId === entry.sourceId),
+    load: service.load, check: service.check,
+  }, canRun);
+}

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

@@ -0,0 +1,108 @@
+import { and, eq, gt, inArray, sql, type SQL } from "drizzle-orm";
+import { heartbeatRuns, type Db } from "@paperclipai/db";
+
+export function collapsePendingConfirmationsToNewest<T extends {
+  id: string;
+  issueId: string;
+  kind: string;
+  createdAt: Date;
+  payload: unknown;
+}>(rows: T[]) {
+  const independent = (row: T) => Boolean(row.payload && typeof row.payload === "object" && "qualityReview" in row.payload && row.payload.qualityReview);
+  const newestByGroup = new Map<string, T>();
+  for (const row of rows) {
+    if (row.kind !== "request_confirmation" || independent(row)) continue;
+    const groupKey = `${row.issueId}:${row.kind}`;
+    const newest = newestByGroup.get(groupKey);
+    if (
+      !newest
+      || row.createdAt.getTime() > newest.createdAt.getTime()
+      || (row.createdAt.getTime() === newest.createdAt.getTime() && row.id > newest.id)
+    ) {
+      newestByGroup.set(groupKey, row);
+    }
+  }
+
+  return rows.filter((row) => (
+    row.kind !== "request_confirmation"
+    || independent(row)
+    || newestByGroup.get(`${row.issueId}:${row.kind}`)?.id === row.id
+  ));
+}
+
+export function readRunIssueId(contextSnapshot: Record<string, unknown> | null) {
+  const issueId = contextSnapshot?.issueId ?? contextSnapshot?.taskId;
+  return typeof issueId === "string" && issueId.length > 0 ? issueId : null;
+}
+
+type FailedRunProbe = { id: string; agentId: string; issueId: string | null; createdAt: Date };
+
+const FAILED_RUN_PROBE_BATCH_SIZE = 1000;
+
+export async function listFailedRunsWithNewerRun(db: Db, companyId: string, failedRuns: FailedRunProbe[]) {
+  const withNewerRun = new Set<string>();
+  const keyed = failedRuns.filter(run => run.issueId !== null);
+  for (let start = 0; start < keyed.length; start += FAILED_RUN_PROBE_BATCH_SIZE) {
+    const batch = keyed.slice(start, start + FAILED_RUN_PROBE_BATCH_SIZE);
+    for (const id of await listFailedRunBatchWithNewerRun(db, companyId, batch)) withNewerRun.add(id);
+  }
+  const unkeyed = failedRuns.filter(run => run.issueId === null);
+  if (unkeyed.length === 0) return withNewerRun;
+  const agentIds = [...new Set(unkeyed.map(run => run.agentId))];
+  const oldestCreatedAt = unkeyed.reduce((oldest, run) => run.createdAt < oldest ? run.createdAt : oldest,
+    unkeyed[0]!.createdAt);
+  const latestByAgent = new Map<string, Date>();
+  for (let start = 0; start < agentIds.length; start += FAILED_RUN_PROBE_BATCH_SIZE) {
+    const rows = await db.select({
+      agentId: heartbeatRuns.agentId,
+      createdAt: sql<Date>`max(${heartbeatRuns.createdAt})`.mapWith(heartbeatRuns.createdAt),
+    }).from(heartbeatRuns).where(and(
+      eq(heartbeatRuns.companyId, companyId),
+      inArray(heartbeatRuns.agentId, agentIds.slice(start, start + FAILED_RUN_PROBE_BATCH_SIZE)),
+      gt(heartbeatRuns.createdAt, oldestCreatedAt),
+      sql`coalesce(${heartbeatRuns.contextSnapshot} ->> 'issueId', ${heartbeatRuns.contextSnapshot} ->> 'taskId', '') = ''`,
+    )).groupBy(heartbeatRuns.agentId);
+    for (const row of rows) latestByAgent.set(row.agentId, row.createdAt);
+  }
+  for (const run of unkeyed) {
+    if ((latestByAgent.get(run.agentId)?.getTime() ?? 0) > run.createdAt.getTime()) withNewerRun.add(run.id);
+  }
+  return withNewerRun;
+}
+
+async function listFailedRunBatchWithNewerRun(db: Db, companyId: string, failedRuns: FailedRunProbe[]) {
+  if (failedRuns.length === 0) return [];
+  const failed = sql.join(
+    failedRuns.map((run) => sql`(${run.id}::uuid, ${run.agentId}::uuid, ${run.issueId}::text, ${new Date(run.createdAt.getTime() + 1).toISOString()}::timestamptz)`),
+    sql`, `,
+  );
+  const runIssueId = sql`(${heartbeatRuns.contextSnapshot} ->> 'issueId')`;
+  const runTaskId = sql`(${heartbeatRuns.contextSnapshot} ->> 'taskId')`;
+  const newerRunExists = (match: SQL) => sql`exists (
+    select 1 from ${heartbeatRuns}
+    where ${heartbeatRuns.companyId} = ${companyId}
+      and ${heartbeatRuns.agentId} = failed.agent_id
+      and ${heartbeatRuns.createdAt} >= failed.newer_from
+      and ${match}
+  )`;
+  const rows = await db.execute(sql`
+    select failed.id
+    from (values ${failed}) as failed (id, agent_id, issue_id, newer_from)
+    where ${newerRunExists(sql`${runIssueId} = failed.issue_id`)}
+      or ${newerRunExists(sql`${runTaskId} = failed.issue_id and ${runIssueId} is null`)}
+  `) as unknown as Array<{ id: string }>;
+  return rows.map((row) => row.id);
+}
+
+
+export function requiresReviewAttention(input: {
+  hasHumanParticipant: boolean; hasUserAssignee: boolean; hasPendingApproval: boolean; stalled: boolean;
+}) {
+  return input.hasHumanParticipant || input.hasUserAssignee || input.hasPendingApproval || input.stalled;
+}
+
+export function terminalBlockerRequiresAttention<T extends { state?: string; blockingTreeLive?: boolean }>(
+  attention: T | null | undefined,
+): attention is T & { state: "stalled" | "needs_attention" } {
+  return Boolean(attention && !attention.blockingTreeLive && ["stalled", "needs_attention"].includes(attention.state ?? ""));
+}

server/src/services/attention.ts

@@ -1,5 +1,6 @@
+import { collapsePendingConfirmationsToNewest, readRunIssueId, listFailedRunsWithNewerRun, requiresReviewAttention, terminalBlockerRequiresAttention } from "./attention-source-eligibility.js";
 import { approvalExecutionIssueCondition, firstApprovalIssueId } from "./approval-visibility.js";
-import { and, asc, desc, eq, gt, inArray, isNotNull, isNull, notInArray, or, sql, type SQL } from "drizzle-orm";
+import { and, asc, desc, eq, inArray, isNull, notInArray, or, sql, type SQL } from "drizzle-orm";
 import type { Db } from "@paperclipai/db";
 import {
   agents,
@@ -58,7 +59,7 @@ import {
   BLOCKER_ATTENTION_MAX_NODES,
   issueService,
 } from "./issues.js";
-import { executionIssueCondition, visibleIssueCondition } from "./issue-visibility.js";
+import { attentionIssueCondition, executionIssueCondition } from "./issue-visibility.js";
 import { parseIssueExecutionState } from "./issue-execution-policy.js";
 import { isProspectiveBlockedTransition } from "./routable-blocked.js";
 import { issueReadSqlCondition, type AuthorizationActor } from "./authorization.js";
@@ -766,37 +767,6 @@ export function interactionResolverAudience(
   };
 }
 
-function collapsePendingConfirmationsToNewest<T extends {
-  id: string;
-  issueId: string;
-  kind: string;
-  createdAt: Date;
-  payload: unknown;
-}>(rows: T[]) {
-  // Quality gates bind independent outputs; a newer visual approval must not
-  // hide a still-pending copy approval (or affect its badge count).
-  const independent = (row: T) => Boolean(row.payload && typeof row.payload === "object" && "qualityReview" in row.payload && row.payload.qualityReview);
-  const newestByGroup = new Map<string, T>();
-  for (const row of rows) {
-    if (row.kind !== "request_confirmation" || independent(row)) continue;
-    const groupKey = `${row.issueId}:${row.kind}`;
-    const newest = newestByGroup.get(groupKey);
-    if (
-      !newest
-      || row.createdAt.getTime() > newest.createdAt.getTime()
-      || (row.createdAt.getTime() === newest.createdAt.getTime() && row.id > newest.id)
-    ) {
-      newestByGroup.set(groupKey, row);
-    }
-  }
-
-  return rows.filter((row) => (
-    row.kind !== "request_confirmation"
-    || independent(row)
-    || newestByGroup.get(`${row.issueId}:${row.kind}`)?.id === row.id
-  ));
-}
-
 function budgetObservedPercent(amountObserved: number, amountLimit: number) {
   return amountLimit > 0 ? Math.round((amountObserved / amountLimit) * 10_000) / 100 : 0;
 }
@@ -866,9 +836,7 @@ async function issueSummaryMap(
     .where(and(
       eq(issues.companyId, companyId),
       inArray(issues.id, ids),
-      includeConversations
-        ? or(executionIssueCondition(), and(visibleIssueCondition(), isNotNull(issues.conversationAgentId)))
-        : executionIssueCondition(),
+      attentionIssueCondition(includeConversations),
       readCondition,
     ));
   return new Map(rows.map((row) => [row.id, {
@@ -1108,70 +1076,6 @@ function resolveBlockingIssue(
   return null;
 }
 
-function readRunIssueId(contextSnapshot: Record<string, unknown> | null) {
-  const issueId = contextSnapshot?.issueId ?? contextSnapshot?.taskId;
-  return typeof issueId === "string" && issueId.length > 0 ? issueId : null;
-}
-
-type FailedRunProbe = { id: string; agentId: string; issueId: string | null; createdAt: Date };
-
-const FAILED_RUN_PROBE_BATCH_SIZE = 1000;
-
-async function listFailedRunsWithNewerRun(db: Db, companyId: string, failedRuns: FailedRunProbe[]) {
-  const withNewerRun = new Set<string>();
-  const keyed = failedRuns.filter(run => run.issueId !== null);
-  for (let start = 0; start < keyed.length; start += FAILED_RUN_PROBE_BATCH_SIZE) {
-    const batch = keyed.slice(start, start + FAILED_RUN_PROBE_BATCH_SIZE);
-    for (const id of await listFailedRunBatchWithNewerRun(db, companyId, batch)) withNewerRun.add(id);
-  }
-  const unkeyed = failedRuns.filter(run => run.issueId === null);
-  if (unkeyed.length === 0) return withNewerRun;
-  const agentIds = [...new Set(unkeyed.map(run => run.agentId))];
-  const oldestCreatedAt = unkeyed.reduce((oldest, run) => run.createdAt < oldest ? run.createdAt : oldest,
-    unkeyed[0]!.createdAt);
-  const latestByAgent = new Map<string, Date>();
-  for (let start = 0; start < agentIds.length; start += FAILED_RUN_PROBE_BATCH_SIZE) {
-    const rows = await db.select({
-      agentId: heartbeatRuns.agentId,
-      createdAt: sql<Date>`max(${heartbeatRuns.createdAt})`.mapWith(heartbeatRuns.createdAt),
-    }).from(heartbeatRuns).where(and(
-      eq(heartbeatRuns.companyId, companyId),
-      inArray(heartbeatRuns.agentId, agentIds.slice(start, start + FAILED_RUN_PROBE_BATCH_SIZE)),
-      gt(heartbeatRuns.createdAt, oldestCreatedAt),
-      sql`coalesce(${heartbeatRuns.contextSnapshot} ->> 'issueId', ${heartbeatRuns.contextSnapshot} ->> 'taskId', '') = ''`,
-    )).groupBy(heartbeatRuns.agentId);
-    for (const row of rows) latestByAgent.set(row.agentId, row.createdAt);
-  }
-  for (const run of unkeyed) {
-    if ((latestByAgent.get(run.agentId)?.getTime() ?? 0) > run.createdAt.getTime()) withNewerRun.add(run.id);
-  }
-  return withNewerRun;
-}
-
-async function listFailedRunBatchWithNewerRun(db: Db, companyId: string, failedRuns: FailedRunProbe[]) {
-  if (failedRuns.length === 0) return [];
-  const failed = sql.join(
-    failedRuns.map((run) => sql`(${run.id}::uuid, ${run.agentId}::uuid, ${run.issueId}::text, ${new Date(run.createdAt.getTime() + 1).toISOString()}::timestamptz)`),
-    sql`, `,
-  );
-  const runIssueId = sql`(${heartbeatRuns.contextSnapshot} ->> 'issueId')`;
-  const runTaskId = sql`(${heartbeatRuns.contextSnapshot} ->> 'taskId')`;
-  const newerRunExists = (match: SQL) => sql`exists (
-    select 1 from ${heartbeatRuns}
-    where ${heartbeatRuns.companyId} = ${companyId}
-      and ${heartbeatRuns.agentId} = failed.agent_id
-      and ${heartbeatRuns.createdAt} >= failed.newer_from
-      and ${match}
-  )`;
-  const rows = await db.execute(sql`
-    select failed.id
-    from (values ${failed}) as failed (id, agent_id, issue_id, newer_from)
-    where ${newerRunExists(sql`${runIssueId} = failed.issue_id`)}
-      or ${newerRunExists(sql`${runTaskId} = failed.issue_id and ${runIssueId} is null`)}
-  `) as unknown as Array<{ id: string }>;
-  return rows.map((row) => row.id);
-}
-
 export function attentionService(db: Db, serviceOptions: AttentionServiceOptions = {}) {
   const openDecisionLimit = Math.min(
     Math.max(Math.trunc(serviceOptions.openDecisionLimit ?? OPEN_DECISION_DEFAULT_LIMIT), 1),
@@ -1636,8 +1540,7 @@ export function attentionService(db: Db, serviceOptions: AttentionServiceOptions
           }));
         }
         const blockerAttention = issue.blockerAttention;
-        if (blockerAttention?.state !== "stalled" && blockerAttention?.state !== "needs_attention") continue;
-        if (blockerAttention.blockingTreeLive) continue;
+        if (!terminalBlockerRequiresAttention(blockerAttention)) continue;
         const issueSummary = blockedIssueSummaries.get(issue.id) ?? null;
         const terminalIssueId = blockerAttention.terminalBlockerIssueId ?? issue.id;
         const terminalSummary = terminalBlockerSummaries.get(terminalIssueId)
@@ -1739,7 +1642,8 @@ export function attentionService(db: Db, serviceOptions: AttentionServiceOptions
         const pendingApprovalId = pendingApprovalByIssueId.get(review.id) ?? null;
         const reviewAttention = reviewAttentionByIssueId.get(review.id);
         const stalled = reviewAttention?.state === "stalled";
-        if (!hasHumanParticipant && !review.assigneeUserId && !pendingApprovalId && !stalled) continue;
+        if (!requiresReviewAttention({ hasHumanParticipant, hasUserAssignee: Boolean(review.assigneeUserId),
+          hasPendingApproval: Boolean(pendingApprovalId), stalled })) continue;
         const issue = reviewIssueMap.get(review.id);
         if (!issue) continue;
         const dedupKey = `review:${review.id}`;

server/src/services/budgets.ts

@@ -1,5 +1,5 @@
 import { notifyApprovalWork } from "./approval-work-notifications.js";
-import { notifyCompanyRetentionWork } from "./retention-work-notifications.js";
+import { notifyAttentionRetentionWork, notifyCompanyRetentionWork } from "./retention-work-notifications.js";
 import { addCents, compareCents, normalizeCents } from "@paperclipai/shared";
 import { and, desc, eq, inArray, sql, or, isNull } from "drizzle-orm";
 import type { Db } from "@paperclipai/db";
@@ -466,6 +466,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica
       windowEnd: end,
     });
 
+    await notifyAttentionRetentionWork(db, { companyId: policy.companyId, sourceKind: "budget_alert" });
     if (thresholdType === "hard") await notifyApprovalWork(db, policy.companyId);
     const approval = thresholdType === "hard"
       ? await db
@@ -505,6 +506,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica
   }
 
   async function resolveOpenSoftIncidents(policyId: string) {
+    await notifyAttentionRetentionWork(db, { sourceKind: "budget_alert" });
     await db
       .update(budgetIncidents)
       .set({
@@ -527,6 +529,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica
     decidedByUserId: string | null,
     thresholdType?: BudgetThresholdType,
   ) {
+    await notifyAttentionRetentionWork(db, { sourceKind: "budget_alert" });
     const condition = and(eq(budgetIncidents.policyId, policyId), eq(budgetIncidents.status, "open"),
       thresholdType ? eq(budgetIncidents.thresholdType, thresholdType) : undefined);
     const openRows = await db
@@ -937,6 +940,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica
         }
 
         const now = new Date();
+        await notifyAttentionRetentionWork(db, { companyId, sourceKind: "budget_alert" });
         await notifyDeliveryWork(db, DELIVERY_QUEUES.budgetEnforcement, companyId);
         await db
           .update(budgetPolicies)

server/src/services/company-portability.ts

@@ -1,3 +1,4 @@
+import { notifyAttentionRetentionWork } from "./retention-work-notifications.js";
 import { createAgentLifecycle } from "./agent-lifecycle.js";
 import { agentAppearanceSchema, agentHarnessType, paperclipRunnerProfileForHarness, agentRunner, type AgentRunnerChoice } from "@paperclipai/shared";
 import { resolveNewAgentRunnerForCompany } from "./agent-runner-selection.js";
@@ -6465,7 +6466,9 @@ export function companyPortabilityService(db: Db, storage?: StorageService) {
           }
           if (relationRows.length > 0) {
             const relationCompanyId = targetCompany.id;
-            await db
+            await db.transaction(async tx => {
+              await notifyAttentionRetentionWork(tx, { companyId: relationCompanyId });
+            return tx
               .insert(issueRelations)
               .values(relationRows.map((row) => ({
                 companyId: relationCompanyId,
@@ -6476,6 +6479,7 @@ export function companyPortabilityService(db: Db, storage?: StorageService) {
                 createdByUserId: actorUserId ?? null,
               })))
               .onConflictDoNothing();
+            });
           }
         }
 

server/src/services/decision-retention.ts

@@ -1,3 +1,4 @@
+import { isProspectiveBlockedTransition } from "./routable-blocked.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";
@@ -15,7 +16,6 @@ import {
   issueRecoveryActions,
   issueThreadInteractions,
   issues,
-  joinRequests,
 } from "@paperclipai/db";
 import type {
   AttentionArchiveManifestEntry,
@@ -145,7 +145,7 @@ export function decisionRetentionService(
       const existing = new Set(rows.map(row => sourceKey(row.sourceKind, row.sourceId)));
       const missing = unique.filter(item => !existing.has(itemSourceKey(item)));
       if (missing.length) {
-        for (const kind of new Set(missing.map(item => item.sourceKind))) await notifyRetentionWork(tx, kind);
+        for (const item of missing) await notifyRetentionWork(tx, item.sourceKind, { companyId, sourceId: item.subject.id });
         await tx.insert(decisionRetention).values(missing.map(item => ({
           companyId, sourceKind: item.sourceKind, sourceId: item.subject.id, sourceActivityAt: new Date(item.activityAt),
         }))).onConflictDoNothing({ target: [decisionRetention.companyId, decisionRetention.sourceKind, decisionRetention.sourceId] });
@@ -155,7 +155,7 @@ export function decisionRetentionService(
       for (const item of unique) {
         const key = itemSourceKey(item), row = byKey.get(key), activityAt = new Date(item.activityAt);
         if (!row || row.sourceActivityAt.getTime() === activityAt.getTime()) continue;
-        await notifyRetentionWork(tx, item.sourceKind);
+        await notifyRetentionWork(tx, item.sourceKind, { companyId, sourceId: item.subject.id });
         const [updated] = await tx.update(decisionRetention).set({
           sourceActivityAt: activityAt, version: sql`${decisionRetention.version} + 1`, updatedAt: new Date(),
         }).where(and(eq(decisionRetention.id, row.id), ne(decisionRetention.sourceActivityAt, activityAt))).returning();
@@ -169,9 +169,9 @@ export function decisionRetentionService(
     return sourceKey(item.sourceKind, item.subject.id);
   }
 
-  async function writeRetention<T>(kind: AttentionSourceKind, write: (tx: Db) => Promise<T>) {
+  async function writeRetention<T>(companyId: string, kind: AttentionSourceKind, sourceId: string, write: (tx: Db) => Promise<T>) {
     return db.transaction(async tx => {
-      await notifyRetentionWork(tx, kind);
+      await notifyRetentionWork(tx, kind, { companyId, sourceId });
       return write(tx as unknown as Db);
     });
   }
@@ -185,13 +185,28 @@ export function decisionRetentionService(
   }
 
   async function materializeSourceState(database: Db, input: { companyId: string; sourceKind: AttentionSourceKind; sourceId: string }) {
-    const table = input.sourceKind === "approval" ? approvals : input.sourceKind === "join_request" ? joinRequests : null;
-    if (!table) return;
-    const [source] = await database.select({ updatedAt: table.updatedAt }).from(table)
-      .where(and(eq(table.companyId, input.companyId), eq(table.id, input.sourceId))).for("update");
+    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 [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)
+        ? source.blockedTransitionAt : source.updatedAt;
+      await decisionRetentionService(database).syncItems(input.companyId, [{
+        sourceKind: input.sourceKind, subject: { id: input.sourceId }, activityAt: activityAt.toISOString(),
+      }]);
+      return;
+    }
+    const activity = input.sourceKind === "failed_run" ? sql`coalesce(finished_at, updated_at, created_at)` : sql`updated_at`;
+    const [source] = await database.execute(sql`select ${activity} as "activityAt" from ${sql.identifier(tables[input.sourceKind])}
+      where company_id = ${input.companyId}::uuid and id = ${input.sourceId}::uuid for update`) as unknown as Array<{ activityAt: Date | string }>;
     if (!source) throw notFound("Attention source not found");
     await decisionRetentionService(database).syncItems(input.companyId, [{
-      sourceKind: input.sourceKind, subject: { id: input.sourceId }, activityAt: source.updatedAt.toISOString(),
+      sourceKind: input.sourceKind, subject: { id: input.sourceId }, activityAt: new Date(source.activityAt).toISOString(),
     }]);
   }
 
@@ -206,7 +221,7 @@ export function decisionRetentionService(
     if (!(await canReadDecisionSource(db, input.authActor, input.companyId, input.sourceKind, input.sourceId))) {
       throw notFound("Attention source not found");
     }
-    const updated = await writeRetention(input.sourceKind, async tx => {
+    const updated = await writeRetention(input.companyId, input.sourceKind, input.sourceId, async tx => {
       await materializeSourceState(tx, input);
       return tx.update(decisionRetention).set({
         keep: input.keep,
@@ -247,7 +262,7 @@ export function decisionRetentionService(
       throw notFound("Attention source not found");
     }
     const now = new Date();
-    const updated = await writeRetention(input.sourceKind, async tx => {
+    const updated = await writeRetention(input.companyId, input.sourceKind, input.sourceId, async tx => {
       await materializeSourceState(tx, input);
       return tx.update(decisionRetention).set(input.archived ? {
         archivedAt: now,
@@ -324,7 +339,7 @@ export function decisionRetentionService(
       }
       const now = new Date();
       for (const entry of input.manifest) {
-        await notifyRetentionWork(txDb, entry.sourceKind as AttentionSourceKind);
+        await notifyRetentionWork(txDb, entry.sourceKind as AttentionSourceKind, { companyId: input.companyId, sourceId: entry.sourceId });
         const archived = await txDb.update(decisionRetention).set({
           archivedAt: now,
           archivedReason: "accepted_proposal",
@@ -368,7 +383,7 @@ export function decisionRetentionService(
     await db.transaction(async (tx) => {
       const txDb = tx as unknown as Db;
       for (const item of candidates) {
-        await notifyRetentionWork(txDb, item.sourceKind);
+        await notifyRetentionWork(txDb, item.sourceKind, { companyId: input.companyId, sourceId: item.subject.id });
         const updated = await txDb.update(decisionRetention).set({
           archivedAt: now,
           archivedReason: "idle_ttl",

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

@@ -1,11 +1,14 @@
+import { notifyAttentionRetentionWork } from "./retention-work-notifications.js";
 import type { Db } from "@paperclipai/db";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
 
-export function notifyDecisionCreated(transaction: object) {
+export async function notifyDecisionCreated(transaction: object) {
+  await notifyAttentionRetentionWork(transaction, { sourceKind: "decision" });
   return notifyDeliveryWork(transaction, DELIVERY_QUEUES.decisionExpiry);
 }
 
 export async function notifyDecisionChange(transaction: object) {
+  await notifyAttentionRetentionWork(transaction, { sourceKind: "decision" });
   await notifyDeliveryWork(transaction, DELIVERY_QUEUES.decisionContinuation);
   await notifyDeliveryWork(transaction, DELIVERY_QUEUES.decisionExpiry);
 }

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

@@ -35,6 +35,7 @@ export const DELIVERY_QUEUES = {
   executionDisposition: "execution-disposition",
   environmentCleanup: "environment-cleanup",
   approvalQueueSeeding: "approval-queue-seeding",
+  attentionRetention: "attention-retention",
   approvalRetention: "approval-retention",
   joinRequestRetention: "join-request-retention",
   decisionTargets: "decision-targets",

server/src/services/execution-control-notifications.ts

@@ -1,7 +1,9 @@
+import { notifyAttentionRetentionWork } from "./retention-work-notifications.js";
 import type { Db } from "@paperclipai/db";
 import { DELIVERY_QUEUES, notifyDeliveryWork, type DeliveryQueue } from "./delivery-work-notifications.js";
 
 export async function notifyExecutionRunWork(transaction: object) {
+  await notifyAttentionRetentionWork(transaction);
   for (const queue of [DELIVERY_QUEUES.executionStatus, DELIVERY_QUEUES.executionFinalization,
     DELIVERY_QUEUES.executionReplacement, DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.issueMonitors]) {
     await notifyDeliveryWork(transaction, queue);
@@ -14,6 +16,7 @@ export function writeExecutionControl<T>(
   write: (tx: Db) => PromiseLike<T>,
 ): Promise<T> {
   return db.transaction(async transaction => {
+    await notifyAttentionRetentionWork(transaction);
     await notifyDeliveryWork(transaction, DELIVERY_QUEUES.issueMonitors);
     for (const topic of typeof queue === "string" ? [queue] : queue) await notifyDeliveryWork(transaction, topic);
     return write(transaction as unknown as Db);

server/src/services/grok-bot/workspace-inspection.ts

@@ -1,3 +1,4 @@
+import { notifyAttentionRetentionWork } from "../retention-work-notifications.js";
 import { and, eq, inArray, isNotNull, or, sql } from "drizzle-orm";
 import { dotRunnerAssignments, environmentLeases, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations, type Db } from "@paperclipai/db";
 import { authorizeExternalBotOperator } from "../external-bot-authority.js";
@@ -74,6 +75,7 @@ export async function inspectGrokWorkspaceObligations(db: Db, companyId: string,
         return { ...rest, externalWorkspaceInspection: inspection };
       };
       await tx.update(heartbeatRuns).set({ resultJson: retireFileObligation(run.resultJson!) }).where(eq(heartbeatRuns.id, run.id));
+      if (actions.length) await notifyAttentionRetentionWork(tx, { companyId, sourceKind: "recovery_action" });
       for (const action of actions) {
         await tx.update(issueRecoveryActions).set({ evidence: retireFileObligation(action.evidence), updatedAt: new Date(),
           nextAction: "Inspect Grok and verify its connection before sending a new task message. The old work will not be replayed.",

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

@@ -1,3 +1,4 @@
+import { notifyAttentionRetentionWork } from "./retention-work-notifications.js";
 import { readNativeEventEvidenceOutcome } from "./native-runtime/evidence-classifier.js";
 import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js";
 import { isNativeResponseEvidence, notifyNativeResponsePresentation } from "./native-response-presentation-notifications.js";
@@ -74,10 +75,11 @@ export async function appendHeartbeatRunEvent(
   db: Db,
   input: AppendHeartbeatRunEventInput,
 ): Promise<AppendHeartbeatRunEventResult> {
+  const projectsRetention = input.eventType === "lifecycle" && Boolean(input.message?.startsWith("Bounded retry exhausted"));
   const projectsReassessment = readNativeEventEvidenceOutcome(input.payload?.prpEvent) !== null;
   const projectsProgress = isSafeNativeChatProgressEventType(input.eventType);
   const projectsResponse = isNativeResponseEvidence(input.eventType, input.payload?.prpEvent);
-  if (!input.nativeSource && !input.retryExhaustion && !projectsProgress && !projectsResponse && !projectsReassessment) {
+  if (!input.nativeSource && !input.retryExhaustion && !projectsProgress && !projectsResponse && !projectsReassessment && !projectsRetention) {
     const allocated = db.$with("allocated_run_event", {}).as(sql`
       update ${heartbeatRuns}
       set next_event_seq = ${heartbeatRuns.nextEventSeq} + 1,
@@ -187,6 +189,7 @@ export async function appendHeartbeatRunEvent(
     if (projectsResponse && run.runtimeMode === "native") {
       await notifyNativeResponsePresentation(tx, { companyId: run.companyId, runId: input.runId });
     }
+    if (projectsRetention) await notifyAttentionRetentionWork(tx, { companyId: input.companyId, sourceKind: "failed_run" });
     const seq = await allocateHeartbeatRunEventSeq(
       tx as unknown as Db,
       input.runId,

server/src/services/heartbeat/agent-status.ts

@@ -1,3 +1,4 @@
+import { notifyAttentionRetentionWork } from "../retention-work-notifications.js";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js";
 import { agents, heartbeatRuns, type Db } from "@paperclipai/db";
 import { and, asc, desc, eq, lte, sql } from "drizzle-orm";
@@ -91,6 +92,7 @@ export function createHeartbeatAgentStatus(db: Db) {
       )).limit(1);
       const nextStatus = running ? "running" : settledStatus(outcome, options?.keepIdleOnFailure);
       const now = new Date();
+      await notifyAttentionRetentionWork(tx, { companyId: existing.companyId });
       await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
       const [updated] = await tx.update(agents).set({
         status: nextStatus,
@@ -140,6 +142,7 @@ export function createHeartbeatAgentStatus(db: Db) {
             eq(heartbeatRuns.agentId, candidate.id)))
           .orderBy(desc(heartbeatRuns.finishedAt), desc(heartbeatRuns.createdAt), desc(heartbeatRuns.id)).limit(1);
         const nextStatus = settledStatus(lastRun?.status);
+        await notifyAttentionRetentionWork(tx, { companyId: candidate.companyId });
         const [agent] = await tx.update(agents).set({ status: nextStatus,
           errorReason: nextStatus === "error" ? errorReason(lastRun?.error) : null, updatedAt: now,
         }).where(and(eq(agents.id, candidate.id), eq(agents.companyId, candidate.companyId),

server/src/services/heartbeat/retries.ts

@@ -1,3 +1,4 @@
+import { notifyAttentionRetentionWork } from "../retention-work-notifications.js";
 import { deriveTaskKeyWithHeartbeatFallback } from "./run-state.js";
 import { infrastructureRetryDelayMs } from "../system-recovery-state.js";
 import {
@@ -997,6 +998,7 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen
           retryContextSnapshot.previousRunId = continuation.previousRunId;
         }
 
+        await notifyAttentionRetentionWork(tx, { companyId: run.companyId });
         const wakeupRequest = await tx
           .insert(agentWakeupRequests)
           .values({

server/src/services/issue-recovery-actions.ts

@@ -1,3 +1,4 @@
+import { notifyAttentionRetentionWork } from "./retention-work-notifications.js";
 import { notifyNativeRecoveryNotices } from "./native-recovery-notice-notifications.js";
 import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js";
 import { writeExecutionControl } from "./execution-control-notifications.js";
@@ -322,6 +323,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) {
   ): Promise<IssueRecoveryAction> {
     try {
       const created = await db.transaction(async (tx) => {
+        await notifyAttentionRetentionWork(tx, { companyId: input.companyId });
         await notifyNativeRecoveryNotices(tx, { companyId: input.companyId, issueId: input.sourceIssueId });
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.executionDisposition);
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.executionContinuation);

server/src/services/issue-review-state-notifications.ts

@@ -1,9 +1,11 @@
+import { notifyAttentionRetentionWork } from "./retention-work-notifications.js";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
 import { notifyExternalConversationSettlement } from "./external-conversation-settlement-notifications.js";
 import { notifyNativeReviewResponse } from "./native-review-response-notifications.js";
 import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js";
 
 export async function notifyIssueReviewState(tx: object, scope: { companyId: string; issueId?: string; runId?: string; agentId?: string }) {
+  await notifyAttentionRetentionWork(tx, { companyId: scope.companyId });
   await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
   await notifyExternalConversationSettlement(tx, scope);
   await notifyNativeReviewResponse(tx, scope);

server/src/services/issue-title.ts

@@ -1,3 +1,4 @@
+import { notifyAttentionRetentionWork } from "./retention-work-notifications.js";
 import { and, eq } from "drizzle-orm";
 import { heartbeatRuns, issues, type Db } from "@paperclipai/db";
 import { setIssueTitleSchema, type SetIssueTitle } from "@paperclipai/shared";
@@ -76,6 +77,7 @@ export async function setIssueTitle(
     || (issue.title === title && !issue.titleNeedsGeneration)) {
     return { result: await saveReceipt({ id: issue.id, title: issue.title, titleNeedsGeneration: issue.titleNeedsGeneration, changed: false }), publication: null };
   }
+  await notifyAttentionRetentionWork(tx, { companyId });
   await tx.update(issues).set({ title, titleNeedsGeneration: false, updatedAt: new Date() }).where(predicate);
   await issueReferenceService(tx).syncIssue(issueId, tx);
   const { publication } = await persistActivity(tx, {

server/src/services/issue-visibility.ts

@@ -1,5 +1,5 @@
 import { nonIdleSlackIssueCondition } from "./slack-conversation-state.js";
-import { and, isNull, type SQL } from "drizzle-orm";
+import { and, isNull, isNotNull, or, type SQL } from "drizzle-orm";
 import { issues } from "@paperclipai/db";
 
 export function visibleIssueCondition(): SQL {
@@ -14,3 +14,9 @@ export function visibleIssueSql(alias = "issues") {
 export function executionIssueCondition(): SQL {
   return and(visibleIssueCondition(), isNull(issues.conversationAgentId), nonIdleSlackIssueCondition())!;
 }
+
+export function attentionIssueCondition(includeConversations = false): SQL {
+  return includeConversations
+    ? or(executionIssueCondition(), and(visibleIssueCondition(), isNotNull(issues.conversationAgentId)))!
+    : executionIssueCondition();
+}

server/src/services/issues.ts

@@ -1,3 +1,4 @@
+import { notifyAttentionRetentionWork } from "./retention-work-notifications.js";
 import { notifyExternalConversationSettlement } from "./external-conversation-settlement-notifications.js";
 import { notifyNativeRecoveryNotices } from "./native-recovery-notice-notifications.js";
 import { notifyNativeRunProjection } from "./native-run-projection-notifications.js";
@@ -7626,6 +7627,7 @@ export function issueService(db: Db) {
       await assertNoBlockingCycles(companyId, issueId, deduped, dbOrTx);
     }
 
+    await notifyAttentionRetentionWork(dbOrTx, { companyId });
     await dbOrTx
       .delete(issueRelations)
       .where(
@@ -10606,6 +10608,7 @@ export function issueService(db: Db) {
         }
         if (values.monitorNextCheckAt) await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
+        await notifyAttentionRetentionWork(tx, { companyId });
         const [issue] = await tx.insert(issues).values(values).returning();
         await attachOwnedDraftImages(tx, issue, issue.description, { userId: issueData.createdByUserId, agentId: issueData.createdByAgentId });
         if (issue.visibility === "private" && !inheritedPrivateSubtree && issueData.createdByAgentId) {
@@ -11491,6 +11494,7 @@ export function issueService(db: Db) {
           || issueData.parentId !== undefined || issueData.executionWorkspaceId !== undefined) {
           await notifyTerminalWorkspaceWork(tx, `issue:${id}`);
         }
+        await notifyAttentionRetentionWork(tx, { companyId: receiptExisting.companyId });
         if (patch.status !== undefined || patch.assigneeAgentId !== undefined || patch.assigneeUserId !== undefined
           || patch.hiddenAt !== undefined || patch.conversationAgentId !== undefined) {
           await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
@@ -11952,6 +11956,7 @@ export function issueService(db: Db) {
           await requestRestorePointDeletion(tx, { sourceIssueId: id });
           const [owner] = await tx.select({ companyId: issues.companyId }).from(issues).where(eq(issues.id, id));
           if (owner) {
+            await notifyAttentionRetentionWork(tx, { companyId: owner.companyId });
             await notifyExternalConversationSettlement(tx, { companyId: owner.companyId });
             await notifyApprovalWork(tx, owner.companyId);
           }
@@ -12084,6 +12089,7 @@ export function issueService(db: Db) {
           )
         : isNull(issues.executionRunId);
       const updated = await db.transaction(async tx => {
+        await notifyAttentionRetentionWork(tx, { companyId: issueCompany.companyId });
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
         return tx
@@ -12213,6 +12219,7 @@ export function issueService(db: Db) {
           }
           const staleExecutionRunId = current.executionRunId;
           const adopted = await db.transaction(async tx => {
+            await notifyAttentionRetentionWork(tx, { companyId: issueCompany.companyId });
             await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
             await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
             return tx
@@ -12468,6 +12475,7 @@ export function issueService(db: Db) {
           existing.status === "in_progress" ? "todo" : existing.status;
         await notifyTerminalWorkspaceWork(tx, `issue:${id}`);
         await notifyNativeStatusReassessment(tx, { companyId: existing.companyId, issueId: id });
+        await notifyAttentionRetentionWork(tx, { companyId: existing.companyId });
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
         const updated = await tx
@@ -12520,6 +12528,7 @@ export function issueService(db: Db) {
           patch.assigneeAgentId = null;
         }
 
+        await notifyAttentionRetentionWork(tx, { companyId: existing.companyId });
         await notifyTerminalWorkspaceWork(tx, `issue:${id}`);
         if (existing.checkoutRunId || existing.executionRunId || options.clearAssignee) {
           await notifyExternalConversationSettlement(tx, { companyId: existing.companyId, issueId: id });

server/src/services/native-finalization-notifications.ts

@@ -1,3 +1,4 @@
+import { notifyAttentionRetentionWork } from "./retention-work-notifications.js";
 import { notifyNativeEnvironmentSettlement } from "./native-environment-settlement-notifications.js";
 import { notifyExternalAgentSettlement } from "../modules/external-agents/index.js";
 import { notifyExternalConversationSettlement } from "./external-conversation-settlement-notifications.js";
@@ -14,6 +15,7 @@ import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notificatio
 export type NativeFinalizationWorkScope = { companyId: string; runId?: string; issueId?: string };
 
 export async function notifyNativeFinalizationWork(transaction: object, scope: NativeFinalizationWorkScope) {
+  await notifyAttentionRetentionWork(transaction, { companyId: scope.companyId });
   if (scope.issueId || scope.runId) {
     await notifyNativeEnvironmentSettlement(transaction, scope);
     await notifyExternalAgentSettlement(transaction, scope);

server/src/services/native-runtime/native-provider-capacity-retry.ts

@@ -1,3 +1,4 @@
+import { notifyAttentionRetentionWork } from "../retention-work-notifications.js";
 import { randomUUID } from "node:crypto";
 import { and, eq } from "drizzle-orm";
 import { agentWakeupRequests, heartbeatRuns, type Db } from "@paperclipai/db";
@@ -49,6 +50,7 @@ export async function materializeNativeProviderCapacityRetry(input: {
     scheduledRetryAt: dueAt.toISOString(),
     executionRetryAccounting: accountingForScheduledRetry(run, NATIVE_PROVIDER_OVERLOADED_CODE, attempt),
   };
+  await notifyAttentionRetentionWork(input.tx, { companyId: input.companyId });
   const [wake] = await input.tx.insert(agentWakeupRequests).values({
     companyId: run.companyId, agentId: run.agentId, source: "automation", triggerDetail: "system",
     reason: NATIVE_PROVIDER_OVERLOADED_CODE, status: "queued", runId: retryId,

server/src/services/native-runtime/native-review-participant.ts

@@ -1,3 +1,4 @@
+import { notifyAttentionRetentionWork } from "../retention-work-notifications.js";
 import {
   agents,
   heartbeatRuns,
@@ -240,6 +241,7 @@ export async function claimNativeReviewExecutionLock(db: Db, input: {
     if (!run || !["queued", "running"].includes(run.status) || !admitted || !expected
       || admitted.nativeReviewInteractionId !== expected.nativeReviewInteractionId
       || admitted.nativeReviewDecisionId !== expected.nativeReviewDecisionId) return false;
+    await notifyAttentionRetentionWork(tx, { companyId: input.companyId });
     const claimed = await tx.update(issues).set({
       executionRunId: input.runId,
       executionAgentNameKey: input.agentNameKey,

server/src/services/native-runtime/native-workspace-reconstruction.ts

@@ -1,3 +1,4 @@
+import { notifyAttentionRetentionWork } from "../retention-work-notifications.js";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js";
 import { notifyUnsafeWorkspaceExportRecovery } from "../unsafe-workspace-export-notifications.js";
 import { and, eq, gt, inArray, ne, notInArray, or, sql } from "drizzle-orm";
@@ -179,6 +180,7 @@ export async function reconstructNativeWorkspace(input: {
         ...(source.interveningExecutions.length ? { interveningExecutions: source.interveningExecutions } : {}) } };
     await markExecutionReconciliation(db, source.action, decision, input.actorId);
     const now = new Date();
+    await notifyAttentionRetentionWork(tx, { companyId: input.companyId });
     await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: input.companyId, issueId: input.issueId });
     await tx.update(issueRecoveryActions).set({ status: "resolved", outcome: "handed_back", resolvedAt: now, updatedAt: now,
       resolutionNote: "workspace_reconstruction_authorized", nextAction: "Continue from saved sources. The original unexported edits are lost; verify the selected attachments before rebuilding. Do not repeat completed external actions or reset the original work limits.",

server/src/services/native-runtime/paperclip-runner-tool-authority.ts

@@ -1,4 +1,5 @@
 import { nativeRunKeyScope, nativeSkillTestToolAllowed, assertNativeSkillTestTool } from "./native-skill-test-scope.js";
+import { notifyAttentionRetentionWork } from "../retention-work-notifications.js";
 import { requiredQualityReviewSchema } from "@paperclipai/shared";
 import { priorRecoveryApiReceipt } from "../recovery-effect-receipts.js";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js";
@@ -2067,6 +2068,7 @@ export class PaperclipRunnerToolAuthority {
         replayMonitor = summarizeIssueMonitor(target, normalizeIssueExecutionPolicy(target.executionPolicy));
         return { taskId: target.id, status: target.status, monitor: replayMonitor };
       }
+      await notifyAttentionRetentionWork(tx, { companyId: this.binding.companyId });
       const patch = prepareIssueMonitorUpdate(target, input.monitor, actor);
       await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
       const [updated] = await tx.update(issues).set({ ...patch, updatedAt: new Date() })

server/src/services/native-runtime/superseded-external-review-requests.ts

@@ -1,3 +1,4 @@
+import { notifyAttentionRetentionWork } from "../retention-work-notifications.js";
 import { and, eq, isNull, ne, sql } from "drizzle-orm";
 import { completionContracts, heartbeatRuns, issues, issueThreadInteractions, statusDecisionEffects, statusDecisions, workAssessments, type Db } from "@paperclipai/db";
 import { hasPublishedGitHubReview } from "../chat-channels/github/review-wait.js";
@@ -70,6 +71,7 @@ export async function supersedeExternalReviewRequestsInTransaction(
   publications: ActivityPublication[],
 ) {
   for (const { interaction } of candidates) {
+    await notifyAttentionRetentionWork(tx, { companyId: binding.companyId });
     const reason = "A later published review no longer requests human input.";
     const [retired] = await tx.update(issueThreadInteractions).set({ status: "cancelled",
       result: { version: 1, outcome: "withdrawn", reason }, resolvedAt: new Date(), updatedAt: new Date(),

server/src/services/native-status-reassessment-notifications.ts

@@ -1,8 +1,10 @@
+import { notifyAttentionRetentionWork } from "./retention-work-notifications.js";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
 
 export type NativeStatusReassessmentScope = { companyId: string; issueId?: string; runId?: string; agentId?: string };
 
-export function notifyNativeStatusReassessment(tx: object, scope: NativeStatusReassessmentScope) {
+export async function notifyNativeStatusReassessment(tx: object, scope: NativeStatusReassessmentScope) {
+  await notifyAttentionRetentionWork(tx, { companyId: scope.companyId });
   return notifyDeliveryWork(tx, DELIVERY_QUEUES.nativeStatusReassessment,
     JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId, runId: scope.runId, agentId: scope.agentId }));
 }

server/src/services/plugin-managed-agents.ts

@@ -1,3 +1,4 @@
+import { notifyAttentionRetentionWork } from "./retention-work-notifications.js";
 import { createAgentLifecycle } from "./agent-lifecycle.js";
 import { and, eq, isNull, ne } from "drizzle-orm";
 import type { Db } from "@paperclipai/db";
@@ -340,6 +341,7 @@ export function pluginManagedAgentService(
       // and its canonical history. Transfer only its proven stable-key marker to
       // the replacement installation, atomically with the new scoped bindings.
       await upsertBinding(companyId, declaration, agentId, {}, adapterType, tx);
+      await notifyAttentionRetentionWork(tx, { companyId, sourceKind: "agent_error_alert" });
       await tx.update(agents).set({
         metadata: managedMetadata(options.pluginId, options.pluginKey, declaration, agent.metadata),
         updatedAt: new Date(),

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

@@ -3,12 +3,19 @@ import type { Db } from "@paperclipai/db";
 import type { AttentionSourceKind } from "@paperclipai/shared";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
 
-export async function notifyRetentionWork(transaction: object, kind: AttentionSourceKind) {
+export async function notifyAttentionRetentionWork(transaction: object, scope: { companyId?: string; sourceKind?: AttentionSourceKind; sourceId?: string } = {}) {
+  await notifyDeliveryWork(transaction, DELIVERY_QUEUES.attentionRetention, JSON.stringify(scope));
+}
+
+export async function notifyRetentionWork(transaction: object, kind: AttentionSourceKind, scope: { companyId?: string; sourceId?: string } = {}) {
   if (kind === "approval") await notifyDeliveryWork(transaction, DELIVERY_QUEUES.approvalRetention);
   if (kind === "join_request") await notifyDeliveryWork(transaction, DELIVERY_QUEUES.joinRequestRetention);
+  if (kind !== "approval" && kind !== "join_request") await notifyAttentionRetentionWork(transaction, { ...scope, sourceKind: kind,
+    sourceId: kind === "blocker_attention" ? undefined : scope.sourceId });
 }
 
 export async function notifyCompanyRetentionWork(transaction: object, companyId: string) {
+  await notifyAttentionRetentionWork(transaction, { companyId });
   await notifyApprovalWork(transaction, companyId);
   await notifyRetentionWork(transaction, "join_request");
 }

server/src/services/slack-conversation-lifecycle.ts

@@ -1,3 +1,4 @@
+import { notifyAttentionRetentionWork } from "./retention-work-notifications.js";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
 import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js";
 import { and, asc, desc, eq, gt, inArray, isNull, sql } from "drizzle-orm";
@@ -126,6 +127,7 @@ export async function settleExternalConversation(db: Db, companyId: string, issu
     const ready = await readSettlement(tx, companyId, issueId, true);
     if (!ready) return false;
     const { binding, issue, run } = ready;
+    await notifyAttentionRetentionWork(tx, { companyId });
     const changed = await tx.update(chatConversations).set({ state: "waiting", updatedAt: new Date() })
       .where(and(eq(chatConversations.id, binding.conversation.id), eq(chatConversations.companyId, companyId), eq(chatConversations.state, "active")))
       .returning({ id: chatConversations.id });

server/src/services/slack-conversation-state.ts

@@ -1,3 +1,4 @@
+import { notifyAttentionRetentionWork } from "./retention-work-notifications.js";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
 import { notifyExternalConversationSettlement } from "./external-conversation-settlement-notifications.js";
 import { chatConversations, issues, type Db } from "@paperclipai/db";
@@ -45,6 +46,7 @@ export function nonIdleSlackIssueCondition() {
 
 /** Called inside message admission while its transaction owns the issue lock. */
 export async function resumeExternalConversation(tx: Db, companyId: string, issueId: string) {
+  await notifyAttentionRetentionWork(tx, { companyId });
   await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
   await notifyExternalConversationSettlement(tx, { companyId, issueId });
   // Only the verified idle convention may leave In Review automatically.