Merge qualified retention producer repair

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

* commit 'a0e2a3515d420270b30cd5e89fbb966ef641b8a9':
  fix(retention): wake blocker retention after no-run workspace export

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

@@ -56,3 +56,5 @@ The ordinary feed retains conservative `needs_attention` entries for roots outsi
 Explicit blocker queries apply the dependency owner's pending-finalization predicate before the graph row limit. A done blocker with no operation debt is excluded; attributed operations take precedence over legacy unattributed workspace operations, and a later successful export clears the barrier. The readiness helper and bounded attention query share this predicate, including the existing optional workspace/operation locks. Each relation resolves its globally unique issue identity before tenant and eligibility checks, avoiding repeated company-index scans.
 
 A fresh archive check consumes the private graph completeness result. An unexpanded frontier cannot establish absence of live work, so that check disarms without archiving. Supported issue, relation, run and workspace changes reactivate the root; restart reconstructs it. Completed-only history remains fully classified and can archive normally. No incomplete-root polling is added.
+
+The ordinary workspace-operation recorder also signals no-run finalization in its owning transaction. Export with a workspace identity invalidates that company's blocker source, including every root sharing the exported workspace; only a known issue without a workspace uses the existing reverse-issue hint. This rare source change uses bounded company/source paging and does not rebuild unrelated retention kinds. Run-backed finalization retains its existing native-status fanout. This lets a parked incomplete tree reactivate when export clears its done blockers, without an incomplete-root retry timer. Recorder rollback leaves the operation barrier and attention state unchanged.

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

@@ -6,6 +6,8 @@ import { subscribeDatabaseWork, authUsers, companyMemberships, agents, approvals
   decisionRetention, decisions, costEvents, heartbeatRunEvents, heartbeatRuns, issueApprovals, issueRecoveryActions,
   issueRelations, issueThreadInteractions, issues, type Db } from "@paperclipai/db";
 import { attentionRetentionService, registerAttentionRetentionWork } from "../services/attention-retention.js";
+import { workspaceOperationService } from "../services/workspace-operations.js";
+import * as workspaceOperationLogs from "../services/workspace-operation-log-store.js";
 import { budgetService } from "../services/budgets.js";
 import { writeExecutionControl } from "../services/execution-control-notifications.js";
 import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js";
@@ -514,6 +516,90 @@ const delay = DEFAULT_DECISION_ARCHIVE_DAYS * 86_400_000;
     await start(); await quiet();
   });
 
+  function simulateRecorderLogStorage() {
+    // The recorder lifecycle and SQL are real; only its log-storage boundary is simulated.
+    vi.spyOn(workspaceOperationLogs, "getWorkspaceOperationLogStore").mockReturnValue({
+      begin: async () => ({ store: "local_file", logRef: "retention-finalize-proof" }),
+      append: async () => {}, finalize: async () => ({ bytes: 0, compressed: false }), read: async () => ({ content: "" }),
+    });
+  }
+
+  it.each(["issue", "workspace"] as const)("publishes retention intent from ordinary no-run %s finalization", async scope => {
+    simulateRecorderLogStorage();
+    const source = await insertIssue(), workspaceId = randomUUID();
+    await db.execute(sql`insert into execution_workspaces (id, company_id, name, mode, strategy_type)
+      values (${workspaceId}::uuid, ${companyId}::uuid, 'Finalize producer', 'shared', 'project_primary')`);
+    await start(); await quiet();
+    const committedHints: string[] = [];
+    const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.attentionRetention, (event, hints) => {
+      if (event === "settled") committedHints.push(...hints ?? []);
+    });
+    try {
+      const result = await workspaceOperationService(db).createRecorder({ companyId,
+        executionWorkspaceId: scope === "workspace" ? workspaceId : undefined,
+        issueId: scope === "issue" ? source.id : undefined,
+      }).recordOperation({ phase: "workspace_finalize", run: async () => ({ status: "succeeded" }) });
+      expect(result.status).toBe("succeeded");
+      expect(committedHints.length).toBeGreaterThan(0);
+      expect(committedHints.map(hint => JSON.parse(hint).companyId).every(id => id === companyId)).toBe(true);
+      await quiet();
+      if (scope === "workspace") {
+        committedHints.length = 0;
+        await workspaceOperationService(db).createRecorder({ companyId })
+          .recordOperation({ phase: "workspace_finalize", run: async () => ({ status: "succeeded" }) });
+        expect(committedHints).toEqual([]); await quiet();
+      }
+    } finally { unsubscribe(); }
+  });
+
+  describe("no-run finalization with an incomplete blocker graph", () => {
+    let rootIds: string[], workspaceId: string, subjectId: string;
+    let loadedScopes: Array<Parameters<ReturnType<typeof attentionRetentionService>["load"]>[0]>;
+    beforeEach(async () => {
+      const began = performance.now();
+      simulateRecorderLogStorage();
+      const roots = await Promise.all([insertIssue("blocked"), insertIssue("blocked")]);
+      rootIds = roots.map(root => root.id); workspaceId = randomUUID();
+      await db.execute(sql`insert into execution_workspaces (id, company_id, name, mode, strategy_type)
+        values (${workspaceId}::uuid, ${companyId}::uuid, 'Pending shared export', 'shared', 'project_primary')`);
+      await workspaceOperationService(db).createRecorder({ companyId, executionWorkspaceId: workspaceId })
+        .recordOperation({ phase: "workspace_provision", run: async () => ({ status: "succeeded" }) });
+      await db.execute(sql`with history as (
+        insert into issues (company_id, title, status, updated_at, execution_workspace_id)
+        select ${companyId}::uuid, 'waiting-for-export-' || n, 'done', ${old().toISOString()}::timestamptz, ${workspaceId}::uuid
+        from generate_series(1, 2100) n returning id)
+        insert into issue_relations (company_id, issue_id, related_issue_id, type)
+        select ${companyId}::uuid, history.id, root.id, 'blocks' from history
+        cross join (values (${rootIds[0]}::uuid), (${rootIds[1]}::uuid)) root(id)`);
+      const [subject] = await db.select({ id: issues.id }).from(issues)
+        .where(and(eq(issues.companyId, companyId), eq(issues.status, "done"))).limit(1);
+      subjectId = subject!.id;
+      const review = await insertIssue("in_review", { assigneeUserId: "board" });
+      await decisionRetentionService(db).setKeep({ ...mutation("review", review.id), keep: true });
+      const service = attentionRetentionService(db), load = service.load.bind(service); loadedScopes = [];
+      vi.spyOn(service, "load").mockImplementation((scope, signal) => { loadedScopes.push(scope); return load(scope, signal); });
+      process.stdout.write(JSON.stringify({ incompleteFinalizePhase: "seeded", setupMs: performance.now() - began }) + "\n");
+      await start(service); await quiet();
+      for (const rootId of rootIds) expect(await state("blocker_attention", rootId)).toBeNull();
+      process.stdout.write(JSON.stringify({ incompleteFinalizeFixtureRows: 2100, incompleteRoots: 2, setupMs: performance.now() - began }) + "\n");
+    }, 30_000);
+    it("archives only after the ordinary workspace-only finalize commits, then settles quiet", async () => {
+      const finish = (database: Db) => workspaceOperationService(database).createRecorder({ companyId, executionWorkspaceId: workspaceId, issueId: subjectId })
+        .recordOperation({ phase: "workspace_finalize", run: async () => ({ status: "succeeded" }) });
+      await expect(db.transaction(async tx => {
+        await finish(tx as unknown as Db); throw new Error("rollback finalization");
+      })).rejects.toThrow("rollback finalization");
+      await settle(); for (const rootId of rootIds) expect(await state("blocker_attention", rootId)).toBeNull();
+      loadedScopes.length = 0;
+      expect((await finish(db)).status).toBe("succeeded");
+      for (const rootId of rootIds) await archived("blocker_attention", rootId);
+      await quiet();
+      expect(loadedScopes.length).toBeGreaterThan(0);
+      expect(loadedScopes.every(scope => scope?.companyId === companyId && scope.sourceKind === "blocker_attention")).toBe(true);
+      expect(await db.select().from(decisionArchiveNotificationOutbox)).toHaveLength(2);
+    });
+  });
+
   it("disarms finite deferred-root checks when large retained graphs have no archive obligation", async () => {
     await db.execute(sql`insert into issues (id, company_id, title, status, updated_at)
       select gen_random_uuid(), ${companyId}::uuid, 'held-root-' || n, 'blocked', ${old().toISOString()}::timestamptz

server/src/services/workspace-operations.ts

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