From efbe3fa67d8dee2fb2037d9b097ed6ae1876a982 Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 18:25:01 -0500 Subject: [PATCH] fix(server): isolate blocker traversal budgets per root Share queried graph facts across roots while retaining independent node/depth limits, expanded reachability and truncation for classification. Unrelated limited graphs must not change terminal subjects or delay archive deadlines. Prove depth/node overflow and overlapping-root liveness red/green, preserve the100-root SQL bound, and qualify existing owner assertions on isolated PostgreSQL. Co-Authored-By: Paperclip --- .../2026-10-30-attention-retention-sleep.md | 2 +- .../attention-retention-work.test.ts | 57 ++++++++++++++++++- .../__tests__/issue-blocker-attention.test.ts | 12 +++- server/src/services/issues.ts | 56 +++++++++++------- 4 files changed, 101 insertions(+), 26 deletions(-) diff --git a/doc/plans/2026-10-30-attention-retention-sleep.md b/doc/plans/2026-10-30-attention-retention-sleep.md index 2e0373fc975..463fbe94188 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. 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. +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. The blocker owner shares fetched node and edge facts while tracking depth, node count, expanded nodes and live-work reachability independently for each root. An unrelated graph reaching its limit cannot change a shallow root’s terminal subject or deadline. 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 fd1a01e8582..29a82f866a5 100644 --- a/server/src/__tests__/attention-retention-work.test.ts +++ b/server/src/__tests__/attention-retention-work.test.ts @@ -15,7 +15,7 @@ import { decisionRetentionService, DEFAULT_DECISION_ARCHIVE_DAYS } from "../serv 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 { BLOCKER_ATTENTION_MAX_DEPTH, BLOCKER_ATTENTION_MAX_NODES, 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"; @@ -256,6 +256,61 @@ const delay = DEFAULT_DECISION_ARCHIVE_DAYS * 86_400_000; expect(refreshed.rows!.some(entry => entry.sourceId === terminals[99].id)).toBe(false); }); + it.each(["depth", "nodes"] as const)("preserves shallow terminal selection when an unrelated graph hits its %s limit", async limit => { + const root = await insertIssue("blocked", { updatedAt: new Date() }); + const middle = await insertIssue("todo", { updatedAt: new Date() }); + const terminal = await insertIssue("backlog", { assigneeAgentId: agentId }); + const heavy = await insertIssue("blocked", { updatedAt: new Date() }); + const count = limit === "depth" ? BLOCKER_ATTENTION_MAX_DEPTH + 1 : BLOCKER_ATTENTION_MAX_NODES; + const heavyNodes = await db.insert(issues).values(Array.from({ length: count }, () => ({ companyId, + title: randomUUID(), status: "todo", updatedAt: new Date() }))).returning(); + await db.insert(issueRelations).values([ + { companyId, issueId: middle.id, relatedIssueId: root.id, type: "blocks" }, + { companyId, issueId: terminal.id, relatedIssueId: middle.id, type: "blocks" }, + ...heavyNodes.map((node, index) => ({ companyId, issueId: node.id, + relatedIssueId: limit === "nodes" || index === 0 ? heavy.id : heavyNodes[index - 1].id, type: "blocks" })), + ]); + const owner = issueService(db); + const expected = (await owner.listBlockerAttention(companyId, [root])).get(root.id)!; + const limited = (await owner.listBlockerAttention(companyId, [heavy])).get(heavy.id)!; + expect(expected.terminalBlockerIssueId).toBe(terminal.id); + expect(limited.terminalBlockerIssueId).toBe(limited.directBlockerIssueId); + for (const input of [[root, heavy], [heavy, root]]) { + const batch = await owner.listBlockerAttention(companyId, input); + expect(batch.get(root.id)).toEqual(expected); expect(batch.get(heavy.id)).toEqual(limited); + } + const page = await attentionRetentionService(db).load({ companyId, sourceKind: "blocker_attention" }); + const entry = page.rows!.find(item => item.rootId === root.id && item.variant === "terminal")!; + expect(entry.sourceId).toBe(terminal.id); expect(entry.dueAt).toBeLessThan(Date.now()); + await attentionRetentionService(db).check(entry); + expect((await state("blocker_attention", terminal.id))?.archivedAt).toBeInstanceOf(Date); + expect(await db.select().from(decisionArchiveNotificationOutbox).where(eq(decisionArchiveNotificationOutbox.sourceId, terminal.id))).toHaveLength(1); + }); + + it("keeps root-specific reachability and cycle budgets when another root explores farther", async () => { + const root = await insertIssue("blocked"); + const chain = await db.insert(issues).values(Array.from({ length: BLOCKER_ATTENTION_MAX_DEPTH + 1 }, (_, index) => ({ + companyId, title: randomUUID(), status: index === BLOCKER_ATTENTION_MAX_DEPTH ? "in_progress" + : index === BLOCKER_ATTENTION_MAX_DEPTH - 2 ? "blocked" : "todo", updatedAt: old() }))).returning(); + const overlapping = chain[BLOCKER_ATTENTION_MAX_DEPTH - 2]; + const cycle = await insertIssue("blocked"); + const leaf = await insertIssue("backlog", { assigneeAgentId: agentId }); + await db.insert(issueRelations).values([ + ...chain.map((node, index) => ({ companyId, issueId: node.id, relatedIssueId: index ? chain[index - 1].id : root.id, type: "blocks" })), + { companyId, issueId: leaf.id, relatedIssueId: cycle.id, type: "blocks" }, + { companyId, issueId: cycle.id, relatedIssueId: leaf.id, type: "blocks" }, + ]); + const owner = issueService(db), roots = [root, overlapping, cycle]; + const singles = new Map(); + for (const item of roots) singles.set(item.id, (await owner.listBlockerAttention(companyId, [item])).get(item.id)); + expect(singles.get(root.id).blockingTreeLive).toBe(false); + expect(singles.get(overlapping.id).blockingTreeLive).toBe(true); + for (const input of [roots, [...roots].reverse()]) { + const batch = await owner.listBlockerAttention(companyId, input); + for (const item of roots) expect(batch.get(item.id)).toEqual(singles.get(item.id)); + } + }); + 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/__tests__/issue-blocker-attention.test.ts b/server/src/__tests__/issue-blocker-attention.test.ts index 1e6c00631e7..4f1a0e71573 100644 --- a/server/src/__tests__/issue-blocker-attention.test.ts +++ b/server/src/__tests__/issue-blocker-attention.test.ts @@ -8,6 +8,7 @@ import { approvals, companies, createDb, + applyPendingMigrations, heartbeatRuns, issueApprovals, issueRelations, @@ -21,7 +22,10 @@ import { import { issueService } from "../services/issues.js"; import { buildIssueGraphLivenessIncidentKey } from "../services/recovery/origins.js"; -const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; + +const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_BLOCKER_ATTENTION_TEST_DATABASE_URL"); +const embeddedPostgresSupport = externalUrl ? { supported: true } : await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; if (!embeddedPostgresSupport.supported) { @@ -36,8 +40,9 @@ describeEmbeddedPostgres("issue blocker attention", () => { let tempDb: Awaited> | null = null; beforeAll(async () => { - tempDb = await startEmbeddedPostgresTestDatabase("paperclip-issue-blocker-attention-"); - db = createDb(tempDb.connectionString); + if (externalUrl) await applyPendingMigrations(externalUrl); + else tempDb = await startEmbeddedPostgresTestDatabase("paperclip-issue-blocker-attention-"); + db = createDb(externalUrl ?? tempDb!.connectionString, { maxConnections: 6 }); svc = issueService(db); }, 20_000); @@ -55,6 +60,7 @@ describeEmbeddedPostgres("issue blocker attention", () => { }); afterAll(async () => { + await db?.$client.end({ timeout: 1 }); await tempDb?.cleanup(); }); diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index 7239075de99..fd1fd9ba819 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -3840,18 +3840,19 @@ async function listIssueBlockerAttentionMap( const edgesByIssueId = new Map(); for (const root of roots) nodesById.set(root.id, { ...root }); - let frontier = roots.map((root) => root.id); - let truncated = false; + type Traversal = { nodes: Set; expanded: Set; frontier: string[]; truncated: boolean }; + const traversals = new Map(roots.map(root => [root.id, { + nodes: new Set([root.id]), expanded: new Set(), frontier: [root.id], truncated: false, + }])); + const expanded = new Set(); const pendingFinalizeBlockerIssueIds = new Set(); - for ( - let depth = 0; - frontier.length > 0 && depth < BLOCKER_ATTENTION_MAX_DEPTH; - depth += 1 - ) { - const nextFrontier = new Set(); + for (let depth = 0; depth < BLOCKER_ATTENTION_MAX_DEPTH; depth += 1) { + const active = [...traversals.values()].filter(traversal => !traversal.truncated && traversal.frontier.length > 0); + if (active.length === 0) break; + const frontier = [...new Set(active.flatMap(traversal => traversal.frontier))].filter(id => !expanded.has(id)); for (const chunk of chunkList( - [...new Set(frontier)], + frontier, ISSUE_LIST_RELATED_QUERY_CHUNK_SIZE, )) { const readinessByIssueId = await listIssueDependencyReadinessMap( @@ -3959,17 +3960,27 @@ async function listIssueBlockerAttentionMap( assigneeAgentId: row.assigneeAgentId, assigneeUserId: row.assigneeUserId, }); - nextFrontier.add(row.blockerIssueId); } + for (const id of chunk) expanded.add(id); } - if (nodesById.size > BLOCKER_ATTENTION_MAX_NODES) { - truncated = true; - break; + for (const traversal of active) { + const nextFrontier = new Set(); + for (const id of traversal.frontier) { + traversal.expanded.add(id); + for (const edge of edgesByIssueId.get(id) ?? []) { + if (traversal.nodes.has(edge.blockerIssueId) || !nodesById.has(edge.blockerIssueId)) continue; + traversal.nodes.add(edge.blockerIssueId); + nextFrontier.add(edge.blockerIssueId); + } + } + traversal.truncated = traversal.nodes.size > BLOCKER_ATTENTION_MAX_NODES; + traversal.frontier = traversal.truncated ? [] : [...nextFrontier]; } - frontier = [...nextFrontier]; } - if (frontier.length > 0) truncated = true; + for (const traversal of traversals.values()) { + if (traversal.frontier.length > 0) traversal.truncated = true; + } const nodeIds = [...nodesById.keys()]; const activeIssueIds = new Set(); @@ -4215,6 +4226,7 @@ async function listIssueBlockerAttentionMap( const classifyPath = ( nodeId: string, seen: Set, + truncated: boolean, ): PathClassification => { const sample = blockerSampleIdentifier(nodesById.get(nodeId)); if (truncated || seen.has(nodeId)) { @@ -4319,7 +4331,7 @@ async function listIssueBlockerAttentionMap( const nextSeen = new Set(seen); nextSeen.add(nodeId); const classified = downstream.map((edge) => - classifyPath(edge.blockerIssueId, nextSeen), + classifyPath(edge.blockerIssueId, nextSeen, truncated), ); const stalledChild = classified.find( (result) => result.stalled || result.sampleStalledBlockerIdentifier, @@ -4385,8 +4397,8 @@ async function listIssueBlockerAttentionMap( }; }; - const pathHasLiveWork = (nodeId: string, seen: Set): boolean => { - if (seen.has(nodeId)) return false; + const pathHasLiveWork = (nodeId: string, seen: Set, traversal: Traversal): boolean => { + if (seen.has(nodeId) || !traversal.nodes.has(nodeId)) return false; const node = nodesById.get(nodeId); if (!node || node.companyId !== companyId) return false; if ( @@ -4397,6 +4409,7 @@ async function listIssueBlockerAttentionMap( ) return true; + if (!traversal.expanded.has(nodeId)) return false; const nextSeen = new Set(seen); nextSeen.add(nodeId); return (edgesByIssueId.get(node.id) ?? []).some((edge) => { @@ -4406,7 +4419,7 @@ async function listIssueBlockerAttentionMap( !pendingFinalizeBlockerIssueIds.has(edge.blockerIssueId) ) return false; - return pathHasLiveWork(edge.blockerIssueId, nextSeen); + return pathHasLiveWork(edge.blockerIssueId, nextSeen, traversal); }); }; @@ -4419,6 +4432,7 @@ async function listIssueBlockerAttentionMap( }; for (const root of roots) { + const traversal = traversals.get(root.id)!; const topLevelEdges = (edgesByIssueId.get(root.id) ?? []).filter((edge) => { const blocker = nodesById.get(edge.blockerIssueId); return ( @@ -4440,7 +4454,7 @@ async function listIssueBlockerAttentionMap( const classified = topLevelEdges.map((edge) => ({ edge, - result: classifyPath(edge.blockerIssueId, new Set([root.id])), + result: classifyPath(edge.blockerIssueId, new Set([root.id]), traversal.truncated), })); const coveredBlockerCount = classified.filter( (entry) => entry.result.covered, @@ -4523,7 +4537,7 @@ async function listIssueBlockerAttentionMap( sampleStalledFromChain ?? null, blockingTreeLive: topLevelEdges.some((edge) => - pathHasLiveWork(edge.blockerIssueId, new Set([root.id])), + pathHasLiveWork(edge.blockerIssueId, new Set([root.id]), traversal), ), directBlockerIssueId: sampleEntry?.edge.blockerIssueId ?? null, terminalBlockerIssueId,