diff --git a/doc/plans/2026-10-30-attention-retention-sleep.md b/doc/plans/2026-10-30-attention-retention-sleep.md index c0e4e168bd0..1b71dc71261 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. +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. 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. @@ -33,7 +33,7 @@ The writer contract follows the existing delivery workers: serving processes use ## Real PostgreSQL evidence proves quiet and reactivation -The new acceptance suite exercises all eight sources, including terminal blocker identity and date, newest confirmations, independent quality gates, unhealthy reporting chains, conversation visibility, maintained reviews and pending approvals, newer-run suppression, ordinary task and interaction producers, ordinary exhaustion-event ingress, and ordinary error clearing. +The new acceptance suite exercises all eight sources, including terminal blocker identity and date, newest confirmations, independent quality gates, unhealthy reporting chains, conversation visibility, maintained reviews and pending approvals, newer-run suppression, ordinary task and interaction producers, ordinary exhaustion-event ingress, and ordinary error clearing. Additional ordinary producer controls cover child creation with a parent-blocking relation, budget incident creation and resolution, and repeated company-scoped execution-control writes. It also checks keep/revive, moved and cancelled deadlines, company activation and isolation, retained terminal history, restart, nested outer commit and rollback, first writes during drain, retry across drain, atomic archive/outbox rollback, and notifications during reconstruction. A 225-item backlog uses three startup pages and source-specific self-invalidations. SQL is counted at the PostgreSQL driver boundary. diff --git a/server/src/__tests__/attention-retention-work.test.ts b/server/src/__tests__/attention-retention-work.test.ts index 22a966d5fef..889030d2c22 100644 --- a/server/src/__tests__/attention-retention-work.test.ts +++ b/server/src/__tests__/attention-retention-work.test.ts @@ -2,9 +2,12 @@ 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, - decisionRetention, decisions, heartbeatRunEvents, heartbeatRuns, issueApprovals, issueRecoveryActions, + decisionRetention, decisions, costEvents, heartbeatRunEvents, heartbeatRuns, issueApprovals, issueRecoveryActions, issueRelations, issueThreadInteractions, issues, type Db } from "@paperclipai/db"; import { attentionRetentionService, registerAttentionRetentionWork } from "../services/attention-retention.js"; +import { budgetService } from "../services/budgets.js"; +import { writeExecutionControl } from "../services/execution-control-notifications.js"; +import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js"; import { attentionService } from "../services/attention.js"; import { companyService } from "../services/companies.js"; import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; @@ -135,6 +138,76 @@ const delay = DEFAULT_DECISION_ARCHIVE_DAYS * 86_400_000; await coordinator!.stop(); coordinator = undefined; await start(); await quiet(); }); + it("registers the ordinary createChild blocker path inside its relation transaction", async () => { + const parent = await issueService(db).create(companyId, { title: "Parent", status: "in_progress", assigneeAgentId: agentId, createdByAgentId: agentId }); + await start(); await quiet(); + const { issue: child, parentBlockerAdded } = await issueService(db).createChild(parent.id, { + title: "Child needs review", status: "in_review", assigneeUserId: "board", createdByAgentId: agentId, + blockParentUntilDone: true, + }); + expect(parentBlockerAdded).toBe(true); + expect(await db.select().from(issueRelations).where(and(eq(issueRelations.issueId, child.id), + eq(issueRelations.relatedIssueId, parent.id)))).toHaveLength(1); + await settle(); expect(coordinator!.nextWakeAt()).not.toBeNull(); + await vi.advanceTimersByTimeAsync(delay + 2000); await archived("review", child.id); await quiet(); + expect(await db.select().from(decisionArchiveNotificationOutbox).where(eq(decisionArchiveNotificationOutbox.sourceId, child.id))).toHaveLength(1); + }); + + it("batches a full page of maintained reviews and confirmations by company", async () => { + const rows = await db.insert(issues).values(Array.from({ length: 100 }, () => ({ companyId, + title: randomUUID(), status: "in_review", assigneeAgentId: agentId, updatedAt: old() }))).returning(); + await db.insert(issueThreadInteractions).values(rows.map(row => ({ companyId, issueId: row.id, + kind: "request_confirmation", addresseeAgentId: agentId, payload: { version: 1, prompt: "Agent verdict" } }))); + statements.length = 0; + const page = await attentionRetentionService(db).load({ companyId, sourceKind: "review" }); + expect(page.rows).toEqual([]); expect(page.complete).toBe(false); + expect(statements.length).toBeLessThanOrEqual(20); + expect(statements.filter(query => query.includes('from "agents"')).length).toBeLessThanOrEqual(1); + await db.update(agents).set({ status: "paused" }).where(eq(agents.id, agentId)); + statements.length = 0; + const interactions = await attentionRetentionService(db).load({ companyId, sourceKind: "issue_thread_interaction" }); + expect(interactions.rows).toHaveLength(100); expect(interactions.complete).toBe(false); + expect(statements.length).toBeLessThanOrEqual(6); + expect(statements.filter(query => query.includes('from "agents"'))).toHaveLength(1); + }); + + it("scopes ordinary execution-control writes to their company", async () => { + const otherCompany = randomUUID(); + await db.insert(companies).values({ id: otherCompany, name: "Other", issuePrefix: "OTHER" }); + await db.insert(issues).values(Array.from({ length: 125 }, () => ({ companyId: otherCompany, + title: randomUUID(), status: "in_review", assigneeUserId: "board", updatedAt: old() }))); + const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId, status: "running" }).returning(); + const service = attentionRetentionService(db), load = vi.spyOn(service, "load"); + await start(service); await quiet(); load.mockClear(); + for (let count = 0; count < 3; count++) { + await writeExecutionControl(db, DELIVERY_QUEUES.executionFinalization, + tx => tx.update(heartbeatRuns).set({ updatedAt: new Date() }).where(eq(heartbeatRuns.id, run.id)), { companyId }); + await quiet(); + } + expect(load).toHaveBeenCalled(); + expect(load.mock.calls.every(([scope]) => scope?.companyId === companyId)).toBe(true); + }); + + it("registers and cancels budget deadlines through ordinary policy and incident producers", async () => { + const service = budgetService(db); + await service.upsertPolicy(companyId, { scopeType: "agent", scopeId: agentId, amount: 100, + warnPercent: 85, hardStopEnabled: false, notifyEnabled: true }, "board"); + const retention = attentionRetentionService(db), load = vi.spyOn(retention, "load"); + await start(retention); await quiet(); load.mockClear(); + const [event] = await db.insert(costEvents).values({ companyId, agentId, provider: "openai", biller: "openai", + billingType: "metered_api", model: "fixture", inputTokens: 100, outputTokens: 20, + costCents: 90, occurredAt: new Date() }).returning(); + await service.evaluateCostEvent(event); + await settle(); expect(coordinator!.nextWakeAt()).not.toBeNull(); + const [incident] = await db.select().from(budgetIncidents).where(eq(budgetIncidents.companyId, companyId)); + expect(incident).toMatchObject({ status: "open", thresholdType: "soft" }); + expect(await state("budget_alert", incident.id)).toBeNull(); + await service.upsertPolicy(companyId, { scopeType: "agent", scopeId: agentId, amount: 200 }, "board"); + await quiet(); + expect((await db.select().from(budgetIncidents).where(eq(budgetIncidents.id, incident.id)))[0].status).toBe("resolved"); + expect(load.mock.calls.every(([scope]) => scope?.companyId === companyId)).toBe(true); + }); + it("matches the existing feed across all eight remaining source kinds and exclusion variants", async () => { const origin = await insertIssue(); const review = await insertIssue("in_review", { assigneeUserId: "board" }); diff --git a/server/src/__tests__/company-portability.test.ts b/server/src/__tests__/company-portability.test.ts index fdf00b828af..ee7db993937 100644 --- a/server/src/__tests__/company-portability.test.ts +++ b/server/src/__tests__/company-portability.test.ts @@ -4184,6 +4184,7 @@ describe("company portability", () => { }), // eslint-disable-next-line @typescript-eslint/no-explicit-any } as any; + db.transaction = async (write: (tx: typeof db) => Promise) => write(db); return { db, insertedRelationValues, monitorUpdates }; } diff --git a/server/src/__tests__/decision-effect-recovery-work.test.ts b/server/src/__tests__/decision-effect-recovery-work.test.ts index 6c1a17d2b39..8a7d4e300c8 100644 --- a/server/src/__tests__/decision-effect-recovery-work.test.ts +++ b/server/src/__tests__/decision-effect-recovery-work.test.ts @@ -73,7 +73,7 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes const claim = (id: string, updatedAt = new Date(), owner = db) => writeDecisionChange(owner, async tx => tx.update(decisions).set({ status: "decided", executionStatus: "running", chosenOptionId: "yes", decidedByUserId: userId, updatedAt, inputValues: {}, metadata: { continuationPending: true }, - }).where(eq(decisions.id, id))); + }).where(eq(decisions.id, id)), { companyId }); async function start() { coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining, canReconcile: () => true, onError: error => errors.push(error) }); @@ -467,7 +467,7 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes } else { await writeDecisionChange(db, async tx => tx.update(decisions).set({ status: "decided", executionStatus: "running", chosenOptionId: "archive", decidedByUserId: userId, updatedAt: new Date(Date.now() - 120_000), - metadata: { ...row.metadata, continuationPending: true } }).where(eq(decisions.id, row.id))); + metadata: { ...row.metadata, continuationPending: true } }).where(eq(decisions.id, row.id)), { companyId }); } await start(); await settled(row.id, committed ? "failed" : "succeeded"); await quiet(); const [retained] = await db.select().from(decisionRetention); diff --git a/server/src/__tests__/decisions-service.test.ts b/server/src/__tests__/decisions-service.test.ts index 7fc2cc50c0d..999bde8d60e 100644 --- a/server/src/__tests__/decisions-service.test.ts +++ b/server/src/__tests__/decisions-service.test.ts @@ -793,7 +793,7 @@ describePg("decisionService", () => { expect((await service().get(id))?.metadata.continuationPending).toBe(false); } const markReady = (id: string, source = db) => writeDecisionChange(source, async tx => tx.update(decisions) - .set({ status: "cancelled", metadata: { continuationPending: true } }).where(eq(decisions.id, id))); + .set({ status: "cancelled", metadata: { continuationPending: true } }).where(eq(decisions.id, id)), { companyId }); it("coalesces sequential tenant writes without dropping pending recovery", async () => { const row = await createCommentDecision(); diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index bbb020c865d..bc72ad57f02 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -1,3 +1,4 @@ +import { attentionRetentionService } from "../services/attention-retention.js"; import { nativeHarnessBackupExpirationService } from "../services/native-runtime/native-harness-backup-expiration.js"; import { skillVersionRetentionService } from "../services/skill-version-retention.js"; import { approvalQueueSeedingService } from "../services/approval-queue-seeding-work.js"; @@ -377,6 +378,11 @@ vi.mock("../realtime/live-events-ws.js", () => ({ setupLiveEventsWebSocketServer: vi.fn(), })); +vi.mock("../services/attention-retention.js", async importOriginal => ({ + ...await importOriginal(), + attentionRetentionService: vi.fn(() => ({ load: vi.fn(async () => ({ rows: [], complete: true })), check: vi.fn() })), +})); + vi.mock("../services/join-request-retention.js", async importOriginal => ({ ...await importOriginal(), joinRequestRetentionService: vi.fn(() => ({ sweep: vi.fn(async () => ({ archived: 0 })), next: vi.fn(async () => null) })), @@ -974,7 +980,7 @@ describe("startServer feedback export wiring", () => { } }); - it.each([false, true])("registers archive delivery separately from retention sweeps (scheduler %s)", async enabled => { + it.each([false, true])("registers source retention separately from archive delivery without tick scans (scheduler %s)", async enabled => { loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled, heartbeatSchedulerIntervalMs: 30000 })); let tick: (() => void) | undefined; const timer = vi.spyOn(globalThis, "setInterval").mockImplementation(((callback: () => void, interval: number) => { @@ -986,12 +992,17 @@ describe("startServer feedback export wiring", () => { const registration = deliveryWorkRegisterMock.mock.calls.find(([queue]) => queue === DELIVERY_QUEUES.decisionArchiveDelivery); if (!enabled) { expect(registration).toBeUndefined(); + expect(attentionRetentionService).not.toHaveBeenCalled(); expect(decisionRetentionService).not.toHaveBeenCalled(); expect(tick).toBeDefined(); tick!(); await settleStartupHousekeeping(); expect(decisionRetentionService).not.toHaveBeenCalled(); return; } expect(registration).toBeDefined(); + expect(deliveryWorkRegisterMock.mock.calls.some(([queue]) => queue === DELIVERY_QUEUES.attentionRetention)).toBe(true); + expect(attentionRetentionService).toHaveBeenCalledOnce(); + const source = vi.mocked(attentionRetentionService).mock.results[0]!.value; + expect(source.load).toHaveBeenCalledOnce(); const retention = vi.mocked(decisionRetentionService).mock.results[0]!.value; expect(retention.deliverNotifications).toHaveBeenCalledOnce(); expect(retention.nextArchiveNotificationAt).toHaveBeenCalledOnce(); @@ -999,9 +1010,10 @@ describe("startServer feedback export wiring", () => { const companyReads = () => vi.mocked(db.select).mock.results.filter(({ value }) => vi.mocked(value.from).mock.calls.some(([table]) => table === companies)).length; const before = companyReads(); - expect(before).toBeGreaterThan(0); expect(tick).toBeDefined(); + expect(before).toBe(0); expect(tick).toBeDefined(); tick!(); await settleStartupHousekeeping(); - expect(companyReads()).toBeGreaterThan(before); + expect(companyReads()).toBe(before); + expect(source.load).toHaveBeenCalledOnce(); expect(retention.deliverNotifications).toHaveBeenCalledOnce(); await registration![1].run(new AbortController().signal); expect(retention.deliverNotifications).toHaveBeenCalledTimes(2); diff --git a/server/src/services/attention-retention.ts b/server/src/services/attention-retention.ts index 4622c4883bf..85e27a0d941 100644 --- a/server/src/services/attention-retention.ts +++ b/server/src/services/attention-retention.ts @@ -1,13 +1,14 @@ -import { and, eq, isNull, sql, type SQL } from "drizzle-orm"; +import { and, eq, 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"; import { evaluateAgentInvokability } from "./agent-invokability.js"; -import { collapsePendingConfirmationsToNewest, listFailedRunsWithNewerRun, readRunIssueId, requiresReviewAttention, terminalBlockerRequiresAttention } from "./attention-source-eligibility.js"; +import { HUMAN_RECOVERY_OWNER_TYPES, ATTENTION_BUDGET_WARNING_PERCENT, ATTENTION_AGENT_ERROR_STATUS, collapsePendingConfirmationsToNewest, listFailedRunsWithNewerRun, readRunIssueId, requiresReviewAttention, terminalBlockerRequiresAttention } from "./attention-source-eligibility.js"; import { decisionRetentionService, DEFAULT_DECISION_ARCHIVE_DAYS } from "./decision-retention.js"; import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; import { DELIVERY_QUEUES } from "./delivery-work-notifications.js"; import { attentionIssueCondition } from "./issue-visibility.js"; +import { RETRY_EXHAUSTION_MESSAGE_PREFIX } from "./heartbeat-run-events.js"; import { issueService } from "./issues.js"; import { parseIssueExecutionState } from "./issue-execution-policy.js"; import { isProspectiveBlockedTransition } from "./routable-blocked.js"; @@ -34,7 +35,7 @@ const sources: Record = { issue_thread_interaction: sql`select 'issue_thread_interaction', i.id::text, i.company_id, i.updated_at 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 - from issue_recovery_actions r where r.status in ('active', 'escalated') and r.owner_type in ('user', 'board')`, + 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 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 @@ -42,11 +43,11 @@ const sources: Record = { failed_run: sql`select 'failed_run', r.id::text, r.company_id, coalesce(r.finished_at, r.updated_at, r.created_at) 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 'Bounded retry exhausted%')`, + 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 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 >= 85))`, - agent_error_alert: sql`select 'agent_error_alert', a.id::text, a.company_id, a.updated_at from agents a where a.status = 'error'`, + 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}`, }; export function attentionRetentionService(db: Db) { @@ -74,7 +75,69 @@ export function attentionRetentionService(db: Db) { return row; } - async function eligible(database: Db, candidate: Candidate, variant?: "owner" | "terminal", lock = false): Promise { + async function interactionEligibility(database: Db, candidates: Candidate[]) { + const result = new Map(); + const byCompany = new Map(); + for (const candidate of candidates) { + if (candidate.sourceKind !== "issue_thread_interaction") continue; + const ids = byCompany.get(candidate.companyId) ?? []; + ids.push(candidate.sourceId); + byCompany.set(candidate.companyId, ids); + } + for (const [companyId, ids] of byCompany) { + const rows = await database.select().from(issueThreadInteractions).where(and( + eq(issueThreadInteractions.companyId, companyId), inArray(issueThreadInteractions.id, ids), + eq(issueThreadInteractions.status, "pending"), isNull(issueThreadInteractions.addresseeUserId))); + if (!rows.length) continue; + const visible = await database.select({ id: issues.id }).from(issues).where(and(eq(issues.companyId, companyId), + inArray(issues.id, rows.map(row => row.issueId)), attentionIssueCondition(true))); + const visibleIds = new Set(visible.map(row => row.id)); + const roster = await database.select().from(agents).where(eq(agents.companyId, companyId)); + const boardAddressed = (row: typeof issueThreadInteractions.$inferSelect) => !row.addresseeAgentId + || !evaluateAgentInvokability(roster.find(agent => agent.id === row.addresseeAgentId), roster).invokable; + const confirmationIssueIds = rows.filter(row => row.kind === "request_confirmation").map(row => row.issueId); + const siblings = confirmationIssueIds.length ? await database.select().from(issueThreadInteractions).where(and( + eq(issueThreadInteractions.companyId, companyId), inArray(issueThreadInteractions.issueId, confirmationIssueIds), + eq(issueThreadInteractions.status, "pending"), eq(issueThreadInteractions.kind, "request_confirmation"), + isNull(issueThreadInteractions.addresseeUserId))) : []; + const newest = new Set(collapsePendingConfirmationsToNewest(siblings.filter(boardAddressed)).map(row => row.id)); + for (const row of rows) result.set(row.id, visibleIds.has(row.issueId) && boardAddressed(row) + && (row.kind !== "request_confirmation" || newest.has(row.id))); + } + return result; + } + + async function reviewEligibility(database: Db, candidates: Candidate[]) { + const result = new Map(); + const byCompany = new Map(); + for (const candidate of candidates) { + if (candidate.sourceKind !== "review") continue; + const ids = byCompany.get(candidate.companyId) ?? []; + ids.push(candidate.sourceId); + byCompany.set(candidate.companyId, ids); + } + for (const [companyId, ids] of byCompany) { + const rows = await database.select().from(issues).where(and(eq(issues.companyId, companyId), + inArray(issues.id, ids), attentionIssueCondition(false))); + const pendingApprovals = await database.select({ issueId: issueApprovals.issueId }).from(issueApprovals) + .innerJoin(approvals, eq(approvals.id, issueApprovals.approvalId)).where(and( + eq(issueApprovals.companyId, companyId), eq(approvals.companyId, companyId), + inArray(issueApprovals.issueId, ids), eq(approvals.status, "pending"))); + const approved = new Set(pendingApprovals.map(row => row.issueId)); + const stalled = await issueService(database).listReviewAttention(companyId, rows); + for (const row of rows) { + const state = parseIssueExecutionState(row.executionState); + result.set(row.id, requiresReviewAttention({ + hasHumanParticipant: state?.status === "pending" && state.currentParticipant?.type === "user", + hasUserAssignee: Boolean(row.assigneeUserId), hasPendingApproval: approved.has(row.id), + stalled: stalled.get(row.id)?.state === "stalled", + })); + } + } + return result; + } + + async function eligible(database: Db, candidate: Candidate, variant?: "owner" | "terminal", lock = false, reviewAttention?: Map, interactionAttention?: Map): Promise { let sourceId = candidate.sourceId; let activityAt = candidate.activityAt; let dueAt = candidate.deadlineAt; @@ -83,38 +146,14 @@ export function attentionRetentionService(db: Db) { .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; } else if (candidate.sourceKind === "issue_thread_interaction") { - const [row] = await database.select().from(issueThreadInteractions).where(and( - eq(issueThreadInteractions.companyId, candidate.companyId), eq(issueThreadInteractions.id, sourceId), - eq(issueThreadInteractions.status, "pending"), isNull(issueThreadInteractions.addresseeUserId))); - if (!row || !await visibleIssue(database, candidate.companyId, row.issueId, true)) return null; - const roster = row.addresseeAgentId ? await database.select().from(agents).where(eq(agents.companyId, candidate.companyId)) : []; - if (row.addresseeAgentId && evaluateAgentInvokability(roster.find(agent => agent.id === row.addresseeAgentId), roster).invokable) return null; - if (row.kind === "request_confirmation") { - const siblings = await database.select().from(issueThreadInteractions).where(and( - eq(issueThreadInteractions.companyId, candidate.companyId), eq(issueThreadInteractions.issueId, row.issueId), - eq(issueThreadInteractions.status, "pending"), eq(issueThreadInteractions.kind, "request_confirmation"), - isNull(issueThreadInteractions.addresseeUserId))); - const companyAgents = roster.length ? roster : await database.select().from(agents).where(eq(agents.companyId, candidate.companyId)); - const board = siblings.filter(sibling => !sibling.addresseeAgentId - || !evaluateAgentInvokability(companyAgents.find(agent => agent.id === sibling.addresseeAgentId), companyAgents).invokable); - if (!collapsePendingConfirmationsToNewest(board).some(sibling => sibling.id === sourceId)) return null; - } + 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; } else if (candidate.sourceKind === "review") { - const row = await visibleIssue(database, candidate.companyId, sourceId); - if (!row) return null; - const state = parseIssueExecutionState(row.executionState); - const [approval] = await database.select({ id: approvals.id }).from(issueApprovals).innerJoin(approvals, eq(approvals.id, issueApprovals.approvalId)) - .where(and(eq(issueApprovals.companyId, candidate.companyId), eq(approvals.companyId, candidate.companyId), - eq(issueApprovals.issueId, sourceId), eq(approvals.status, "pending"))).limit(1); - if (!requiresReviewAttention({ hasHumanParticipant: state?.status === "pending" && state.currentParticipant?.type === "user", - hasUserAssignee: Boolean(row.assigneeUserId), hasPendingApproval: Boolean(approval), stalled: false })) { - const attention = await issueService(database).listReviewAttention(candidate.companyId, [row]); - if (!requiresReviewAttention({ hasHumanParticipant: false, hasUserAssignee: false, hasPendingApproval: false, - stalled: attention.get(sourceId)?.state === "stalled" })) return null; - } + 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); if (!row) return null; @@ -157,10 +196,12 @@ export function attentionRetentionService(db: Db) { const key = JSON.stringify(scope ?? {}); const rows = await candidates(db, scope, cursors.get(key)); const entries: Entry[] = []; + const reviews = await reviewEligibility(db, rows); + const interactions = await interactionEligibility(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); + const entry = await eligible(db, row, variant, false, reviews, interactions); if (entry) entries.push(entry); } } diff --git a/server/src/services/attention-source-eligibility.ts b/server/src/services/attention-source-eligibility.ts index 6e3f484c03a..f129c522130 100644 --- a/server/src/services/attention-source-eligibility.ts +++ b/server/src/services/attention-source-eligibility.ts @@ -1,6 +1,10 @@ import { and, eq, gt, inArray, sql, type SQL } from "drizzle-orm"; import { heartbeatRuns, type Db } from "@paperclipai/db"; +export const HUMAN_RECOVERY_OWNER_TYPES = ["user", "board"] as const; +export const ATTENTION_BUDGET_WARNING_PERCENT = 85; +export const ATTENTION_AGENT_ERROR_STATUS = "error"; + export function collapsePendingConfirmationsToNewest = { const PENDING_INTERACTION_STATUSES = ["pending"] as const; const OPEN_RECOVERY_STATUSES = ["active", "escalated"] as const; -const HUMAN_RECOVERY_OWNER_TYPES = ["user", "board"] as const; const DETAIL_EXCERPT_LENGTH = 160; const DETAIL_IMAGE_LIMIT = 3; const OPEN_DECISION_DEFAULT_LIMIT = 500; @@ -1762,7 +1761,7 @@ export function attentionService(db: Db, serviceOptions: AttentionServiceOptions const budgetOverview = await budgetService(db).overview(companyId); for (const incident of budgetOverview.activeIncidents) { const observedPercent = budgetObservedPercent(incident.amountObserved, incident.amountLimit); - if (incident.thresholdType !== "hard" && observedPercent < 85) continue; + if (incident.thresholdType !== "hard" && observedPercent < ATTENTION_BUDGET_WARNING_PERCENT) continue; const dedupKey = `budget:${incident.policyId}:${toIso(incident.windowStart)}:${incident.thresholdType}`; add(createItem({ companyId, @@ -1825,7 +1824,7 @@ export function attentionService(db: Db, serviceOptions: AttentionServiceOptions updatedAt: agents.updatedAt, }) .from(agents) - .where(and(eq(agents.companyId, companyId), eq(agents.status, "error"))) + .where(and(eq(agents.companyId, companyId), eq(agents.status, ATTENTION_AGENT_ERROR_STATUS))) .orderBy(desc(agents.updatedAt), desc(agents.id)); for (const agent of erroredAgents) { diff --git a/server/src/services/budgets.ts b/server/src/services/budgets.ts index 93403926772..c9cedb7f7ee 100644 --- a/server/src/services/budgets.ts +++ b/server/src/services/budgets.ts @@ -505,8 +505,8 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica return incident ? { incident, created: true } : null; } - async function resolveOpenSoftIncidents(policyId: string) { - await notifyAttentionRetentionWork(db, { sourceKind: "budget_alert" }); + async function resolveOpenSoftIncidents(policyId: string, companyId: string) { + await notifyAttentionRetentionWork(db, { companyId, sourceKind: "budget_alert" }); await db .update(budgetIncidents) .set({ @@ -525,11 +525,12 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica async function resolveOpenIncidentsForPolicy( policyId: string, + companyId: string, approvalStatus: "approved" | "rejected" | null, decidedByUserId: string | null, thresholdType?: BudgetThresholdType, ) { - await notifyAttentionRetentionWork(db, { sourceKind: "budget_alert" }); + await notifyAttentionRetentionWork(db, { companyId, sourceKind: "budget_alert" }); const condition = and(eq(budgetIncidents.policyId, policyId), eq(budgetIncidents.status, "open"), thresholdType ? eq(budgetIncidents.thresholdType, thresholdType) : undefined); const openRows = await db @@ -618,9 +619,9 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica if (policy.isActive && policy.hardStopEnabled && policy.amount > 0 && observed.recoveringRunCount > 0) recoveringPolicy ??= policy; // A closed UTC window, inactive policy, or budget raise releases only // budget-owned pauses. Manual pauses remain operator decisions. - await resolveOpenIncidentsForPolicy(policy.id, "approved", "budget_service", "hard"); + await resolveOpenIncidentsForPolicy(policy.id, policy.companyId, "approved", "budget_service", "hard"); if (!policy.isActive || !policy.notifyEnabled || compareCents(observed.totalExact, Math.ceil(policy.amount * policy.warnPercent / 100)) < 0) { - await resolveOpenIncidentsForPolicy(policy.id, "approved", "budget_service", "soft"); + await resolveOpenIncidentsForPolicy(policy.id, policy.companyId, "approved", "budget_service", "soft"); } } // Agent status has its own admission gate and error code. Preserve a @@ -677,7 +678,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica } if (observedBlocks(policy, observed)) { - await resolveOpenSoftIncidents(policy.id); + await resolveOpenSoftIncidents(policy.id, policy.companyId); const hardIncident = await createIncidentIfNeeded(policy, "hard", observed); await pauseAndCancelScopeForBudget(policy); if (hardIncident?.created) { @@ -847,11 +848,11 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica const observed = await computeObservedSpend(db, row); if (!observedBlocks(row, observed)) { await resumeScopeFromBudget(row); - await resolveOpenIncidentsForPolicy(row.id, "approved", actorUserId ?? "budget_service", "hard"); + await resolveOpenIncidentsForPolicy(row.id, row.companyId, "approved", actorUserId ?? "budget_service", "hard"); if (row.notifyEnabled && compareCents(observed.totalExact, Math.ceil(row.amount * row.warnPercent / 100)) >= 0) { await createIncidentIfNeeded(row, "soft", observed); } else { - await resolveOpenIncidentsForPolicy(row.id, null, null, "soft"); + await resolveOpenIncidentsForPolicy(row.id, row.companyId, null, null, "soft"); } } else { const softThreshold = Math.ceil((row.amount * row.warnPercent) / 100); @@ -859,14 +860,14 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica await createIncidentIfNeeded(row, "soft", observed); } if (observedBlocks(row, observed)) { - await resolveOpenSoftIncidents(row.id); + await resolveOpenSoftIncidents(row.id, row.companyId); await createIncidentIfNeeded(row, "hard", observed); await pauseAndCancelScopeForBudget(row); } } } else { await resumeScopeFromBudget(row); - await resolveOpenIncidentsForPolicy(row.id, "approved", actorUserId ?? "budget_service"); + await resolveOpenIncidentsForPolicy(row.id, row.companyId, "approved", actorUserId ?? "budget_service"); } await recordActivity({ diff --git a/server/src/services/decision-expiry.ts b/server/src/services/decision-expiry.ts index ea5236a6ac0..796200128a7 100644 --- a/server/src/services/decision-expiry.ts +++ b/server/src/services/decision-expiry.ts @@ -23,7 +23,7 @@ export function decisionExpiryService(db: Db, deliverContinuation: (id: string) agentId: row.originAgentId, runId: row.originRunId, action: "decision.expired", entityType: "decision", entityId: row.id, details: { expiredReason: reason } }, publications); return row; - }); + }, { companyId: decision.companyId, sourceId: decision.id }); if (!updated) return 0; for (const publication of publications) publishActivity(publication); await deliverContinuation(updated.id); diff --git a/server/src/services/decision-work-notifications.ts b/server/src/services/decision-work-notifications.ts index 0abc8e95829..9e2ca840960 100644 --- a/server/src/services/decision-work-notifications.ts +++ b/server/src/services/decision-work-notifications.ts @@ -2,20 +2,22 @@ import { notifyAttentionRetentionWork } from "./retention-work-notifications.js" import type { Db } from "@paperclipai/db"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; -export async function notifyDecisionCreated(transaction: object) { - await notifyAttentionRetentionWork(transaction, { sourceKind: "decision" }); +type DecisionWorkScope = { companyId: string; sourceId?: string }; + +export async function notifyDecisionCreated(transaction: object, scope: DecisionWorkScope) { + await notifyAttentionRetentionWork(transaction, { ...scope, sourceKind: "decision" }); return notifyDeliveryWork(transaction, DELIVERY_QUEUES.decisionExpiry); } -export async function notifyDecisionChange(transaction: object) { - await notifyAttentionRetentionWork(transaction, { sourceKind: "decision" }); +export async function notifyDecisionChange(transaction: object, scope: DecisionWorkScope) { + await notifyAttentionRetentionWork(transaction, { ...scope, sourceKind: "decision" }); await notifyDeliveryWork(transaction, DELIVERY_QUEUES.decisionContinuation); await notifyDeliveryWork(transaction, DELIVERY_QUEUES.decisionExpiry); } -export function writeDecisionChange(db: Db, write: (transaction: Db) => Promise): Promise { +export function writeDecisionChange(db: Db, write: (transaction: Db) => Promise, scope: DecisionWorkScope): Promise { return db.transaction(async transaction => { - await notifyDecisionChange(transaction); + await notifyDecisionChange(transaction, scope); await notifyDeliveryWork(transaction, DELIVERY_QUEUES.decisionEffectRecovery); return write(transaction as unknown as Db); }); diff --git a/server/src/services/decisions.ts b/server/src/services/decisions.ts index 282327d433f..8a84743fcef 100644 --- a/server/src/services/decisions.ts +++ b/server/src/services/decisions.ts @@ -239,7 +239,7 @@ export function decisionService(db: Db, options: DecisionServiceOptions) { ...(input.additionalTargetSnapshots ?? {}), }; const id = randomUUID(); - await notifyDecisionCreated(dbOrTx); + await notifyDecisionCreated(dbOrTx, { companyId: input.companyId, sourceId: id }); await notifyDecisionTargetWork(dbOrTx, `decision:${id}`); const [created] = await dbOrTx.insert(decisions).values({ id, companyId: input.companyId, bundleId: input.bundleId ?? null, originAgentId: input.agentId, originIssueId: provenance.issueId, originRunId: input.runId, ruleKey: input.ruleKey ?? null, @@ -581,7 +581,8 @@ export function decisionService(db: Db, options: DecisionServiceOptions) { return attempt; } - async function settleInvalidEffectOutcome(id: string) { + async function settleInvalidEffectOutcome(decision: typeof decisions.$inferSelect) { + const id = decision.id; const publications: ActivityPublication[] = []; await writeDecisionChange(db, async tx => { const [current] = await tx.select().from(decisions).where(eq(decisions.id, id)).for("update"); @@ -601,14 +602,14 @@ export function decisionService(db: Db, options: DecisionServiceOptions) { action: "decision.recovery_failed", entityType: "decision", entityId: id, details: { reason, executionStatus, chosenOptionId: current.chosenOptionId, decidedByUserId: current.decidedByUserId }, }, publications); - }); + }, { companyId: decision.companyId, sourceId: id }); for (const publication of publications) publishActivity(publication); return outcome(id); } async function executeDecisionEffects(decision: typeof decisions.$inferSelect, userActor: AuthorizationActor) { const option = decision.options.find((item) => item.id === decision.chosenOptionId); - if (!option || !decision.decidedByUserId) return settleInvalidEffectOutcome(decision.id); + if (!option || !decision.decidedByUserId) return settleInvalidEffectOutcome(decision); const run = await db.select({ responsibleUserId: heartbeatRuns.responsibleUserId }).from(heartbeatRuns).where(eq(heartbeatRuns.id, decision.originRunId)).then((rows) => rows[0] ?? null); const metadata = decision.metadata as Record; if (metadata.kind === "attention_archive_proposal") { @@ -618,7 +619,7 @@ export function decisionService(db: Db, options: DecisionServiceOptions) { targetSnapshots: decision.targetSnapshots as Record, }), decision.signedSpec)) { await writeDecisionChange(db, async tx => tx.update(decisions).set({ executionStatus: "failed", updatedAt: new Date(), metadata: { ...metadata, archiveProposalError: "invalid_signature", ...(decision.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } }) - .where(and(eq(decisions.id, decision.id), eq(decisions.executionStatus, "running")))); + .where(and(eq(decisions.id, decision.id), eq(decisions.executionStatus, "running"))), { companyId: decision.companyId, sourceId: decision.id }); return outcome(decision.id); } if (option.id === "archive") { @@ -629,7 +630,7 @@ export function decisionService(db: Db, options: DecisionServiceOptions) { const manifestHash = manifest ? hashAttentionArchiveManifest(manifest) : null; if (!manifest || manifestHash !== metadata.manifestHash) { await writeDecisionChange(db, async tx => tx.update(decisions).set({ executionStatus: "failed", updatedAt: new Date(), metadata: { ...metadata, archiveProposalError: "manifest_mismatch", ...(decision.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } }) - .where(and(eq(decisions.id, decision.id), eq(decisions.executionStatus, "running")))); + .where(and(eq(decisions.id, decision.id), eq(decisions.executionStatus, "running"))), { companyId: decision.companyId, sourceId: decision.id }); return outcome(decision.id); } const originActor: AuthorizationActor = { @@ -651,7 +652,7 @@ export function decisionService(db: Db, options: DecisionServiceOptions) { } catch (error) { const message = error instanceof Error ? error.message : "archive_proposal_failed"; await writeDecisionChange(db, async tx => tx.update(decisions).set({ executionStatus: "failed", updatedAt: new Date(), metadata: { ...metadata, archiveProposalError: message, ...(decision.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } }) - .where(and(eq(decisions.id, decision.id), eq(decisions.executionStatus, "running")))); + .where(and(eq(decisions.id, decision.id), eq(decisions.executionStatus, "running"))), { companyId: decision.companyId, sourceId: decision.id }); return outcome(decision.id); } } @@ -659,7 +660,7 @@ export function decisionService(db: Db, options: DecisionServiceOptions) { executionStatus: "succeeded", updatedAt: new Date(), metadata: { ...metadata, ...(decision.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) }, - }).where(and(eq(decisions.id, decision.id), eq(decisions.executionStatus, "running")))); + }).where(and(eq(decisions.id, decision.id), eq(decisions.executionStatus, "running"))), { companyId: decision.companyId, sourceId: decision.id }); return outcome(decision.id); } for (let index = 0; index < option.effects.length; index += 1) await executeEffect(decision, option.effects[index]!, index, userActor, decision.decidedByUserId, run?.responsibleUserId ?? null); @@ -667,7 +668,7 @@ export function decisionService(db: Db, options: DecisionServiceOptions) { const successful = rows.filter((row) => row.status === "executed").length; const status = rows.every((row) => row.status === "executed") ? "succeeded" : successful ? "partial" : "failed"; await writeDecisionChange(db, async tx => tx.update(decisions).set({ executionStatus: status, updatedAt: new Date(), metadata: { ...decision.metadata, - ...(decision.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } }).where(and(eq(decisions.id, decision.id), eq(decisions.executionStatus, "running")))); + ...(decision.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } }).where(and(eq(decisions.id, decision.id), eq(decisions.executionStatus, "running"))), { companyId: decision.companyId, sourceId: decision.id }); return outcome(decision.id); } @@ -706,7 +707,7 @@ export function decisionService(db: Db, options: DecisionServiceOptions) { decideIdempotencyKey: input.idempotencyKey ?? null, ...(current.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}), ...(input.dismissed ? { dismissed: true, dismissReason: input.dismissReason ?? null } : {}) } }) - .where(and(eq(decisions.id, current.id), eq(decisions.status, "open"))).returning()); + .where(and(eq(decisions.id, current.id), eq(decisions.status, "open"))).returning(), { companyId: current.companyId, sourceId: current.id }); if (!claimed) throw conflict("decision_already_resolved", { code: "decision_already_resolved" }); const run = await db.select({ responsibleUserId: heartbeatRuns.responsibleUserId }).from(heartbeatRuns).where(eq(heartbeatRuns.id, claimed.originRunId)).then((rows) => rows[0] ?? null); await logActivity(db, { companyId: claimed.companyId, actorType: "system", actorId: "decision-executor", agentId: claimed.originAgentId, runId: claimed.originRunId, @@ -724,7 +725,7 @@ export function decisionService(db: Db, options: DecisionServiceOptions) { return (await get(id))!; } const [updated] = await writeDecisionChange(db, async tx => tx.update(decisions).set({ status: "cancelled", updatedAt: new Date(), metadata: { ...current.metadata, - ...(current.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } }).where(and(eq(decisions.id, id), eq(decisions.status, "open"))).returning()); + ...(current.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } }).where(and(eq(decisions.id, id), eq(decisions.status, "open"))).returning(), { companyId: current.companyId, sourceId: id }); if (!updated) throw conflict("decision_already_resolved", { code: "decision_already_resolved" }); await logActivity(db, { companyId: updated.companyId, actorType: actor.actorType, actorId: actor.actorId, runId: actor.runId, action: "decision.cancelled", entityType: "decision", entityId: id }); await continuations.deliver(updated.id); @@ -738,7 +739,7 @@ export function decisionService(db: Db, options: DecisionServiceOptions) { if (empty) return decide({ id, optionId: empty.id, decidedByUserId: userId, userActor, dismissed: true, dismissReason: reason }); const [updated] = await writeDecisionChange(db, async tx => tx.update(decisions).set({ status: "decided", executionStatus: "succeeded", chosenOptionId: "dismissed", decidedByUserId: userId, decidedAt: new Date(), updatedAt: new Date(), metadata: { ...current.metadata, dismissed: true, dismissReason: reason ?? null, - ...(current.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } }).where(and(eq(decisions.id, id), eq(decisions.status, "open"))).returning()); + ...(current.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } }).where(and(eq(decisions.id, id), eq(decisions.status, "open"))).returning(), { companyId: current.companyId, sourceId: id }); if (!updated) throw conflict("decision_already_resolved", { code: "decision_already_resolved" }); await logActivity(db, { companyId: updated.companyId, actorType: "system", actorId: "decision-executor", agentId: updated.originAgentId, runId: updated.originRunId, responsibleUserIdOverride: userId, action: "decision.dismissed", entityType: "decision", entityId: updated.id, diff --git a/server/src/services/heartbeat-run-events.ts b/server/src/services/heartbeat-run-events.ts index 22219811668..ced43f5376f 100644 --- a/server/src/services/heartbeat-run-events.ts +++ b/server/src/services/heartbeat-run-events.ts @@ -10,6 +10,8 @@ import type { Db } from "@paperclipai/db"; import { heartbeatRunEvents, heartbeatRuns } from "@paperclipai/db"; import { nativeSha256 } from "./native-runtime/canonical.js"; +export const RETRY_EXHAUSTION_MESSAGE_PREFIX = "Bounded retry exhausted"; + export interface AppendHeartbeatRunEventInput { companyId: string; runId: string; @@ -75,7 +77,7 @@ export async function appendHeartbeatRunEvent( db: Db, input: AppendHeartbeatRunEventInput, ): Promise { - const projectsRetention = input.eventType === "lifecycle" && Boolean(input.message?.startsWith("Bounded retry exhausted")); + const projectsRetention = input.eventType === "lifecycle" && Boolean(input.message?.startsWith(RETRY_EXHAUSTION_MESSAGE_PREFIX)); const projectsReassessment = readNativeEventEvidenceOutcome(input.payload?.prpEvent) !== null; const projectsProgress = isSafeNativeChatProgressEventType(input.eventType); const projectsResponse = isNativeResponseEvidence(input.eventType, input.payload?.prpEvent); @@ -131,7 +133,7 @@ export async function appendHeartbeatRunEvent( eq(heartbeatRunEvents.runId, input.runId), eq(heartbeatRunEvents.agentId, input.agentId), eq(heartbeatRunEvents.eventType, "lifecycle"), - sql`${heartbeatRunEvents.message} like 'Bounded retry exhausted%'`, + sql`${heartbeatRunEvents.message} like ${`${RETRY_EXHAUSTION_MESSAGE_PREFIX}%`}`, sql`${heartbeatRunEvents.payload} @> ${JSON.stringify(input.retryExhaustion)}::jsonb`, )) .limit(1) diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index 4ee7e681907..7239075de99 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -7598,7 +7598,7 @@ export function issueService(db: Db) { companyId: string, blockedByIssueIds: string[], actor: { agentId?: string | null; userId?: string | null } = {}, - dbOrTx: any = db, + dbOrTx: any, ) { const deduped = [...new Set(blockedByIssueIds)]; if (deduped.some((candidate) => candidate === issueId)) { @@ -9676,27 +9676,32 @@ export function issueService(db: Db) { }); if (blockParentUntilDone) { - const existingBlockers = await db - .select({ blockerIssueId: issueRelations.issueId }) - .from(issueRelations) - .where( - and( - eq(issueRelations.companyId, parent.companyId), - eq(issueRelations.relatedIssueId, parent.id), - eq(issueRelations.type, "blocks"), - ), + await db.transaction(async tx => { + await tx.select({ id: issues.id }).from(issues).where(and( + eq(issues.companyId, parent.companyId), eq(issues.id, parent.id))).for("update"); + const existingBlockers = await tx + .select({ blockerIssueId: issueRelations.issueId }) + .from(issueRelations) + .where( + and( + eq(issueRelations.companyId, parent.companyId), + eq(issueRelations.relatedIssueId, parent.id), + eq(issueRelations.type, "blocks"), + ), + ); + await syncBlockedByIssueIds( + parent.id, + parent.companyId, + [ + ...new Set([ + ...existingBlockers.map((row) => row.blockerIssueId), + child.id, + ]), + ], + { agentId: actorAgentId ?? null, userId: actorUserId ?? null }, + tx, ); - await syncBlockedByIssueIds( - parent.id, - parent.companyId, - [ - ...new Set([ - ...existingBlockers.map((row) => row.blockerIssueId), - child.id, - ]), - ], - { agentId: actorAgentId ?? null, userId: actorUserId ?? null }, - ); + }); [child] = await withIssueRelationSummaries( parent.companyId, [child],