From 2f6b52c187d228c38b6f265b648c4f0f6e7b28ba Mon Sep 17 00:00:00 2001 From: Dotta Date: Sat, 24 Oct 2026 19:29:45 -0700 Subject: [PATCH 1/2] test(server): isolate route imports and watchdog fixture identities Co-Authored-By: Paperclip --- .../issue-comment-reopen-routes.test.ts | 12 ++++-------- .../__tests__/task-watchdogs-scheduler.test.ts | 16 +++++++++++----- 2 files changed, 15 insertions(+), 13 deletions(-) diff --git a/server/src/__tests__/issue-comment-reopen-routes.test.ts b/server/src/__tests__/issue-comment-reopen-routes.test.ts index dbbef012e9f..28c6900a190 100644 --- a/server/src/__tests__/issue-comment-reopen-routes.test.ts +++ b/server/src/__tests__/issue-comment-reopen-routes.test.ts @@ -2,7 +2,10 @@ import express from "express"; import request from "supertest"; import { getTableName } from "drizzle-orm"; import { beforeEach, describe, expect, it, vi } from "vitest"; -import { HttpError } from "../errors.js"; +import { HttpError, unprocessable } from "../errors.js"; +import { normalizeIssueExecutionPolicy } from "../services/issue-execution-policy.js"; +import { issueRoutes } from "../routes/issues.js"; +import { errorHandler } from "../middleware/index.js"; vi.mock("../services/fast-responses.js", () => ({ enqueueFastResponse: vi.fn(async () => undefined), @@ -233,10 +236,6 @@ async function installActor( app: express.Express, actor?: Record, ) { - const [{ issueRoutes }, { errorHandler }] = await Promise.all([ - import("../routes/issues.js"), - import("../middleware/index.js"), - ]); app.use((req, _res, next) => { (req as any).actor = actor ?? { type: "board", @@ -261,8 +260,6 @@ async function normalizePolicy(input: { >; }>; }) { - const { normalizeIssueExecutionPolicy } = - await import("../services/issue-execution-policy.js"); return normalizeIssueExecutionPolicy(input); } @@ -3666,7 +3663,6 @@ describe("issue comment reopen routes", () => { authorAgentId: reviewerAgentId, authorUserId: null, }); - const { unprocessable } = await import("../errors.js"); mockIssueService.update.mockRejectedValue( unprocessable("Issue can only have one assignee"), ); diff --git a/server/src/__tests__/task-watchdogs-scheduler.test.ts b/server/src/__tests__/task-watchdogs-scheduler.test.ts index 1124c163eb2..a687d6597dd 100644 --- a/server/src/__tests__/task-watchdogs-scheduler.test.ts +++ b/server/src/__tests__/task-watchdogs-scheduler.test.ts @@ -28,7 +28,10 @@ import { } from "./helpers/embedded-postgres.js"; import { taskWatchdogService } from "../services/task-watchdogs.ts"; -const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; + +const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_WATCHDOG_TEST_DATABASE_URL"); +const embeddedPostgresSupport = externalUrl ? { supported: true, reason: undefined } : await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; if (!embeddedPostgresSupport.supported) { @@ -42,8 +45,8 @@ describeEmbeddedPostgres("task watchdog scheduler", () => { let tempDb: Awaited> | null = null; beforeAll(async () => { - tempDb = await startEmbeddedPostgresTestDatabase("paperclip-task-watchdogs-scheduler-"); - db = createDb(tempDb.connectionString); + if (!externalUrl) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-task-watchdogs-scheduler-"); + db = createDb(externalUrl ?? tempDb!.connectionString); }, 20_000); afterEach(async () => { @@ -64,6 +67,7 @@ describeEmbeddedPostgres("task watchdog scheduler", () => { }); afterAll(async () => { + await db?.$client.end({ timeout: 1 }); await tempDb?.cleanup(); }); @@ -96,16 +100,18 @@ describeEmbeddedPostgres("task watchdog scheduler", () => { return id; } + let nextFixtureIssueNumber = 10_000; async function seedIssue(companyId: string, overrides: Partial = {}) { const id = overrides.id ?? randomUUID(); + const issueNumber = overrides.issueNumber ?? ++nextFixtureIssueNumber; await db.insert(issues).values({ id, companyId, title: overrides.title ?? "Watched issue", status: overrides.status ?? "done", priority: overrides.priority ?? "medium", - identifier: overrides.identifier ?? `WDOG-${Math.floor(Math.random() * 10_000)}`, - issueNumber: overrides.issueNumber ?? Math.floor(Math.random() * 10_000), + identifier: overrides.identifier ?? `WDOG-${issueNumber}`, + issueNumber, parentId: overrides.parentId, assigneeAgentId: overrides.assigneeAgentId, originKind: overrides.originKind, From 400f4c0fc8c23c84c0c05108f7ebc4934c352ce2 Mon Sep 17 00:00:00 2001 From: Dotta Date: Sat, 24 Oct 2026 23:05:41 -0500 Subject: [PATCH 2/2] fix(server): scope approval queue wakes to their company Co-Authored-By: Paperclip --- doc/idle-approval-queue-seeding.md | 30 ++++++-- ...2026-10-24-recurring-server-sleep-stack.md | 15 ++++ .../approval-queue-seeding-work.test.ts | 68 ++++++++++++++++++- server/src/lib/approval-records.ts | 2 +- .../services/approval-queue-seeding-work.ts | 54 +++++++++++---- .../services/approval-work-notifications.ts | 10 +-- server/src/services/approvals.ts | 10 +-- server/src/services/budgets.ts | 10 +-- server/src/services/companies.ts | 4 +- server/src/services/company-deletion.ts | 2 +- server/src/services/decision-queues.ts | 6 +- server/src/services/issue-approvals.ts | 8 +-- server/src/services/issues.ts | 5 +- .../services/retention-work-notifications.ts | 4 +- server/src/services/routines.ts | 2 +- server/src/services/tool-gateway.ts | 4 +- server/src/services/work-products.ts | 12 ++-- 17 files changed, 191 insertions(+), 55 deletions(-) diff --git a/doc/idle-approval-queue-seeding.md b/doc/idle-approval-queue-seeding.md index b724522a12e..7f7e967d837 100644 --- a/doc/idle-approval-queue-seeding.md +++ b/doc/idle-approval-queue-seeding.md @@ -67,18 +67,21 @@ unscoped maintenance feed. No new deadline is inferred from those display states ## Worker lifecycle -Startup reconstructs missing memberships. A committed producer wakes the shared -coordinator. Each sweep processes at most 100 approvals, with an ID cursor that +Startup reconstructs missing memberships across the instance. Each later committed +producer carries its company ID and wakes only that company's reconciliation. +Pending scopes coalesce in memory; one company's ordinary mutations cannot restart +a global scan. Scope rotation prevents one failing company from starving another. Each sweep processes at most 100 approvals, with an ID cursor that advances past failures and rotates to revisit them. Each row is rechecked under an approval lock before the queue writer runs. One bad row cannot permanently hide later candidates. Atomic existing membership uniqueness prevents duplicate rows and duplicate insertion audit events. After a sweep, another lookup decides whether actual projection debt remains. -Only that debt or a failed attempt retains a one-second retry. Once reconciled, -there is no deadline and no recurring worker SQL. The shared deadline worker -preserves requested work during drain, does not run passive domain SQL while -paused, and reconstructs from rows after restart. Shutdown aborts between rows +Only that debt, a newer committed signal, or a failed attempt retains a one-second +retry. The generation check follows the final awaited lookup, so a commit arriving +during an empty read cannot be erased. Once reconciled, there is no deadline and +no recurring worker SQL. The shared coordinator preserves requested work during +drain. Passive pending checks inspect memory only; startup reconstructs from rows. Shutdown aborts between rows and waits for an admitted transaction; unprocessed rows remain recoverable. Registration follows the existing heartbeat-maintenance enablement switch. @@ -118,3 +121,18 @@ membership changes. A two-connection PostgreSQL regression reproduced `deadlock detected`. Manual add/remove now lock the queue before touching its items, matching the seed writer. The regression and the existing queue route suite pass (27 cases). Server typecheck and build pass after the repair. Fresh-head CI/review is required. + +Security review found that normal producer wakes could restart a cross-company +scan. The regression reproduced another company's queue being populated by an +unrelated company write. Producers now include company hints, including batch +imports and nested transactions. A separate controlled PostgreSQL test holds the +final empty read while a real producer commits and verifies the retained wake. + +After company scoping, all 169 selected cases pass across approval queue seeding, +approval retention, queue routes, approval services, work products, and issue +comment routes. Server typechecking passes. The preceding run suffered timeouts +and a consequent cleanup foreign-key error; it is not counted as passing. The +route suite separately exposed cold imports inside timed tests. Moving those +imports to suite loading preserves assertions and prevents timed-out requests +from leaking into later mock state. The watchdog fixture collision from CI uses +monotonic identifiers instead; its 19 real-PostgreSQL cases pass. diff --git a/doc/plans/2026-10-24-recurring-server-sleep-stack.md b/doc/plans/2026-10-24-recurring-server-sleep-stack.md index ca388a5f3b9..eae91729127 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -931,3 +931,18 @@ after its single justified retry. The duplicate cancelled workflows for #17490 and #17497 were each rerun once to clear stale cancelled contexts. The #17490 retry has a new E2E shard 3 failure under investigation; do not call it green or rerun again without a repair. #17497's retry was still running at this observation. + + +Slice 16 security review reproduced cross-company work amplification from normal +producer writes. The repair carries company hints through all supported producers +and maintains separate pending scopes. Startup recovery remains instance-wide. +A generation comparison after the final awaited read preserves concurrent commits. +All 169 affected cases and server typechecking pass. CI fixture repairs move cold +route imports outside timed tests and replace random watchdog identifiers. The +initial rerun had timeouts and a cleanup failure; the final selected run passes +without relaxing timeouts. Fresh remote CI/review remains required. + +Plugin scheduling is published as #17518 at 8b4f551dc2. Its Actions checks passed, +but review found retained completion writes missing from shutdown and a stale +deadline update race. Repair those and preserve independent job progress when +one completion write fails before treating the slice as ready. diff --git a/server/src/__tests__/approval-queue-seeding-work.test.ts b/server/src/__tests__/approval-queue-seeding-work.test.ts index 375dcb724b9..f3961b2c9e1 100644 --- a/server/src/__tests__/approval-queue-seeding-work.test.ts +++ b/server/src/__tests__/approval-queue-seeding-work.test.ts @@ -1,7 +1,7 @@ import { randomUUID } from "node:crypto"; import { eq, sql } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; -import { activityLog, approvals, companies, createDb, decisionQueueItems, decisionQueues, decisionRetention, decisionTriageEvents, issueApprovals, issues, issueWorkProducts, type Db } from "@paperclipai/db"; +import { activityLog, approvals, companies, createDb, decisionQueueItems, decisionQueues, decisionRetention, decisionTriageEvents, issueApprovals, issues, issueWorkProducts, subscribeDatabaseWork, type Db } from "@paperclipai/db"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { approvalRecords } from "../services/approvals.js"; @@ -14,6 +14,7 @@ import { issueApprovalService } from "../services/issue-approvals.js"; import { issueService } from "../services/issues.js"; import { workProductService } from "../services/work-products.js"; import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js"; import { idleWorkSnapshot } from "../services/task-admission.js"; const external = resolveExternalTestDatabaseUrl("PAPERCLIP_APPROVAL_QUEUE_TEST_DATABASE_URL"); @@ -89,6 +90,71 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu reads.mockRestore(); writes.mockRestore(); executes.mockRestore(); } + it("retains a company commit arriving during its final empty read", async () => { + const service = approvalQueueSeedingService(db); + await service.sweep(); + expect(await service.next()).toBeNull(); + const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.approvalQueueSeeding, (event, hints) => { + if (event === "settled") service.request(hints); + }); + const select = db.select.bind(db); + let release!: () => void, observed!: () => void; + const observedEmpty = new Promise(resolve => { observed = resolve; }); + const barrier = new Promise(resolve => { release = resolve; }); + let intercepted = false; + const reads = vi.spyOn(db, "select").mockImplementation((...args: any[]) => { + const query = (select as any)(...args); + const from = query.from.bind(query); + query.from = (...tables: any[]) => { + const builder = from(...tables); + const limit = builder.limit.bind(builder); + builder.limit = (count: number) => { + const result = limit(count); + if (count === 1 && !intercepted) { + intercepted = true; + const then = result.then.bind(result); + result.then = (resolve: (rows: unknown[]) => unknown, reject: (err: unknown) => unknown) => + then(async (rows: unknown[]) => { observed(); await barrier; return resolve(rows); }, reject); + } + return result; + }; + return builder; + }; + return query; + }); + service.request(new Set([companyId])); + const sweep = service.sweep(); + try { + await observedEmpty; + const approval = await pending(); + release(); await sweep; reads.mockRestore(); + expect(await service.next()).not.toBeNull(); + await service.sweep(); + expect(await memberships()).toEqual([expect.objectContaining({ sourceId: approval.id })]); + expect(await service.next()).toBeNull(); + } finally { release(); await sweep; reads.mockRestore(); unsubscribe(); } + }); + + it("does not let a company write reconcile another company's approvals", async () => { + await start(); await quiet(); + const foreignId = randomUUID(); + await db.insert(companies).values({ id: foreignId, name: "Other tenant", issuePrefix: `F${foreignId.slice(0, 6)}` }); + const [foreignIssue] = await db.insert(issues).values({ companyId: foreignId, title: "Other PR", status: "todo" }).returning(); + const [foreignApproval] = await db.insert(approvals).values({ companyId: foreignId, type: "request_board_approval", payload: {} }).returning(); + await db.insert(issueApprovals).values({ companyId: foreignId, issueId: foreignIssue.id, approvalId: foreignApproval.id }); + await db.insert(issueWorkProducts).values({ companyId: foreignId, issueId: foreignIssue.id, type: "pull_request", provider: "github", title: "Other PR", status: "open" }); + const ownApproval = await pending(); await seeded(ownApproval.id); await quiet(); + expect(await db.select().from(decisionQueueItems).where(eq(decisionQueueItems.companyId, foreignId))).toEqual([]); + await companyService(db).update(foreignId, { status: "paused" }); await settle(); + await companyService(db).update(foreignId, { status: "active" }); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(await db.select().from(decisionQueueItems).where(eq(decisionQueueItems.companyId, foreignId))) + .toEqual([expect.objectContaining({ sourceId: foreignApproval.id })]); + }); + await quiet(); + }); + it("populates the PR queue without an attention reader after approvals leave retention maintenance", async () => { await start(); await quiet(); const approval = await pending(); diff --git a/server/src/lib/approval-records.ts b/server/src/lib/approval-records.ts index aaecd65bfda..32f4d6f384c 100644 --- a/server/src/lib/approval-records.ts +++ b/server/src/lib/approval-records.ts @@ -38,7 +38,7 @@ export function approvalResolutionRecords(db: Db) { } const now = new Date(); - const updated = await writeApprovalChange(database, tx => tx + const updated = await writeApprovalChange(database, existing.companyId, tx => tx .update(approvals) .set({ status: targetStatus, diff --git a/server/src/services/approval-queue-seeding-work.ts b/server/src/services/approval-queue-seeding-work.ts index bf0c0358e0a..86cb4bec400 100644 --- a/server/src/services/approval-queue-seeding-work.ts +++ b/server/src/services/approval-queue-seeding-work.ts @@ -3,7 +3,6 @@ import { approvals, companies, issues, type Db } from "@paperclipai/db"; import { executionIssueCondition } from "./issue-visibility.js"; import { decisionQueueService } from "./decision-queues.js"; import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; -import { registerDeadlineWork } from "./deadline-work.js"; import { DELIVERY_QUEUES } from "./delivery-work-notifications.js"; const linkedIssue = sql`(select linked.issue_id from issue_approvals linked @@ -11,11 +10,18 @@ const linkedIssue = sql`(select linked.issue_id from issue_approvals lin order by linked.issue_id limit 1)`; export function approvalQueueSeedingService(db: Db) { - let cursor: string | undefined; - function candidates(database: Db, condition?: SQL) { + const pending = new Map([[undefined, { generation: 0 }]]); + function request(companyIds: ReadonlySet | undefined) { + for (const companyId of companyIds ?? []) { + const existing = pending.get(companyId); + if (existing) existing.generation++; + else pending.set(companyId, { generation: 0 }); + } + } + function candidates(database: Db, companyId: string | undefined, condition?: SQL) { return database.select({ id: approvals.id, companyId: approvals.companyId, issueId: linkedIssue }).from(approvals) .innerJoin(companies, eq(companies.id, approvals.companyId)) - .where(and(eq(approvals.status, "pending"), eq(companies.status, "active"), condition, + .where(and(eq(approvals.status, "pending"), eq(companies.status, "active"), companyId === undefined ? undefined : eq(approvals.companyId, companyId), condition, sql`exists (select 1 from ${issues} where ${issues.id} = ${linkedIssue} and ${issues.companyId} = ${approvals.companyId} and ${executionIssueCondition()} and exists (select 1 from issue_work_products product where product.company_id = ${approvals.companyId} @@ -26,30 +32,41 @@ export function approvalQueueSeedingService(db: Db) { )).orderBy(asc(approvals.id)); } async function sweep(signal?: AbortSignal) { - const rows = await candidates(db, cursor ? gt(approvals.id, cursor) : undefined).limit(100); + const scope = pending.entries().next().value; + if (!scope) return; + const [companyId, state] = scope; + const generation = state.generation; + const rows = await candidates(db, companyId, state.cursor ? gt(approvals.id, state.cursor) : undefined).limit(100); const errors: unknown[] = []; for (const row of rows) { if (signal?.aborted) break; try { await db.transaction(async tx => { const database = tx as unknown as Db; - const [current] = await candidates(database, eq(approvals.id, row.id)).for("update", { of: approvals }); + const [current] = await candidates(database, row.companyId, eq(approvals.id, row.id)).for("update", { of: approvals }); if (!current) return; await decisionQueueService(database).materializeSeedSources(current.companyId, [{ sourceKind: "approval", sourceId: current.id, issueId: current.issueId, }]); }); } catch (error) { errors.push(error); } - cursor = row.id; + state.cursor = row.id; + } + if (rows.length < 100) state.cursor = undefined; + const remaining = await candidates(db, companyId).limit(1); + if (!signal?.aborted && generation === state.generation && remaining.length === 0) { + pending.delete(companyId); + } else { + pending.delete(companyId); + pending.set(companyId, state); } - if (rows.length < 100) cursor = undefined; if (errors.length === 1) throw errors[0]; if (errors.length) throw new AggregateError(errors, "Approval queue seeding failed"); } async function next() { - return (await candidates(db).limit(1)).length ? Date.now() : null; + return pending.size ? Date.now() : null; } - return { sweep, next }; + return { sweep, next, request }; } export function registerApprovalQueueSeedingWork( @@ -57,7 +74,20 @@ export function registerApprovalQueueSeedingWork( service: ReturnType, canRun: () => boolean, ) { - return registerDeadlineWork(coordinator, DELIVERY_QUEUES.approvalQueueSeeding, { - run: signal => service.sweep(signal), next: service.next, canRun, retryMs: 1_000, + let earliest = 0; + let next: number | null = null; + return coordinator.register(DELIVERY_QUEUES.approvalQueueSeeding, { + retryMs: 1_000, + pausePassiveMaintenance: true, + onSettled: service.request, + run: async signal => { + if (!canRun()) return; + if (Date.now() < earliest) { next = earliest; return; } + earliest = Date.now() + 1_000; + await service.sweep(signal); + next = await service.next(); + }, + hasPending: async () => (await service.next()) !== null, + nextRunAt: () => next, }); } diff --git a/server/src/services/approval-work-notifications.ts b/server/src/services/approval-work-notifications.ts index 39d8e9fd03e..5c914477d15 100644 --- a/server/src/services/approval-work-notifications.ts +++ b/server/src/services/approval-work-notifications.ts @@ -1,16 +1,16 @@ import type { Db } from "@paperclipai/db"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; -export const notifyApprovalQueueWork = (transaction: object) => notifyDeliveryWork(transaction, DELIVERY_QUEUES.approvalQueueSeeding); +export const notifyApprovalQueueWork = (transaction: object, companyId: string) => notifyDeliveryWork(transaction, DELIVERY_QUEUES.approvalQueueSeeding, companyId); -export async function notifyApprovalWork(transaction: object) { +export async function notifyApprovalWork(transaction: object, companyId: string) { await notifyDeliveryWork(transaction, DELIVERY_QUEUES.approvalRetention); - await notifyApprovalQueueWork(transaction); + await notifyApprovalQueueWork(transaction, companyId); } -export function writeApprovalChange(db: Db, write: (transaction: Db) => Promise): Promise { +export function writeApprovalChange(db: Db, companyId: string, write: (transaction: Db) => Promise): Promise { return db.transaction(async tx => { - await notifyApprovalWork(tx); + await notifyApprovalWork(tx, companyId); return write(tx as unknown as Db); }); } diff --git a/server/src/services/approvals.ts b/server/src/services/approvals.ts index b806e112368..7110b94261e 100644 --- a/server/src/services/approvals.ts +++ b/server/src/services/approvals.ts @@ -54,7 +54,7 @@ export function approvalRecords(db: Db) { }, create: (companyId: string, data: Omit) => - writeApprovalChange(db, tx => tx + writeApprovalChange(db, companyId, tx => tx .insert(approvals) .values({ ...data, companyId }) .returning() @@ -64,8 +64,10 @@ export function approvalRecords(db: Db) { // decision — e.g. when its paired agent is terminated during duplicate // cleanup. Idempotent: a no-op on already-resolved approvals. cancel: async (id: string, reason?: string | null) => { + const [existing] = await db.select({ companyId: approvals.companyId }).from(approvals).where(eq(approvals.id, id)); + if (!existing) return null; const now = new Date(); - const updated = await writeApprovalChange(db, tx => tx + const updated = await writeApprovalChange(db, existing.companyId, tx => tx .update(approvals) .set({ status: "cancelled", @@ -86,7 +88,7 @@ export function approvalRecords(db: Db) { } const now = new Date(); - return writeApprovalChange(db, tx => tx + return writeApprovalChange(db, existing.companyId, tx => tx .update(approvals) .set({ status: "revision_requested", @@ -107,7 +109,7 @@ export function approvalRecords(db: Db) { } const now = new Date(); - return writeApprovalChange(db, tx => tx + return writeApprovalChange(db, existing.companyId, tx => tx .update(approvals) .set({ status: "pending", diff --git a/server/src/services/budgets.ts b/server/src/services/budgets.ts index 0ce6009fce2..ccff957a9ed 100644 --- a/server/src/services/budgets.ts +++ b/server/src/services/budgets.ts @@ -260,7 +260,9 @@ async function markApprovalStatus( decidedByUserId: string, ) { if (!approvalId) return; - await notifyApprovalWork(db); + const [approval] = await db.select({ companyId: approvals.companyId }).from(approvals).where(eq(approvals.id, approvalId)); + if (!approval) return; + await notifyApprovalWork(db, approval.companyId); await db .update(approvals) .set({ @@ -298,7 +300,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica const scope = await resolveScopeRecord(db, "company", policy.scopeId); if (scope.paused) return; - await notifyCompanyRetentionWork(db); + await notifyCompanyRetentionWork(db, policy.companyId); await notifyDeliveryWork(db, DELIVERY_QUEUES.agentLifecycle, policy.companyId); await notifyDeliveryWork(db, DELIVERY_QUEUES.skillSourceSync); await db @@ -355,7 +357,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica await notifyDeliveryWork(db, DELIVERY_QUEUES.skillSourceSync); const scope = await resolveScopeRecord(db, "company", policy.scopeId); if (scope.pauseReason !== "budget") return; - await notifyCompanyRetentionWork(db); + await notifyCompanyRetentionWork(db, policy.companyId); await notifyDeliveryWork(db, DELIVERY_QUEUES.agentLifecycle, policy.companyId); await db .update(companies) @@ -458,7 +460,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica windowEnd: end, }); - if (thresholdType === "hard") await notifyApprovalWork(db); + if (thresholdType === "hard") await notifyApprovalWork(db, policy.companyId); const approval = thresholdType === "hard" ? await db .insert(approvals) diff --git a/server/src/services/companies.ts b/server/src/services/companies.ts index 4f17b854339..580df76b6e9 100644 --- a/server/src/services/companies.ts +++ b/server/src/services/companies.ts @@ -331,7 +331,7 @@ export function companyService(db: Db, _budgetHooks: BudgetServiceHooks = {}) { if (companyPatch.status !== undefined && companyPatch.status !== existing.status) { await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync); - await notifyCompanyRetentionWork(tx); + await notifyCompanyRetentionWork(tx, id); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentLifecycle, id); } const updated = await tx @@ -483,7 +483,7 @@ export function companyService(db: Db, _budgetHooks: BudgetServiceHooks = {}) { if (!wasAlreadyArchived) { await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync); - await notifyCompanyRetentionWork(tx); + await notifyCompanyRetentionWork(tx, id); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentLifecycle, id); await tx .update(companies) diff --git a/server/src/services/company-deletion.ts b/server/src/services/company-deletion.ts index 44eed1a462e..b48a982cc22 100644 --- a/server/src/services/company-deletion.ts +++ b/server/src/services/company-deletion.ts @@ -95,7 +95,7 @@ export async function deleteCompany(db: Db, id: string) { await tx.delete(agentRuntimeState).where(eq(agentRuntimeState.companyId, id)); await tx.delete(issueComments).where(eq(issueComments.companyId, id)); await tx.delete(approvalComments).where(eq(approvalComments.companyId, id)); - await notifyCompanyRetentionWork(tx); + await notifyCompanyRetentionWork(tx, id); await tx.delete(approvals).where(eq(approvals.companyId, id)); await tx.delete(companySecrets).where(eq(companySecrets.companyId, id)); await tx.delete(joinRequests).where(eq(joinRequests.companyId, id)); diff --git a/server/src/services/decision-queues.ts b/server/src/services/decision-queues.ts index 1468d0444cc..e89ac14ace6 100644 --- a/server/src/services/decision-queues.ts +++ b/server/src/services/decision-queues.ts @@ -505,7 +505,7 @@ export function decisionQueueService(db: Db) { }) => { const result = await db.transaction(async (tx) => { const txDb = tx as unknown as Db; - if (input.key === "prs") await notifyApprovalQueueWork(txDb); + if (input.key === "prs") await notifyApprovalQueueWork(txDb, input.companyId); const inserted = await txDb.insert(decisionQueues).values({ companyId: input.companyId, key: input.key, @@ -563,7 +563,7 @@ export function decisionQueueService(db: Db) { }) => { const row = await db.transaction(async (tx) => { const txDb = tx as unknown as Db; - if (input.key === "prs") await notifyApprovalQueueWork(txDb); + if (input.key === "prs") await notifyApprovalQueueWork(txDb, input.companyId); const updated = await txDb.update(decisionQueues).set({ ...input.patch, updatedAt: new Date() }) .where(and(eq(decisionQueues.companyId, input.companyId), eq(decisionQueues.key, input.key))) .returning().then((rows) => rows[0] ?? null); @@ -673,7 +673,7 @@ export function decisionQueueService(db: Db) { return db.transaction(async (tx) => { const txDb = tx as unknown as Db; await lockQueue(txDb, input.companyId, queue.id); - if (input.key === "prs" && input.sourceKind === "approval") await notifyApprovalQueueWork(txDb); + if (input.key === "prs" && input.sourceKind === "approval") await notifyApprovalQueueWork(txDb, input.companyId); const removed = await txDb.delete(decisionQueueItems).where(and( eq(decisionQueueItems.companyId, input.companyId), eq(decisionQueueItems.queueId, queue.id), diff --git a/server/src/services/issue-approvals.ts b/server/src/services/issue-approvals.ts index 993355ebaeb..682262ca899 100644 --- a/server/src/services/issue-approvals.ts +++ b/server/src/services/issue-approvals.ts @@ -108,7 +108,7 @@ export function issueApprovalService(db: Db) { link: async (issueId: string, approvalId: string, actor?: LinkActor) => { const { issue } = await assertIssueAndApprovalSameCompany(issueId, approvalId); - await writeApprovalChange(db, tx => tx + await writeApprovalChange(db, issue.companyId, tx => tx .insert(issueApprovals) .values({ companyId: issue.companyId, @@ -127,8 +127,8 @@ export function issueApprovalService(db: Db) { }, unlink: async (issueId: string, approvalId: string) => { - await assertIssueAndApprovalSameCompany(issueId, approvalId); - await writeApprovalChange(db, tx => tx + const { issue } = await assertIssueAndApprovalSameCompany(issueId, approvalId); + await writeApprovalChange(db, issue.companyId, tx => tx .delete(issueApprovals) .where(and(eq(issueApprovals.issueId, issueId), eq(issueApprovals.approvalId, approvalId)))); }, @@ -158,7 +158,7 @@ export function issueApprovalService(db: Db) { } } - await writeApprovalChange(db, tx => tx + await writeApprovalChange(db, approval.companyId, tx => tx .insert(issueApprovals) .values( uniqueIssueIds.map((issueId) => ({ diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index b6318522947..2964c3a122a 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -11302,7 +11302,7 @@ export function issueService(db: Db) { await notifyDecisionTargetWork(tx, `issue:${id}`); } if (patch.hiddenAt !== undefined || patch.harnessKind !== undefined || patch.conversationAgentId !== undefined) { - await notifyApprovalWork(tx); + await notifyApprovalWork(tx, receiptExisting.companyId); } const updated = await tx .update(issues) @@ -11711,7 +11711,8 @@ export function issueService(db: Db) { try { await notifyTerminalWorkspaceWork(tx, `issue:${id}`); await requestRestorePointDeletion(tx, { sourceIssueId: id }); - await notifyApprovalWork(tx); + const [owner] = await tx.select({ companyId: issues.companyId }).from(issues).where(eq(issues.id, id)); + if (owner) await notifyApprovalWork(tx, owner.companyId); await notifyDecisionTargetWork(tx); removedIssue = await tx .delete(issues) diff --git a/server/src/services/retention-work-notifications.ts b/server/src/services/retention-work-notifications.ts index e42efd359bd..ffedb0b1a4d 100644 --- a/server/src/services/retention-work-notifications.ts +++ b/server/src/services/retention-work-notifications.ts @@ -8,8 +8,8 @@ export async function notifyRetentionWork(transaction: object, kind: AttentionSo if (kind === "join_request") await notifyDeliveryWork(transaction, DELIVERY_QUEUES.joinRequestRetention); } -export async function notifyCompanyRetentionWork(transaction: object) { - await notifyApprovalWork(transaction); +export async function notifyCompanyRetentionWork(transaction: object, companyId: string) { + await notifyApprovalWork(transaction, companyId); await notifyRetentionWork(transaction, "join_request"); } diff --git a/server/src/services/routines.ts b/server/src/services/routines.ts index 78a71cbbdc3..5240a1b5142 100644 --- a/server/src/services/routines.ts +++ b/server/src/services/routines.ts @@ -1966,7 +1966,7 @@ export function routineService( return updated ?? createdRun; } catch (error) { if (createdIssue) { - await notifyApprovalWork(txDb); + await notifyApprovalWork(txDb, createdIssue.companyId); await notifyDecisionTargetWork(txDb); await txDb.delete(issues).where(eq(issues.id, createdIssue.id)); } diff --git a/server/src/services/tool-gateway.ts b/server/src/services/tool-gateway.ts index 1a20945f14c..9f15e0a3be0 100644 --- a/server/src/services/tool-gateway.ts +++ b/server/src/services/tool-gateway.ts @@ -2578,7 +2578,7 @@ export function createToolGatewayService( let formalApprovalId: string | null = null; if (toolRequiresFormalApproval(input.tool)) { - const [approval] = await writeApprovalChange(db, tx => tx + const [approval] = await writeApprovalChange(db, input.session.companyId, tx => tx .insert(approvals) .values({ companyId: input.session.companyId, @@ -2604,7 +2604,7 @@ export function createToolGatewayService( .returning()); formalApprovalId = approval.id; const approvalIssueId = input.session.issueId; - await writeApprovalChange(db, tx => tx + await writeApprovalChange(db, input.session.companyId, tx => tx .insert(issueApprovals) .values({ companyId: input.session.companyId, diff --git a/server/src/services/work-products.ts b/server/src/services/work-products.ts index a041b7a6d8f..4aa44ef7f49 100644 --- a/server/src/services/work-products.ts +++ b/server/src/services/work-products.ts @@ -276,7 +276,7 @@ export function workProductService( const row = await db.transaction(async (tx) => { if (data.type === "pull_request") { await notifyTerminalWorkspaceWork(tx, `issue:${issueId}`); - await notifyApprovalQueueWork(tx); + await notifyApprovalQueueWork(tx, companyId); } await validateExecutionWorkspace(tx, companyId, data.executionWorkspaceId); if (data.isPrimary) { @@ -315,7 +315,7 @@ export function workProductService( if (existing.type === "pull_request" || patch.type === "pull_request") { await notifyTerminalWorkspaceWork(tx, `issue:${existing.issueId}`); - await notifyApprovalQueueWork(tx); + await notifyApprovalQueueWork(tx, existing.companyId); } await validateExecutionWorkspace(tx, existing.companyId, patch.executionWorkspaceId); @@ -385,7 +385,9 @@ export function workProductService( // prefix behind (which a retry would then duplicate). Mirrors the // per-writer transaction the batched issue/document writers use. await db.transaction(async (tx) => { - if (values.some(value => value.type === "pull_request")) await notifyApprovalQueueWork(tx); + for (const companyId of new Set(values.filter(value => value.type === "pull_request").map(value => value.companyId))) { + await notifyApprovalQueueWork(tx, companyId); + } for (const value of values) { if (value.type === "pull_request") await notifyTerminalWorkspaceWork(tx, `issue:${value.issueId}`); } @@ -395,11 +397,11 @@ export function workProductService( remove: async (id: string) => { const row = await db.transaction(async tx => { - const [existing] = await tx.select({ type: issueWorkProducts.type, issueId: issueWorkProducts.issueId }) + const [existing] = await tx.select({ type: issueWorkProducts.type, issueId: issueWorkProducts.issueId, companyId: issueWorkProducts.companyId }) .from(issueWorkProducts).where(eq(issueWorkProducts.id, id)); if (existing?.type === "pull_request") { await notifyTerminalWorkspaceWork(tx, `issue:${existing.issueId}`); - await notifyApprovalQueueWork(tx); + await notifyApprovalQueueWork(tx, existing.companyId); } return tx.delete(issueWorkProducts).where(eq(issueWorkProducts.id, id)) .returning().then(rows => rows[0] ?? null);