From 097ac579da3a19154d28ba980906d3a23080363e Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 17:59:04 -0500 Subject: [PATCH] perf(server): batch remaining attention retention page evidence Load company-partitioned decision/recovery links, run/agent visibility and newer-run probes once per page; share one blocker graph reconstruction and exact SQL timestamps across page entries. Keep fresh per-entry archive locks and revalidation. Add full100-source query budgets and late visibility/supersession safety controls. Co-Authored-By: Paperclip --- .../2026-10-30-attention-retention-sleep.md | 2 +- .../attention-retention-work.test.ts | 85 +++++++++++++++ server/src/services/attention-retention.ts | 103 ++++++++++++++---- 3 files changed, 169 insertions(+), 21 deletions(-) diff --git a/doc/plans/2026-10-30-attention-retention-sleep.md b/doc/plans/2026-10-30-attention-retention-sleep.md index 1b71dc71261..2e0373fc975 100644 --- a/doc/plans/2026-10-30-attention-retention-sleep.md +++ b/doc/plans/2026-10-30-attention-retention-sleep.md @@ -25,7 +25,7 @@ The feed and worker share confirmation collapse, newer-run probes, issue visibil ## Existing mutation boundaries reactivate work -Committed hints identify a company, source kind, and source ID where those values are known. Retention's own writes use exact source hints; blocker retention changes invalidate the company/source because multiple blocked roots can project one terminal subject. Reconstruction pages and due batches remain bounded. Review attention and interaction invokability are computed in company batches per candidate page; a full 100-candidate review page uses at most 20 statements and 100 confirmations use at most six. Ordinary execution, decision and budget mutations pass their affected company; exact decision sources are included. +Committed hints identify a company, source kind, and source ID where those values are known. Retention's own writes use exact source hints; blocker retention changes invalidate the company/source because multiple blocked roots can project one terminal subject. Reconstruction pages and due batches remain bounded. Review attention and interaction invokability are computed in company batches per candidate page; a full 100-candidate review page uses at most 20 statements and 100 confirmations use at most six. Decisions, recovery source links, runs, eligible agents, issue visibility, newer-run probes and blocker graphs also load in company batches per page. A 100-source decision or recovery page uses at most three statements; 100 failed runs use at most six; 100 blocker roots use at most 30 graph/source statements plus retention-state reads for their emitted entries. These evidence maps exist only for one load. Archive checks still lock and freshly re-read sources, terminal subjects and related eligibility. Ordinary execution, decision and budget mutations pass their affected company; exact decision sources are included. Producer coverage includes decision creation and changes; interaction creation, resolution, expiry and supersession; issue creation, updates, blocker relations, checkout, adoption and release; agent hire, configuration, lifecycle, error clearing and deletion; budget incidents and company pause/resume; exhausted-run events and retry-run creation; native status, review, finalization and reconstruction; external conversation transitions; and imported blocker relations. Existing service transactions retain outer commit ownership. Savepoint release does not publish work before outer commit. diff --git a/server/src/__tests__/attention-retention-work.test.ts b/server/src/__tests__/attention-retention-work.test.ts index 889030d2c22..fd1a01e8582 100644 --- a/server/src/__tests__/attention-retention-work.test.ts +++ b/server/src/__tests__/attention-retention-work.test.ts @@ -171,6 +171,91 @@ const delay = DEFAULT_DECISION_ARCHIVE_DAYS * 86_400_000; expect(statements.filter(query => query.includes('from "agents"'))).toHaveLength(1); }); + it("batches decision and recovery source links and rechecks a late hidden origin", async () => { + const origins = await db.insert(issues).values(Array.from({ length: 100 }, () => ({ companyId, + title: randomUUID(), status: "todo", updatedAt: old() }))).returning(); + const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId, status: "succeeded" }).returning(); + const decisionRows = await db.insert(decisions).values(origins.map(origin => ({ 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() }))).returning(); + const recoveryRows = await db.insert(issueRecoveryActions).values(origins.map(origin => ({ companyId, + sourceIssueId: origin.id, kind: "execution", ownerType: "board", cause: "fixture", fingerprint: "fixture", + nextAction: "Choose", updatedAt: old() }))).returning(); + const service = attentionRetentionService(db); + const pages = []; + for (const sourceKind of ["decision", "recovery_action"] as const) { + statements.length = 0; + const page = await service.load({ companyId, sourceKind }); + expect(page.rows).toHaveLength(100); expect(page.complete).toBe(false); + expect(statements.length).toBeLessThanOrEqual(3); + pages.push(page.rows!); + } + await db.update(issues).set({ hiddenAt: new Date() }).where(eq(issues.id, origins[99].id)); + const hiddenDecision = pages[0].find(entry => entry.sourceId === decisionRows[99].id)!; + const hiddenRecovery = pages[1].find(entry => entry.sourceId === recoveryRows[99].id)!; + expect(await service.check(hiddenDecision)).toBeNull(); expect(await service.check(hiddenRecovery)).toBeNull(); + expect(await state("decision", hiddenDecision.sourceId)).toBeNull(); expect(await state("recovery_action", hiddenRecovery.sourceId)).toBeNull(); + expect(await db.select().from(decisionArchiveNotificationOutbox)).toHaveLength(0); + }); + + it("batches 100 failed runs and rechecks late visibility and supersession before archive", async () => { + const origins = await db.insert(issues).values(Array.from({ length: 100 }, () => ({ companyId, + title: randomUUID(), status: "todo", updatedAt: old() }))).returning(); + const runs = await db.insert(heartbeatRuns).values(origins.map(origin => ({ companyId, agentId, + status: "failed", contextSnapshot: { issueId: origin.id }, createdAt: old(), finishedAt: old() }))).returning(); + await db.insert(heartbeatRunEvents).values(runs.map(run => ({ companyId, agentId, runId: run.id, + seq: 1, eventType: "lifecycle", message: "Bounded retry exhausted: batch fixture" }))); + const service = attentionRetentionService(db), scope = { companyId, sourceKind: "failed_run" as const }; + statements.length = 0; + const page = await service.load(scope); + expect(page.rows).toHaveLength(100); expect(page.complete).toBe(false); + expect(statements.length).toBeLessThanOrEqual(6); + const hiddenRun = runs[98], supersededRun = runs[99]; + const hidden = page.rows!.find(entry => entry.sourceId === hiddenRun.id)!; + const superseded = page.rows!.find(entry => entry.sourceId === supersededRun.id)!; + await db.update(issues).set({ hiddenAt: new Date() }).where(eq(issues.id, origins[98].id)); + await db.insert(heartbeatRuns).values({ companyId, agentId, status: "succeeded", + contextSnapshot: { issueId: origins[99].id }, createdAt: new Date(old().getTime() + 60_000), finishedAt: old() }); + expect(await service.check(hidden)).toBeNull(); expect(await service.check(superseded)).toBeNull(); + expect(await state("failed_run", hiddenRun.id)).toBeNull(); expect(await state("failed_run", supersededRun.id)).toBeNull(); + expect(await db.select().from(decisionArchiveNotificationOutbox)).toHaveLength(0); + expect((await service.load(scope)).complete).toBe(true); + const refreshed = await service.load(scope); + expect(refreshed.rows).toHaveLength(98); + expect(refreshed.rows!.some(entry => [hiddenRun.id, supersededRun.id].includes(entry.sourceId))).toBe(false); + }); + + it("batches 100 blocker graphs with exact terminal timestamps and fresh terminal visibility", async () => { + const roots = await db.insert(issues).values(Array.from({ length: 100 }, () => ({ companyId, + title: randomUUID(), status: "blocked", updatedAt: old(), blockedTransitionAt: old(), + unblockDescriptor: { owner: "board", action: "Pick an owner" } }))).returning(); + const terminals = await db.insert(issues).values(Array.from({ length: 100 }, () => ({ companyId, + title: randomUUID(), status: "backlog", assigneeAgentId: agentId, updatedAt: old() }))).returning(); + await db.insert(issueRelations).values(roots.map((root, index) => ({ companyId, + issueId: terminals[index].id, relatedIssueId: root.id, type: "blocks" }))); + await db.execute(sql`update issues set updated_at = updated_at + interval '0.000321 seconds' where id = ${terminals[99].id}::uuid`); + const service = attentionRetentionService(db), scope = { companyId, sourceKind: "blocker_attention" as const }; + statements.length = 0; + const page = await service.load(scope); + const queries = [...statements]; + expect(page.rows).toHaveLength(200); expect(page.complete).toBe(false); + const retentionReads = queries.filter(query => query.includes('from "decision_retention"')); + expect(retentionReads).toHaveLength(200); + expect(queries.length - retentionReads.length).toBeLessThanOrEqual(30); + const terminal = page.rows!.find(entry => entry.rootId === roots[99].id && entry.variant === "terminal")!; + expect(terminal.sourceId).toBe(terminals[99].id); + const [expected] = await db.execute(sql`select updated_at::text as at, + ceil(extract(epoch from updated_at) * 1000 + ${delay}) as due from issues where id = ${terminals[99].id}::uuid`); + expect(terminal.activityAt).toBe(expected.at); expect(terminal.dueAt).toBe(Number(expected.due)); + await db.update(issues).set({ hiddenAt: new Date() }).where(eq(issues.id, terminals[99].id)); + expect(await service.check(terminal)).toBeNull(); + expect(await state("blocker_attention", terminals[99].id)).toBeNull(); + expect(await db.select().from(decisionArchiveNotificationOutbox)).toHaveLength(0); + expect((await service.load(scope)).complete).toBe(true); + const refreshed = await service.load(scope); + expect(refreshed.rows!.some(entry => entry.sourceId === terminals[99].id)).toBe(false); + }); + it("scopes ordinary execution-control writes to their company", async () => { const otherCompany = randomUUID(); await db.insert(companies).values({ id: otherCompany, name: "Other", issuePrefix: "OTHER" }); diff --git a/server/src/services/attention-retention.ts b/server/src/services/attention-retention.ts index 85e27a0d941..c72947e5349 100644 --- a/server/src/services/attention-retention.ts +++ b/server/src/services/attention-retention.ts @@ -1,4 +1,4 @@ -import { and, eq, inArray, isNull, sql, type SQL } from "drizzle-orm"; +import { and, eq, getTableColumns, inArray, isNull, sql, type SQL } from "drizzle-orm"; import { agents, decisionRetention, decisions, heartbeatRuns, issueApprovals, issueRecoveryActions, issueThreadInteractions, issues, approvals, type Db } from "@paperclipai/db"; import type { AttentionSourceKind } from "@paperclipai/shared"; @@ -23,6 +23,18 @@ type Candidate = { sourceKind: Kind; sourceId: string; companyId: string; activi type Entry = Candidate & { id: string; dueAt: number; rootId: string; variant?: "owner" | "terminal" }; type Cursor = { sourceKind: string; sourceId: string }; +type PageIssue = typeof issues.$inferSelect & { activityAt: string; deadlineAt: number; + blockedActivityAt: string | null; blockedDeadlineAt: number | null }; +type PageEvidence = { + decisions: Map; + recoveries: Map; + runs: Map; + agents: Set; + visibleIssues: Map; + newerRuns: Set; + blockers: Awaited["listBlockerAttention"]>>; +}; + const sourceTables: Record = { decision: sql`decisions`, issue_thread_interaction: sql`issue_thread_interactions`, recovery_action: sql`issue_recovery_actions`, blocker_attention: sql`issues`, review: sql`issues`, @@ -137,56 +149,106 @@ export function attentionRetentionService(db: Db) { return result; } - async function eligible(database: Db, candidate: Candidate, variant?: "owner" | "terminal", lock = false, reviewAttention?: Map, interactionAttention?: Map): Promise { + async function pageEvidence(database: Db, candidates: Candidate[]) { + const byCompany = new Map(); + for (const candidate of candidates) { + if (!["decision", "recovery_action", "failed_run", "blocker_attention"].includes(candidate.sourceKind)) continue; + const rows = byCompany.get(candidate.companyId) ?? []; + rows.push(candidate); + byCompany.set(candidate.companyId, rows); + } + const result = new Map(); + for (const [companyId, rows] of byCompany) { + const ids = (kind: Kind) => rows.filter(row => row.sourceKind === kind).map(row => row.sourceId); + const decisionIds = ids("decision"), recoveryIds = ids("recovery_action"), runIds = ids("failed_run"), blockerIds = ids("blocker_attention"); + const decisionRows = decisionIds.length ? await database.select({ id: decisions.id, originIssueId: decisions.originIssueId }) + .from(decisions).where(and(eq(decisions.companyId, companyId), inArray(decisions.id, decisionIds), eq(decisions.status, "open"))) : []; + const recoveryRows = recoveryIds.length ? await database.select().from(issueRecoveryActions).where(and( + eq(issueRecoveryActions.companyId, companyId), inArray(issueRecoveryActions.id, recoveryIds))) : []; + const runRows = runIds.length ? await database.select().from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, companyId), inArray(heartbeatRuns.id, runIds))) : []; + const runAgents = runRows.length ? await database.select({ id: agents.id }).from(agents).where(and( + eq(agents.companyId, companyId), inArray(agents.id, [...new Set(runRows.map(row => row.agentId))]), + sql`${agents.status} <> 'terminated'`)) : []; + const readIssues = async (issueIds: string[]) => issueIds.length ? database.select({ ...getTableColumns(issues), + activityAt: sql`${issues.updatedAt}::text`, + deadlineAt: sql`ceil(extract(epoch from ${issues.updatedAt}) * 1000 + ${archiveDelay})`.mapWith(Number), + blockedActivityAt: sql`${issues.blockedTransitionAt}::text`, + blockedDeadlineAt: sql`ceil(extract(epoch from ${issues.blockedTransitionAt}) * 1000 + ${archiveDelay})`.mapWith(Number), + }).from(issues).where(and(eq(issues.companyId, companyId), inArray(issues.id, [...new Set(issueIds)]), + attentionIssueCondition(false))) : []; + const issueIds = [...blockerIds, ...decisionRows.map(row => row.originIssueId), ...recoveryRows.map(row => row.sourceIssueId), + ...runRows.map(row => readRunIssueId(row.contextSnapshot))].filter((id): id is string => Boolean(id)); + const visibleRows = await readIssues(issueIds); + const visible = new Map(visibleRows.map(row => [row.id, row])); + const blockerRoots = visibleRows.filter(row => blockerIds.includes(row.id)); + const blockers: PageEvidence["blockers"] = blockerRoots.length ? await issueService(database).listBlockerAttention(companyId, blockerRoots) : new Map(); + const terminalIds = [...blockers.values()].filter(terminalBlockerRequiresAttention) + .map(attention => attention.terminalBlockerIssueId).filter((id): id is string => id != null && !visible.has(id)); + for (const row of await readIssues(terminalIds)) visible.set(row.id, row); + result.set(companyId, { + decisions: new Map(decisionRows.map(row => [row.id, row])), recoveries: new Map(recoveryRows.map(row => [row.id, row])), + runs: new Map(runRows.map(row => [row.id, row])), agents: new Set(runAgents.map(row => row.id)), visibleIssues: visible, + newerRuns: await listFailedRunsWithNewerRun(database, companyId, + runRows.map(run => ({ ...run, issueId: readRunIssueId(run.contextSnapshot) }))), blockers, + }); + } + return result; + } + + async function eligible(database: Db, candidate: Candidate, variant?: "owner" | "terminal", lock = false, reviewAttention?: Map, interactionAttention?: Map, evidence?: PageEvidence): Promise { 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; + const row = evidence ? evidence.decisions.get(sourceId) : (await database.select({ originIssueId: decisions.originIssueId }).from(decisions) + .where(and(eq(decisions.companyId, candidate.companyId), eq(decisions.id, sourceId), eq(decisions.status, "open"))))[0]; + if (!row || row.originIssueId && !(evidence ? evidence.visibleIssues.has(row.originIssueId) + : await visibleIssue(database, candidate.companyId, row.originIssueId))) return null; } else if (candidate.sourceKind === "issue_thread_interaction") { const attention = interactionAttention ?? await interactionEligibility(database, [candidate]); if (!attention.get(sourceId)) return null; } else if (candidate.sourceKind === "recovery_action") { - const [row] = await database.select().from(issueRecoveryActions).where(and(eq(issueRecoveryActions.companyId, candidate.companyId), eq(issueRecoveryActions.id, sourceId))); - if (!row || !await visibleIssue(database, candidate.companyId, row.sourceIssueId)) return null; + const row = evidence ? evidence.recoveries.get(sourceId) : (await database.select().from(issueRecoveryActions).where(and(eq(issueRecoveryActions.companyId, candidate.companyId), eq(issueRecoveryActions.id, sourceId))))[0]; + if (!row || !(evidence ? evidence.visibleIssues.has(row.sourceIssueId) + : await visibleIssue(database, candidate.companyId, row.sourceIssueId))) return null; } else if (candidate.sourceKind === "review") { const attention = reviewAttention ?? await reviewEligibility(database, [candidate]); if (!attention.get(sourceId)) return null; } else if (candidate.sourceKind === "blocker_attention") { - const row = await visibleIssue(database, candidate.companyId, sourceId); + const row = evidence ? evidence.visibleIssues.get(sourceId) : 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, + const activity = evidence ? { at: evidence.visibleIssues.get(sourceId)!.blockedActivityAt!, due: evidence.visibleIssues.get(sourceId)!.blockedDeadlineAt! } : (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 }>; + where company_id = ${candidate.companyId}::uuid and id = ${row.id}::uuid`) as unknown as Array<{ at: string; due: string }>)[0]; activityAt = activity.at; dueAt = Number(activity.due); } else { - const attention = (await issueService(database).listBlockerAttention(candidate.companyId, [row])).get(row.id); + const attention = evidence ? evidence.blockers.get(row.id) : (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); + const terminal = evidence ? evidence.visibleIssues.get(sourceId) : await visibleIssue(database, candidate.companyId, sourceId); if (!terminal) return null; - const [activity] = await database.execute(sql`select updated_at::text as at, + const activity = evidence ? { at: evidence.visibleIssues.get(sourceId)!.activityAt, due: evidence.visibleIssues.get(sourceId)!.deadlineAt } : (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 }>; + where company_id = ${candidate.companyId}::uuid and id = ${sourceId}::uuid`) as unknown as Array<{ at: string; due: string }>)[0]; 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))); + const run = evidence ? evidence.runs.get(sourceId) : (await database.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, candidate.companyId), eq(heartbeatRuns.id, sourceId))))[0]; 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 agent = evidence ? evidence.agents.has(run.agentId) : (await database.select({ id: agents.id }).from(agents).where(and(eq(agents.companyId, candidate.companyId), + eq(agents.id, run.agentId), sql`${agents.status} <> 'terminated'`)))[0]; 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; + if (!agent || issueId && !(evidence ? evidence.visibleIssues.has(issueId) + : await visibleIssue(database, candidate.companyId, issueId))) return null; + if ((evidence ? evidence.newerRuns : 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 }; @@ -198,10 +260,11 @@ export function attentionRetentionService(db: Db) { const entries: Entry[] = []; const reviews = await reviewEligibility(db, rows); const interactions = await interactionEligibility(db, rows); + const evidence = await pageEvidence(db, rows); for (const row of rows) { if (signal?.aborted) return { rows: null, complete: true }; for (const variant of row.sourceKind === "blocker_attention" ? ["owner", "terminal"] as const : [undefined]) { - const entry = await eligible(db, row, variant, false, reviews, interactions); + const entry = await eligible(db, row, variant, false, reviews, interactions, evidence.get(row.companyId)); if (entry) entries.push(entry); } }