From 8911cca5d0b43a9f8e8adddc8677b333b3f3d2bb Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 19:49:06 -0500 Subject: [PATCH] fix(server): retain live evidence beyond blocker graph admission Keep bounded returned child facts independent of graph node and edge admission, and use the existing live-work predicate for running status, active runs and pending waits. Cover the opposite query ordering with 1,999 explicit blockers, fresh retention checks, no archive/outbox and unchanged SQL/resource bounds. Co-Authored-By: Paperclip --- .../2026-10-30-attention-retention-sleep.md | 2 +- .../attention-retention-work.test.ts | 31 +++++++++++++++++ server/src/services/issues.ts | 34 +++++++++++++------ 3 files changed, 55 insertions(+), 12 deletions(-) diff --git a/doc/plans/2026-10-30-attention-retention-sleep.md b/doc/plans/2026-10-30-attention-retention-sleep.md index aa78f1bb746..2be011ccf9d 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, source ID or changed issue where those values are known. Routine issue writers pass identity to the retention domain helper before their mutation. The helper captures old and new parent paths, reverse blocker ancestors, run-linked issues and liveness-origin references; issue-scoped source queries then select reviews, blockers, interactions, decisions, recovery actions and failed runs by their issue identity. Budget incidents and agent errors remain tied to their own source changes. Ordinary issue writes do not reconstruct the unrelated company feed. Opaque liveness and stranded-recovery origin references resolve through company-scoped issue rows before entering UUID queries, so valid none sentinels do not roll back writes and moved recovery tasks still invalidate their original review. Queue keys explicitly pick supported identity fields. 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. The shared graph fetch stores at most 2,000 nodes and 4,000 edges, and bounds each frontier query before materializing rows. An internal batch result identifies roots whose evaluation is incomplete; the legacy map API exposes no extra transport fields. Retention schedules only those roots for an immediate fresh single-root check, which disarms or saves the actual source deadline. It never archives a partial intermediary or adds a recurring incomplete-root poll. Fully evaluated roots preserve their exact classification. All later graph queries, including recovery source IDs, agent IDs and workspace-finalize inputs, use bounded chunks. A SQL-generated 70,000-node graph proves PostgreSQL parameter limits are respected at the driver boundary. 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, source ID or changed issue where those values are known. Routine issue writers pass identity to the retention domain helper before their mutation. The helper captures old and new parent paths, reverse blocker ancestors, run-linked issues and liveness-origin references; issue-scoped source queries then select reviews, blockers, interactions, decisions, recovery actions and failed runs by their issue identity. Budget incidents and agent errors remain tied to their own source changes. Ordinary issue writes do not reconstruct the unrelated company feed. Opaque liveness and stranded-recovery origin references resolve through company-scoped issue rows before entering UUID queries, so valid none sentinels do not roll back writes and moved recovery tasks still invalidate their original review. Queue keys explicitly pick supported identity fields. 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. The shared graph fetch stores at most 2,000 nodes and 4,000 edges, and bounds each frontier query before materializing rows. Positive live-work evidence from the bounded returned rows is retained independently of node/edge admission, including running status, active runs and pending confirmations. Filling the graph with explicit blockers cannot hide a subsequently returned active child. An internal batch result identifies roots whose evaluation is incomplete; the legacy map API exposes no extra transport fields. Retention schedules only those roots for an immediate fresh single-root check, which disarms or saves the actual source deadline. It never archives a partial intermediary or adds a recurring incomplete-root poll. Fully evaluated roots preserve their exact classification. All later graph queries, including recovery source IDs, agent IDs and workspace-finalize inputs, use bounded chunks. A SQL-generated 70,000-node graph proves PostgreSQL parameter limits are respected at the driver boundary. 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 275b4f64570..fb2dca6f6b5 100644 --- a/server/src/__tests__/attention-retention-work.test.ts +++ b/server/src/__tests__/attention-retention-work.test.ts @@ -377,6 +377,37 @@ const delay = DEFAULT_DECISION_ARCHIVE_DAYS * 86_400_000; expect(await db.select().from(decisionArchiveNotificationOutbox)).toHaveLength(0); }); + it.each(["in_progress", "run", "confirmation"])("preserves observed child live evidence (%s) after explicit blockers fill the node budget", async (kind) => { + const root = await insertIssue("blocked"); + await db.execute(sql`with blockers as ( + insert into issues (id, company_id, title, status, assignee_agent_id, updated_at) + select gen_random_uuid(), ${companyId}::uuid, 'explicit-leaf-' || n, 'backlog', ${agentId}::uuid, ${old().toISOString()}::timestamptz + from generate_series(1, 1999) n returning id) + insert into issue_relations (company_id, issue_id, related_issue_id, type) + select ${companyId}::uuid, id, ${root.id}::uuid, 'blocks' from blockers`); + const child = await insertIssue(kind === "in_progress" ? "in_progress" : "todo", { parentId: root.id, assigneeAgentId: agentId }); + if (kind === "run") { + const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId, status: "running" }).returning(); + await db.update(issues).set({ executionRunId: run.id }).where(eq(issues.id, child.id)); + } else if (kind === "confirmation") { + await db.insert(issueThreadInteractions).values({ companyId, issueId: child.id, kind: "request_confirmation", status: "pending", payload: { prompt: "Approve?" } }); + } + parameterCounts.length = 0; + const batch = await issueService(db).listBlockerAttentionBatch(companyId, [root]); + expect(batch.nodeCount).toBeLessThanOrEqual(BLOCKER_ATTENTION_MAX_NODES); + expect(batch.edgeCount).toBeLessThanOrEqual(BLOCKER_ATTENTION_MAX_NODES * 2); + expect(batch.fetchedRows).toBeLessThanOrEqual(BLOCKER_ATTENTION_MAX_NODES * 6); + expect(parameterCounts.length).toBeLessThanOrEqual(100); + expect(Math.max(...parameterCounts)).toBeLessThanOrEqual(1100); + expect(batch.deferredRootIds.has(root.id)).toBe(true); + expect(batch.attention.get(root.id)?.blockingTreeLive).toBe(true); + const service = attentionRetentionService(db), page = await service.load({ companyId, sourceKind: "blocker_attention" }); + const entry = page.rows!.find(item => item.rootId === root.id && item.variant === "terminal")!; + expect(await service.check(entry)).toBeNull(); + expect(await db.select().from(decisionRetention)).toHaveLength(0); + expect(await db.select().from(decisionArchiveNotificationOutbox)).toHaveLength(0); + }); + 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 diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index 4d5fd5f85f6..b8f78009a59 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -3850,6 +3850,10 @@ async function readIssueBlockerAttentionBatch( const nodesById = new Map(); const edgesByIssueId = new Map(); + // Returned rows are already bounded by the residual fetch budget. Keep their + // positive live-work evidence even when they cannot enter the traversal graph. + const observedNodes = new Map(); + const observedChildrenByParent = new Map>(); const admittedRoots = roots.slice(0, BLOCKER_ATTENTION_MAX_NODES); for (const root of admittedRoots) nodesById.set(root.id, { ...root }); @@ -3946,6 +3950,10 @@ async function readIssueBlockerAttentionBatch( ); for (const row of [...unresolvedExplicitBlockerRows, ...childRows]) { if (!row.issueId) continue; + observedNodes.set(row.blockerIssueId, row); + const observed = observedChildrenByParent.get(row.issueId) ?? new Set(); + observed.add(row.blockerIssueId); + observedChildrenByParent.set(row.issueId, observed); if (!nodesById.has(row.blockerIssueId)) { if (nodesById.size >= BLOCKER_ATTENTION_MAX_NODES) { complete = false; continue; } nodesById.set(row.blockerIssueId, { @@ -3993,11 +4001,12 @@ async function readIssueBlockerAttentionBatch( if (traversal.frontier.length > 0) traversal.truncated = true; } - const nodeIds = [...nodesById.keys()]; + const evidenceNodes = new Map([...nodesById, ...observedNodes]); + const nodeIds = [...evidenceNodes.keys()]; const activeIssueIds = new Set(); const agentIds = new Set(); const issueIdByExecutionRunId = new Map(); - for (const node of nodesById.values()) { + for (const node of evidenceNodes.values()) { if (node.assigneeAgentId) agentIds.add(node.assigneeAgentId); if (node.executionRunId) issueIdByExecutionRunId.set(node.executionRunId, node.id); @@ -4055,7 +4064,7 @@ async function readIssueBlockerAttentionBatch( } } - const explicitWaitCandidateIds = [...nodesById.values()] + const explicitWaitCandidateIds = [...evidenceNodes.values()] .filter((node) => node.status !== "done") .map((node) => node.id); const explicitWaitingIssueIds = new Set(); @@ -4400,19 +4409,22 @@ async function readIssueBlockerAttentionBatch( }; }; + const nodeHasLiveWork = (node: IssueBlockerAttentionNode) => + node.status === "in_progress" || activeIssueIds.has(node.id) || + explicitWaitingIssueIds.has(node.id) || Boolean(node.assigneeUserId); + const observedChildHasLiveWork = (parentId: string) => + [...observedChildrenByParent.get(parentId) ?? []].some(id => { + const node = observedNodes.get(id); + return node?.companyId === companyId && nodeHasLiveWork(node); + }); 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 ( - node.status === "in_progress" || - activeIssueIds.has(node.id) || - explicitWaitingIssueIds.has(node.id) || - Boolean(node.assigneeUserId) - ) - return true; + if (nodeHasLiveWork(node)) return true; if (!traversal.expanded.has(nodeId)) return false; + if (observedChildHasLiveWork(nodeId)) return true; const nextSeen = new Set(seen); nextSeen.add(nodeId); return (edgesByIssueId.get(node.id) ?? []).some((edge) => { @@ -4539,7 +4551,7 @@ async function readIssueBlockerAttentionBatch( stalledEntry?.result.sampleStalledBlockerIdentifier ?? sampleStalledFromChain ?? null, - blockingTreeLive: topLevelEdges.some((edge) => + blockingTreeLive: observedChildHasLiveWork(root.id) || topLevelEdges.some((edge) => pathHasLiveWork(edge.blockerIssueId, new Set([root.id]), traversal), ), directBlockerIssueId: sampleEntry?.edge.blockerIssueId ?? null,