From e41bde214638004b51752091938834691eb1c616 Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 19:01:12 -0500 Subject: [PATCH] fix(server): bound graph queries and scope issue retention work Chunk graph recovery, agent and workspace inputs; prove a70000-node PostgreSQL graph stays below parameter limits. Route ordinary issue writes through identity-scoped dependency invalidation for old/new parents, blocker ancestors, run links and liveness origins. Correlate all issue-related attention sources, preserve unrelated deadlines, and canonicalize notification keys. Exercise actual held/burst producers, visibility, relation and incident transitions. Co-Authored-By: Paperclip --- .../2026-10-30-attention-retention-sleep.md | 2 +- .../attention-retention-work.test.ts | 111 ++++++++++++++- server/src/services/attention-retention.ts | 31 +++-- .../issue-review-state-notifications.ts | 2 - server/src/services/issues.ts | 127 +++++++++--------- ...ative-status-reassessment-notifications.ts | 4 +- .../services/retention-work-notifications.ts | 48 ++++++- 7 files changed, 237 insertions(+), 88 deletions(-) diff --git a/doc/plans/2026-10-30-attention-retention-sleep.md b/doc/plans/2026-10-30-attention-retention-sleep.md index 463fbe94188..24212b4b98a 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. 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. +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. 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. An unrelated graph reaching its limit cannot change a shallow root’s terminal subject or deadline. 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 29a82f866a5..b1e7949a8b7 100644 --- a/server/src/__tests__/attention-retention-work.test.ts +++ b/server/src/__tests__/attention-retention-work.test.ts @@ -1,7 +1,7 @@ import { randomUUID } from "node:crypto"; import { and, eq, sql } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; -import { authUsers, companyMemberships, agents, approvals, budgetIncidents, budgetPolicies, companies, createDb, applyPendingMigrations, decisionArchiveNotificationOutbox, +import { subscribeDatabaseWork, authUsers, companyMemberships, agents, approvals, budgetIncidents, budgetPolicies, companies, createDb, applyPendingMigrations, decisionArchiveNotificationOutbox, decisionRetention, decisions, costEvents, heartbeatRunEvents, heartbeatRuns, issueApprovals, issueRecoveryActions, issueRelations, issueThreadInteractions, issues, type Db } from "@paperclipai/db"; import { attentionRetentionService, registerAttentionRetentionWork } from "../services/attention-retention.js"; @@ -13,6 +13,7 @@ import { companyService } from "../services/companies.js"; import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; import { decisionRetentionService, DEFAULT_DECISION_ARCHIVE_DAYS } from "../services/decision-retention.js"; import { appendHeartbeatRunEvent } from "../services/heartbeat-run-events.js"; +import { buildIssueGraphLivenessIncidentKey, RECOVERY_ORIGIN_KINDS } from "../services/recovery/origins.js"; import { issueThreadInteractionService } from "../services/issue-thread-interactions.js"; import { createAgentLifecycle } from "../services/agent-lifecycle.js"; import { BLOCKER_ATTENTION_MAX_DEPTH, BLOCKER_ATTENTION_MAX_NODES, issueService } from "../services/issues.js"; @@ -32,12 +33,13 @@ const delay = DEFAULT_DECISION_ARCHIVE_DAYS * 86_400_000; let coordinator: ReturnType | undefined; const errors: unknown[] = []; const statements: string[] = []; + const parameterCounts: number[] = []; beforeAll(async () => { if (externalUrl) await applyPendingMigrations(externalUrl); else temporary = await startEmbeddedPostgresTestDatabase("attention-retention-"); db = createDb(externalUrl ?? temporary!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); db.$client.options.onnotice = () => {}; - db.$client.options.debug = (_connection, query) => statements.push(query); + db.$client.options.debug = (_connection, query, parameters) => { statements.push(query); parameterCounts.push(parameters.length); }; }, 30_000); beforeEach(async () => { companyId = randomUUID(); agentId = randomUUID(); draining = false; errors.length = 0; @@ -311,6 +313,111 @@ const delay = DEFAULT_DECISION_ARCHIVE_DAYS * 86_400_000; } }); + it("bounds every graph query when independent root trees contain over 65535 unfinished nodes", async () => { + await db.execute(sql`insert into issues (id, company_id, title, status, updated_at) + select gen_random_uuid(), ${companyId}::uuid, 'parameter-root-' || n, 'blocked', ${old().toISOString()}::timestamptz + from generate_series(1, 70) n`); + await db.execute(sql`insert into issues (id, company_id, parent_id, title, status, assignee_agent_id, updated_at) + select gen_random_uuid(), ${companyId}::uuid, root.id, 'parameter-leaf-' || n, 'backlog', ${agentId}::uuid, ${old().toISOString()}::timestamptz + from issues root cross join generate_series(1, 999) n + where root.company_id = ${companyId}::uuid and root.status = 'blocked'`); + const roots = await db.select().from(issues).where(and(eq(issues.companyId, companyId), eq(issues.status, "blocked"))); + parameterCounts.length = 0; + const result = await issueService(db).listBlockerAttention(companyId, roots).catch(error => { + throw new Error(`Large graph failed: ${String(error.cause?.message ?? error.message).slice(0, 300)}`); + }); + expect(result.size).toBe(70); + expect([...result.values()].every(item => item.state === "needs_attention" && item.terminalBlockerIssueId)).toBe(true); + expect(Math.max(...parameterCounts)).toBeLessThanOrEqual(1100); + }, 120_000); + + it("canonicalizes wide circular notification variables to the supported identity fields", async () => { + const sourceId = randomUUID(), issueId = randomUUID(), hints: string[] = []; + const wide = { companyId, sourceKind: "review" as const, sourceId, issueId, ignored: {} as unknown }; + wide.ignored = wide; + const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.attentionRetention, (event, values) => { + if (event === "settled") hints.push(...values ?? []); + }); + try { await db.transaction(tx => notifyAttentionRetentionWork(tx, wide)); } finally { unsubscribe(); } + expect(hints).toEqual([JSON.stringify({ companyId, sourceKind: "review", sourceId, issueId })]); + }); + + it("keeps held, burst and sustained ordinary issue writes scoped while an unrelated deadline fires", async () => { + const row = await issueService(db).create(companyId, { title: "Active review", status: "in_review", assigneeUserId: "board", createdByAgentId: agentId }); + const urgent = await insertIssue("in_review", { assigneeUserId: "board", updatedAt: new Date(Date.now() - delay + 10_000) }); + const otherCompany = randomUUID(); + await db.insert(companies).values({ id: otherCompany, name: "Unrelated", issuePrefix: "BURST" }); + const foreign = await db.insert(issues).values(Array.from({ length: 125 }, () => ({ companyId: otherCompany, + title: randomUUID(), status: "in_review", assigneeUserId: "board", updatedAt: old() }))).returning(); + await db.insert(decisionRetention).values(foreign.map(issue => ({ companyId: otherCompany, sourceKind: "review", sourceId: issue.id, sourceActivityAt: issue.updatedAt, keep: true }))); + const service = attentionRetentionService(db), original = service.load, load = vi.spyOn(service, "load"); + await start(service); await settle(); load.mockClear(); + let release!: () => void, entered = false; + const gate = new Promise(resolve => { release = resolve; }); + load.mockImplementationOnce(async (...args) => { const page = await original(...args); entered = true; await gate; return page; }); + statements.length = 0; + await issueService(db).update(row.id, { title: "Held mutation" }); + await vi.waitFor(() => expect(entered).toBe(true), { timeout: 15_000 }); + await Promise.all(Array.from({ length: 5 }, (_, index) => issueService(db).update(row.id, { title: `Burst ${index}` }))); + release(); await settle(); + for (let index = 0; index < 8; index++) { + await issueService(db).update(row.id, { title: `Steady ${index}` }); await vi.advanceTimersByTimeAsync(1000); await settle(); + } + await archived("review", urgent.id); + const routine = load.mock.calls.map(([scope]) => scope).filter(scope => !scope?.sourceId); + expect(routine.length).toBeGreaterThan(0); + expect(routine.every(scope => scope?.companyId === companyId && scope.issueId === row.id)).toBe(true); + const scans = statements.filter(query => query.includes("with source_candidates")); + expect(scans.length).toBeGreaterThan(0); + expect(scans.every(query => !query.includes("budget_incidents") && !query.includes("agent_error_alert"))).toBe(true); + expect(await state("review", row.id)).toBeNull(); + }); + + it("wakes all linked issue sources when an ordinary visibility mutation reveals their origin", async () => { + const origin = await insertIssue("in_review", { hiddenAt: old(), assigneeUserId: "board", responsibleUserId: "board" }); + const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId, status: "failed", contextSnapshot: { issueId: origin.id }, createdAt: old(), finishedAt: old() }).returning(); + await db.insert(heartbeatRunEvents).values({ companyId, agentId, runId: run.id, seq: 1, eventType: "lifecycle", message: "Bounded retry exhausted: visibility" }); + const [decision] = await db.insert(decisions).values({ 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 [recovery] = await db.insert(issueRecoveryActions).values({ companyId, sourceIssueId: origin.id, kind: "execution", ownerType: "board", cause: "fixture", fingerprint: "fixture", nextAction: "Choose", updatedAt: old() }).returning(); + const [interaction] = await db.insert(issueThreadInteractions).values({ companyId, issueId: origin.id, kind: "ask_user_questions", payload: { questions: [] }, updatedAt: old() }).returning(); + await start(); await quiet(); + await issueService(db).update(origin.id, { hiddenAt: null }); + for (const [kind, id] of [["decision", decision.id], ["recovery_action", recovery.id], ["failed_run", run.id], ["issue_thread_interaction", interaction.id]]) await archived(kind, id); + await settle(); expect(await state("review", origin.id)).toBeNull(); expect(coordinator!.nextWakeAt()).not.toBeNull(); + }); + + it("invalidates old and new parents, terminal visibility and repointed blocker relations", async () => { + const oldParent = await insertIssue("blocked", { updatedAt: new Date(), responsibleUserId: "board" }); + const newParent = await insertIssue("blocked", { updatedAt: new Date(), responsibleUserId: "board" }); + const terminal = await insertIssue("backlog", { parentId: oldParent.id, hiddenAt: old(), assigneeAgentId: agentId, responsibleUserId: "board" }); + const service = attentionRetentionService(db), load = vi.spyOn(service, "load"); await start(service); await settle(); load.mockClear(); + await issueService(db).update(terminal.id, { parentId: newParent.id }); await settle(); + const ids = load.mock.calls.map(([scope]) => scope?.issueId); + expect(ids).toEqual(expect.arrayContaining([oldParent.id, newParent.id, terminal.id])); + load.mockClear(); + await issueService(db).update(terminal.id, { hiddenAt: null }); await settle(); + expect(await state("blocker_attention", terminal.id)).toBeNull(); + const visible = await service.load({ companyId, issueId: newParent.id }); + expect(visible.rows!.some(entry => entry.sourceId === terminal.id && entry.dueAt > Date.now())).toBe(true); + expect(load.mock.calls.some(([scope]) => scope?.issueId === newParent.id)).toBe(true); + const next = await insertIssue("backlog", { assigneeAgentId: agentId }); + await issueService(db).update(oldParent.id, { blockedByIssueIds: [next.id] }); await archived("blocker_attention", next.id); + }); + + it("cancels and reactivates a blocker deadline through ordinary liveness incident creation and hiding", async () => { + const root = await insertIssue("blocked"); + const terminal = await insertIssue("backlog", { assigneeAgentId: agentId, updatedAt: new Date(Date.now() - delay + 10_000) }); + await db.insert(issueRelations).values({ companyId, issueId: terminal.id, relatedIssueId: root.id, type: "blocks" }); + await start(); await settle(); expect(coordinator!.nextWakeAt()).not.toBeNull(); + const incident = await issueService(db).create(companyId, { title: "Resolve liveness", status: "todo", assigneeUserId: "board", + createdByUserId: "board", originKind: RECOVERY_ORIGIN_KINDS.issueGraphLivenessEscalation, + originId: buildIssueGraphLivenessIncidentKey({ companyId, issueId: root.id, state: "needs_attention", blockerIssueId: terminal.id }) }); + await quiet(); expect(await state("blocker_attention", terminal.id)).toBeNull(); + await issueService(db).update(incident.id, { hiddenAt: new Date() }); + await archived("blocker_attention", terminal.id); await quiet(); + }); + 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 c72947e5349..27aedba8c47 100644 --- a/server/src/services/attention-retention.ts +++ b/server/src/services/attention-retention.ts @@ -18,8 +18,8 @@ const archiveDelay = DEFAULT_DECISION_ARCHIVE_DAYS * 86_400_000; const kinds = ["decision", "issue_thread_interaction", "recovery_action", "blocker_attention", "review", "failed_run", "budget_alert", "agent_error_alert"] as const; type Kind = typeof kinds[number]; -type Scope = { companyId?: string; sourceKind?: AttentionSourceKind; sourceId?: string }; -type Candidate = { sourceKind: Kind; sourceId: string; companyId: string; activityAt: string; deadlineAt: number }; +type Scope = { companyId?: string; sourceKind?: AttentionSourceKind; sourceId?: string; issueId?: string }; +type Candidate = { sourceKind: Kind; sourceId: string; companyId: string; activityAt: string; deadlineAt: number; issueId: string | null }; type Entry = Candidate & { id: string; dueAt: number; rootId: string; variant?: "owner" | "terminal" }; type Cursor = { sourceKind: string; sourceId: string }; @@ -42,37 +42,39 @@ const sourceTables: Record = { }; const sources: Record = { - decision: sql`select 'decision' as source_kind, d.id::text as source_id, d.company_id, d.updated_at as activity_at + decision: sql`select 'decision' as source_kind, d.id::text as source_id, d.company_id, d.updated_at as activity_at, d.origin_issue_id::text as issue_id from decisions d where d.status = 'open'`, - issue_thread_interaction: sql`select 'issue_thread_interaction', i.id::text, i.company_id, i.updated_at + issue_thread_interaction: sql`select 'issue_thread_interaction', i.id::text, i.company_id, i.updated_at, i.issue_id::text from issue_thread_interactions i where i.status = 'pending' and i.addressee_user_id is null`, - recovery_action: sql`select 'recovery_action', r.id::text, r.company_id, r.updated_at + recovery_action: sql`select 'recovery_action', r.id::text, r.company_id, r.updated_at, r.source_issue_id::text from issue_recovery_actions r where r.status in ('active', 'escalated') and r.owner_type in (${sql.join(HUMAN_RECOVERY_OWNER_TYPES.map(owner => sql`${owner}`), sql`, `)})`, - blocker_attention: sql`select 'blocker_attention', i.id::text, i.company_id, i.updated_at + blocker_attention: sql`select 'blocker_attention', i.id::text, i.company_id, i.updated_at, i.id::text from issues i where i.status = 'blocked' and i.hidden_at is null and i.harness_kind is null`, - review: sql`select 'review', i.id::text, i.company_id, i.updated_at + review: sql`select 'review', i.id::text, i.company_id, i.updated_at, i.id::text from issues i where i.status = 'in_review' and i.hidden_at is null and i.harness_kind is null`, - failed_run: sql`select 'failed_run', r.id::text, r.company_id, coalesce(r.finished_at, r.updated_at, r.created_at) + failed_run: sql`select 'failed_run', r.id::text, r.company_id, coalesce(r.finished_at, r.updated_at, r.created_at), coalesce(r.context_snapshot->>'issueId', r.context_snapshot->>'taskId') from heartbeat_runs r where r.status in ('failed', 'timed_out') and exists ( select 1 from heartbeat_run_events e where e.run_id = r.id and e.company_id = r.company_id and e.event_type = 'lifecycle' and e.message like ${`${RETRY_EXHAUSTION_MESSAGE_PREFIX}%`})`, - budget_alert: sql`select 'budget_alert', b.id::text, b.company_id, b.updated_at from budget_incidents b + budget_alert: sql`select 'budget_alert', b.id::text, b.company_id, b.updated_at, null::text from budget_incidents b where b.status = 'open' and (b.threshold_type = 'hard' or (b.amount_limit > 0 and round(b.amount_observed::numeric / b.amount_limit * 10000) / 100 >= ${ATTENTION_BUDGET_WARNING_PERCENT}))`, - agent_error_alert: sql`select 'agent_error_alert', a.id::text, a.company_id, a.updated_at from agents a where a.status = ${ATTENTION_AGENT_ERROR_STATUS}`, + agent_error_alert: sql`select 'agent_error_alert', a.id::text, a.company_id, a.updated_at, null::text from agents a where a.status = ${ATTENTION_AGENT_ERROR_STATUS}`, }; export function attentionRetentionService(db: Db) { const cursors = new Map(); async function candidates(database: Db, scope: Scope = {}, cursor?: Cursor, sourceId?: string): Promise { - const selected = kinds.filter(kind => !scope.sourceKind || scope.sourceKind === kind); + const selected = kinds.filter(kind => (!scope.sourceKind || scope.sourceKind === kind) + && (!scope.issueId || !["budget_alert", "agent_error_alert"].includes(kind))); if (!selected.length) return []; - const rows = await database.execute(sql`with source_candidates(source_kind, source_id, company_id, activity_at) as (${sql.join(selected.map(kind => sources[kind]), sql` union all `)}) + const rows = await database.execute(sql`with source_candidates(source_kind, source_id, company_id, activity_at, issue_id) as (${sql.join(selected.map(kind => sources[kind]), sql` union all `)}) select s.source_kind as "sourceKind", s.source_id as "sourceId", s.company_id as "companyId", - s.activity_at::text as "activityAt", ceil(extract(epoch from s.activity_at) * 1000 + ${archiveDelay}) as "deadlineAt" + s.issue_id as "issueId", s.activity_at::text as "activityAt", ceil(extract(epoch from s.activity_at) * 1000 + ${archiveDelay}) as "deadlineAt" from source_candidates s inner join companies c on c.id = s.company_id left join decision_retention r on r.company_id = s.company_id and r.source_kind = s.source_kind and r.source_id = s.source_id where c.status = 'active' + and (${scope.issueId ?? null}::text is null or s.issue_id = ${scope.issueId ?? null}) and (${scope.companyId ?? null}::uuid is null or s.company_id = ${scope.companyId ?? null}::uuid) and (${sourceId ?? scope.sourceId ?? null}::text is null or s.source_id = ${sourceId ?? scope.sourceId ?? null}) and (s.source_kind = 'blocker_attention' or ((r.id is null or r.keep = false) and r.archived_at is null)) @@ -302,7 +304,8 @@ export function registerAttentionRetentionWork(coordinator: ReturnType hint === "*" ? undefined : JSON.parse(hint) as Scope, matches: (entry, scope) => (!scope?.companyId || scope.companyId === entry.companyId) && (!scope?.sourceKind || scope.sourceKind === entry.sourceKind) - && (!scope?.sourceId || scope.sourceId === entry.sourceId), + && (!scope?.sourceId || scope.sourceId === entry.sourceId) + && (!scope?.issueId || scope.issueId === entry.issueId), load: service.load, check: service.check, }, canRun); } diff --git a/server/src/services/issue-review-state-notifications.ts b/server/src/services/issue-review-state-notifications.ts index 353114a9bf0..679b7e37475 100644 --- a/server/src/services/issue-review-state-notifications.ts +++ b/server/src/services/issue-review-state-notifications.ts @@ -1,11 +1,9 @@ -import { notifyAttentionRetentionWork } from "./retention-work-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { notifyExternalConversationSettlement } from "./external-conversation-settlement-notifications.js"; import { notifyNativeReviewResponse } from "./native-review-response-notifications.js"; import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js"; export async function notifyIssueReviewState(tx: object, scope: { companyId: string; issueId?: string; runId?: string; agentId?: string }) { - await notifyAttentionRetentionWork(tx, { companyId: scope.companyId }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); await notifyExternalConversationSettlement(tx, scope); await notifyNativeReviewResponse(tx, scope); diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index fd1fd9ba819..c91a325ad39 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -1,4 +1,4 @@ -import { notifyAttentionRetentionWork } from "./retention-work-notifications.js"; +import { notifyIssueAttentionRetentionWork } from "./retention-work-notifications.js"; import { notifyExternalConversationSettlement } from "./external-conversation-settlement-notifications.js"; import { notifyNativeRecoveryNotices } from "./native-recovery-notice-notifications.js"; import { notifyNativeRunProjection } from "./native-run-projection-notifications.js"; @@ -2538,33 +2538,37 @@ async function listPendingFinalizeBlockerIssueIds( ), ); - if (lockSources) { + const rows: Array<{ issueId: string | null; executionWorkspaceId: string | null; + phase: string; status: string; startedAt: Date }> = []; + for (const chunk of chunkList(executionWorkspaceIds, ISSUE_LIST_RELATED_QUERY_CHUNK_SIZE)) { + if (lockSources) { - await dbOrTx.select({ id: executionWorkspaces.id }).from(executionWorkspaces) - .where(and(eq(executionWorkspaces.companyId, companyId), inArray(executionWorkspaces.id, executionWorkspaceIds))) - .for("update", { noWait: true }); - } - const query = dbOrTx - .select({ - issueId: workspaceOperations.issueId, - executionWorkspaceId: workspaceOperations.executionWorkspaceId, - phase: workspaceOperations.phase, - status: workspaceOperations.status, - startedAt: workspaceOperations.startedAt, - }) - .from(workspaceOperations) - .where( - and( - eq(workspaceOperations.companyId, companyId), - inArray( - workspaceOperations.executionWorkspaceId, - executionWorkspaceIds, + await dbOrTx.select({ id: executionWorkspaces.id }).from(executionWorkspaces) + .where(and(eq(executionWorkspaces.companyId, companyId), inArray(executionWorkspaces.id, chunk))) + .for("update", { noWait: true }); + } + const query = dbOrTx + .select({ + issueId: workspaceOperations.issueId, + executionWorkspaceId: workspaceOperations.executionWorkspaceId, + phase: workspaceOperations.phase, + status: workspaceOperations.status, + startedAt: workspaceOperations.startedAt, + }) + .from(workspaceOperations) + .where( + and( + eq(workspaceOperations.companyId, companyId), + inArray( + workspaceOperations.executionWorkspaceId, + chunk, + ), ), - ), - ); + ); - const rows = await (lockSources ? query.for("share", { noWait: true }) : query); + rows.push(...await (lockSources ? query.for("share", { noWait: true }) : query)); + } const latestAttributedByBlockerWorkspace = new Map< string, @@ -4119,23 +4123,26 @@ async function listIssueBlockerAttentionMap( ownerType: string; ownerAgentId: string | null; ownerUserId: string | null; - }> = await dbOrTx - .select({ - id: issueRecoveryActions.id, - sourceIssueId: issueRecoveryActions.sourceIssueId, - status: issueRecoveryActions.status, - ownerType: issueRecoveryActions.ownerType, - ownerAgentId: issueRecoveryActions.ownerAgentId, - ownerUserId: issueRecoveryActions.ownerUserId, - }) - .from(issueRecoveryActions) - .where( - and( - eq(issueRecoveryActions.companyId, companyId), - inArray(issueRecoveryActions.status, ["active", "escalated"]), - inArray(issueRecoveryActions.sourceIssueId, explicitWaitCandidateIds), - ), - ); + }> = []; + for (const chunk of chunkList(explicitWaitCandidateIds, ISSUE_LIST_RELATED_QUERY_CHUNK_SIZE)) { + recoveryActionRows.push(...await dbOrTx + .select({ + id: issueRecoveryActions.id, + sourceIssueId: issueRecoveryActions.sourceIssueId, + status: issueRecoveryActions.status, + ownerType: issueRecoveryActions.ownerType, + ownerAgentId: issueRecoveryActions.ownerAgentId, + ownerUserId: issueRecoveryActions.ownerUserId, + }) + .from(issueRecoveryActions) + .where( + and( + eq(issueRecoveryActions.companyId, companyId), + inArray(issueRecoveryActions.status, ["active", "escalated"]), + inArray(issueRecoveryActions.sourceIssueId, chunk), + ), + )); + } const recoveryActionIds = recoveryActionRows.map((row) => row.id); const liveRecoveryActionIds = new Set(); for (const chunk of chunkList( @@ -4198,22 +4205,11 @@ async function listIssueBlockerAttentionMap( } } - const agentRows: IssueBlockerAttentionAgentRow[] = - agentIds.size > 0 - ? await dbOrTx - .select({ - id: agents.id, - companyId: agents.companyId, - status: agents.status, - }) - .from(agents) - .where( - and( - eq(agents.companyId, companyId), - inArray(agents.id, [...agentIds]), - ), - ) - : []; + const agentRows: IssueBlockerAttentionAgentRow[] = []; + for (const chunk of chunkList([...agentIds], ISSUE_LIST_RELATED_QUERY_CHUNK_SIZE)) { + agentRows.push(...await dbOrTx.select({ id: agents.id, companyId: agents.companyId, status: agents.status }) + .from(agents).where(and(eq(agents.companyId, companyId), inArray(agents.id, chunk)))); + } const agentsById = new Map(agentRows.map((agent) => [agent.id, agent])); type PathClassification = { @@ -7641,7 +7637,7 @@ export function issueService(db: Db) { await assertNoBlockingCycles(companyId, issueId, deduped, dbOrTx); } - await notifyAttentionRetentionWork(dbOrTx, { companyId }); + await notifyIssueAttentionRetentionWork(dbOrTx, { companyId, issueId }); await dbOrTx .delete(issueRelations) .where( @@ -10627,7 +10623,8 @@ export function issueService(db: Db) { } if (values.monitorNextCheckAt) await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); - await notifyAttentionRetentionWork(tx, { companyId }); + values.id ??= randomUUID(); + await notifyIssueAttentionRetentionWork(tx, { companyId, issueId: values.id, parentId: values.parentId, originKind: values.originKind, originId: values.originId }); const [issue] = await tx.insert(issues).values(values).returning(); await attachOwnedDraftImages(tx, issue, issue.description, { userId: issueData.createdByUserId, agentId: issueData.createdByAgentId }); if (issue.visibility === "private" && !inheritedPrivateSubtree && issueData.createdByAgentId) { @@ -11513,7 +11510,7 @@ export function issueService(db: Db) { || issueData.parentId !== undefined || issueData.executionWorkspaceId !== undefined) { await notifyTerminalWorkspaceWork(tx, `issue:${id}`); } - await notifyAttentionRetentionWork(tx, { companyId: receiptExisting.companyId }); + await notifyIssueAttentionRetentionWork(tx, { companyId: receiptExisting.companyId, issueId: id, parentId: patch.parentId, originKind: patch.originKind, originId: patch.originId }); if (patch.status !== undefined || patch.assigneeAgentId !== undefined || patch.assigneeUserId !== undefined || patch.hiddenAt !== undefined || patch.conversationAgentId !== undefined) { await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); @@ -11975,7 +11972,7 @@ export function issueService(db: Db) { await requestRestorePointDeletion(tx, { sourceIssueId: id }); const [owner] = await tx.select({ companyId: issues.companyId }).from(issues).where(eq(issues.id, id)); if (owner) { - await notifyAttentionRetentionWork(tx, { companyId: owner.companyId }); + await notifyIssueAttentionRetentionWork(tx, { companyId: owner.companyId, issueId: id }); await notifyExternalConversationSettlement(tx, { companyId: owner.companyId }); await notifyApprovalWork(tx, owner.companyId); } @@ -12108,7 +12105,7 @@ export function issueService(db: Db) { ) : isNull(issues.executionRunId); const updated = await db.transaction(async tx => { - await notifyAttentionRetentionWork(tx, { companyId: issueCompany.companyId }); + await notifyIssueAttentionRetentionWork(tx, { companyId: issueCompany.companyId, issueId: id }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); return tx @@ -12238,7 +12235,7 @@ export function issueService(db: Db) { } const staleExecutionRunId = current.executionRunId; const adopted = await db.transaction(async tx => { - await notifyAttentionRetentionWork(tx, { companyId: issueCompany.companyId }); + await notifyIssueAttentionRetentionWork(tx, { companyId: issueCompany.companyId, issueId: id }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); return tx @@ -12494,7 +12491,7 @@ export function issueService(db: Db) { existing.status === "in_progress" ? "todo" : existing.status; await notifyTerminalWorkspaceWork(tx, `issue:${id}`); await notifyNativeStatusReassessment(tx, { companyId: existing.companyId, issueId: id }); - await notifyAttentionRetentionWork(tx, { companyId: existing.companyId }); + await notifyIssueAttentionRetentionWork(tx, { companyId: existing.companyId, issueId: id }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); const updated = await tx @@ -12547,7 +12544,7 @@ export function issueService(db: Db) { patch.assigneeAgentId = null; } - await notifyAttentionRetentionWork(tx, { companyId: existing.companyId }); + await notifyIssueAttentionRetentionWork(tx, { companyId: existing.companyId, issueId: id }); await notifyTerminalWorkspaceWork(tx, `issue:${id}`); if (existing.checkoutRunId || existing.executionRunId || options.clearAssignee) { await notifyExternalConversationSettlement(tx, { companyId: existing.companyId, issueId: id }); diff --git a/server/src/services/native-status-reassessment-notifications.ts b/server/src/services/native-status-reassessment-notifications.ts index 3a5ec8b6c37..30c1bbd098b 100644 --- a/server/src/services/native-status-reassessment-notifications.ts +++ b/server/src/services/native-status-reassessment-notifications.ts @@ -1,10 +1,10 @@ -import { notifyAttentionRetentionWork } from "./retention-work-notifications.js"; +import { notifyIssueAttentionRetentionWork } from "./retention-work-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; export type NativeStatusReassessmentScope = { companyId: string; issueId?: string; runId?: string; agentId?: string }; export async function notifyNativeStatusReassessment(tx: object, scope: NativeStatusReassessmentScope) { - await notifyAttentionRetentionWork(tx, { companyId: scope.companyId }); + await notifyIssueAttentionRetentionWork(tx, { companyId: scope.companyId, issueId: scope.issueId, runId: scope.runId }); return notifyDeliveryWork(tx, DELIVERY_QUEUES.nativeStatusReassessment, JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId, runId: scope.runId, agentId: scope.agentId })); } diff --git a/server/src/services/retention-work-notifications.ts b/server/src/services/retention-work-notifications.ts index 0331f53f258..cee2486f54c 100644 --- a/server/src/services/retention-work-notifications.ts +++ b/server/src/services/retention-work-notifications.ts @@ -1,10 +1,54 @@ import { notifyApprovalWork } from "./approval-work-notifications.js"; +import { RECOVERY_ORIGIN_KINDS, parseIssueGraphLivenessIncidentKey } from "./recovery/origins.js"; +import { sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import type { AttentionSourceKind } from "@paperclipai/shared"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; -export async function notifyAttentionRetentionWork(transaction: object, scope: { companyId?: string; sourceKind?: AttentionSourceKind; sourceId?: string } = {}) { - await notifyDeliveryWork(transaction, DELIVERY_QUEUES.attentionRetention, JSON.stringify(scope)); +export async function notifyAttentionRetentionWork(transaction: object, scope: { companyId?: string; sourceKind?: AttentionSourceKind; issueId?: string; sourceId?: string } = {}) { + await notifyDeliveryWork(transaction, DELIVERY_QUEUES.attentionRetention, JSON.stringify({ + companyId: scope.companyId, sourceKind: scope.sourceKind, sourceId: scope.sourceId, issueId: scope.issueId, + })); +} + +export async function notifyIssueAttentionRetentionWork(transaction: object, + scope: { companyId: string; issueId?: string; parentId?: string | null; runId?: string; originKind?: string | null; originId?: string | null }) { + const database = transaction as Db; + const seeds = [scope.issueId, scope.parentId].filter((id): id is string => Boolean(id)); + if (!seeds.length && scope.runId) { + const rows = await database.execute(sql`select issue_id::text as id from heartbeat_runs + where company_id = ${scope.companyId}::uuid and id = ${scope.runId}::uuid and issue_id is not null + union select native_issue_id::text from heartbeat_runs + where company_id = ${scope.companyId}::uuid and id = ${scope.runId}::uuid and native_issue_id is not null + union select coalesce(context_snapshot->>'issueId', context_snapshot->>'taskId') from heartbeat_runs + where company_id = ${scope.companyId}::uuid and id = ${scope.runId}::uuid + and coalesce(context_snapshot->>'issueId', context_snapshot->>'taskId') is not null`); + for (const row of rows) seeds.push(row.id as string); + } + if (!seeds.length) return notifyAttentionRetentionWork(transaction, { companyId: scope.companyId }); + const origins = await database.execute(sql`select id::text as id, origin_kind as "originKind", origin_id as "originId" from issues + where company_id = ${scope.companyId}::uuid + and id in (${sql.join([...new Set(seeds)].map(id => sql`${id}::uuid`), sql`, `)})`); + const current = origins.find(source => source.id === scope.issueId); + origins.push({ originKind: scope.originKind ?? current?.originKind, originId: scope.originId ?? current?.originId }); + for (const source of origins) { + const incident = source.originKind === RECOVERY_ORIGIN_KINDS.issueGraphLivenessEscalation + ? parseIssueGraphLivenessIncidentKey(source.originId as string | null) : null; + if (incident?.companyId === scope.companyId) seeds.push(incident.issueId, incident.leafIssueId); + } + const affected = await database.execute(sql`with recursive affected(id) as ( + select unnest(array[${sql.join([...new Set(seeds)].map(id => sql`${id}::uuid`), sql`, `)}]) + union + select parent.id from affected current cross join lateral ( + select related_issue_id as id from issue_relations where company_id = ${scope.companyId}::uuid + and issue_id = current.id and type = 'blocks' + union select parent_id from issues where company_id = ${scope.companyId}::uuid + and id = current.id and parent_id is not null + ) parent + ) select id::text as "issueId" from affected`); + for (const row of affected) await notifyAttentionRetentionWork(transaction, { + companyId: scope.companyId, issueId: row.issueId as string, + }); } export async function notifyRetentionWork(transaction: object, kind: AttentionSourceKind, scope: { companyId?: string; sourceId?: string } = {}) {