fix(retention): wake blocker retention after no-run workspace export

Register finalization intent before state writes, preserving caller transactions. Shared workspace export invalidates only the company blocker source; known issue-only export stays scoped and unattached operations do not activate retention. Add actual recorder controls and the shared-root integration journey.

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-10-31 02:00:39 -05:00
1 parent a53f22861f
commit a0e2a3515d
3 files changed
+96 -1

No files matched your search

@@ -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
+8 -1
View File
@@ -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,