From e04a0375dc0ec98324956e7e6d6df2a81e2b6cca Mon Sep 17 00:00:00 2001 From: Dotta Date: Sat, 24 Oct 2026 15:56:37 -0700 Subject: [PATCH] fix(server): wake approval queue seeding from committed changes Co-Authored-By: Paperclip --- doc/idle-approval-queue-seeding.md | 114 +++++++ ...2026-10-24-recurring-server-sleep-stack.md | 26 ++ .../approval-queue-seeding-work.test.ts | 293 ++++++++++++++++++ .../__tests__/decision-queues-routes.test.ts | 9 +- .../server-startup-feedback-export.test.ts | 18 ++ server/src/index.ts | 3 + server/src/lib/approval-records.ts | 2 +- .../services/approval-queue-seeding-work.ts | 63 ++++ .../services/approval-work-notifications.ts | 16 + server/src/services/approvals.ts | 2 +- server/src/services/budgets.ts | 7 +- server/src/services/decision-queues.ts | 214 +++++++------ .../services/delivery-work-notifications.ts | 1 + server/src/services/issue-approvals.ts | 2 +- server/src/services/issues.ts | 6 +- .../services/retention-work-notifications.ts | 10 +- server/src/services/routines.ts | 4 +- server/src/services/tool-gateway.ts | 2 +- server/src/services/work-products.ts | 17 +- 19 files changed, 687 insertions(+), 122 deletions(-) create mode 100644 doc/idle-approval-queue-seeding.md create mode 100644 server/src/__tests__/approval-queue-seeding-work.test.ts create mode 100644 server/src/services/approval-queue-seeding-work.ts create mode 100644 server/src/services/approval-work-notifications.ts diff --git a/doc/idle-approval-queue-seeding.md b/doc/idle-approval-queue-seeding.md new file mode 100644 index 00000000000..14642e8ed97 --- /dev/null +++ b/doc/idle-approval-queue-seeding.md @@ -0,0 +1,114 @@ +# Approval queue seeding without retention polling + +## Obligation and stack dependency + +The attention maintenance scan does two different jobs: retention and automatic +queue membership. Its call to `materializeSeededQueues` adds pending approvals to +the PR queue when the first linked issue has a pull-request work product. Queue +API readers use saved membership; they do not first build the attention feed. +Removing approvals from maintenance therefore lost automatic queue population. +A real PostgreSQL regression reproduced an empty queue with a pending linked +approval and PR work product, without an attention reader. + +This slice owns automatic PR-queue membership for approvals. It depends on the +approval producer boundaries introduced in #17497 and is required with that +retention migration before treating the combined behavior as complete. Join +requests have no issue association or matching seed rule. Other attention-source +seed discovery remains in the existing feed scan and must migrate with an +explicit owner before those sources leave that scan. + +## Source of truth and ownership + +Existing approvals, issue approvals, issue work products, queues, and queue items +are the only durable state. No new table, outbox, external API, or provider branch +is introduced. `approvalQueueSeedingService` selects missing memberships for: + +- A pending approval in an active company. +- Its first linked issue, ordered by issue ID, in the same company and eligible + under the attention feed's execution-issue visibility predicate. +- A pull-request work product belonging to that issue and company. +- A missing PR queue or an enabled seed rule without this approval's membership. + +Disabled rules, nonmatching approvals, completed approvals, and existing seeded +history are passive. A pending approval itself prevents its linked Slack/Speko +conversation from projecting as idle, so this source does not depend on every +other conversation-state producer. Hidden/harness/conversation-container issue +changes retain their explicit signals. + +The queue service owns seed matching, creation, membership uniqueness, and audit +writes. The feed and worker use the same writer through a small normalized source +contract. A worker does not construct a presentation-layer AttentionItem. Queue +reads keep their existing source authorization checks. A queue-rule update and +seeding serialize on the existing queue row. Existing memberships are not pruned +when an approval resolves. This preserves current behavior; pruning is a separate +change, including the proposal in #13364. + +## Producer audit + +| Producer | Commit boundary and effect | +| --- | --- | +| Approval create, cancel, revision, resubmit, resolve | `writeApprovalChange` notifies retention and queue seeding before writes in the owning transaction. | +| Shared approval records, budget approvals, formal tool approvals | The same approval notification covers their existing transactions. | +| Issue approval link, unlink, link-many, tool-created links | Existing approval transaction helper. Link removal can expose a different first issue. | +| Issue visibility changes and issue deletion | Existing approval notification before the update or cascading delete. Routine failure cleanup retains this signal. | +| Company activation, pause, archive, deletion and budget pause/resume | Existing company notification wakes both approval consumers. | +| Work-product create, type/issue update, import batch, delete | The owning work-product transaction notifies seeding when a pull-request row is involved. | +| PR queue creation, rule update and manual approval membership removal | The queue mutation transaction notifies seeding. An enabled rule can re-add a still-matching manually removed item, as the old scan did. | + +Other direct work-product writes only create artifacts or update title, metadata, +or trust. They cannot create a pull-request match. Supported writes use these +in-process service boundaries. Independent SQL writers and multiple server +replicas are not covered by local transaction signals. + +Retention-only keep/archive/revive changes do not create queue projection debt. +The old seed writer ran before retention filtering, so archived or kept approvals +still follow the same seeding rule. User-specific dismissals do not filter the +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 +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 +and waits for an admitted transaction; unprocessed rows remain recoverable. +Registration follows the existing heartbeat-maintenance enablement switch. + +This slice does not extend Cloud alarm coverage or loosen host sleep admission. +It proves a worker's quiet/reactivation behavior, not deployed host or database +suspension. The remaining attention maintenance interval is still present for +other source owners. + +## Verification + +- Before the fix, the PostgreSQL regression found no PR queue after the normal + maintenance path excluded approvals. After the fix, the queue API finds it + without a full attention read. +- Twenty real-database worker tests cover empty-to-active-to-quiet behavior, + startup and completed history, actual approval/link/work-product producers, + batch import, first-link selection, company isolation, hidden targets, queue + rule enablement, company activation, resubmission, outer commit and rollback, + first write during drain, database failure through drain, multiple pages, + failed-page progress and retry, concurrent resolution, archived/kept history, shutdown/restart, and an admitted transaction at shutdown. +- 115 existing tests pass across queue routes, attention, approval retention, + approval service, work products, and startup wiring. The shared writer still + passes existing plan/question/PR seed tests and authorization tests. +- Workspace typechecks and the production build pass. Fresh PR checks are + tracked in the PR and must pass before readiness is claimed. +- The actual pending lookup on 40,000 completed approvals plus one pending PR + approval uses the existing partial pending-retention index. The probe runs in + a rolled-back transaction; execution took 0.094 ms on this local fixture. No new index or migration is needed for this case. + +The first expanded test run caught two fixture errors: a hand-inserted queue +omitted required creator attribution, and a failure test expected no idle hold +while committed projection work was still unresolved. The corrected test waits +for the active attempt to finish and checks passive SQL directly while preserving +that required hold. Neither required weakening the worker's recovery contract. 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 10b50209fb6..968a2d6fe72 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -888,3 +888,29 @@ claimed ready based only on local tests, and no merge is authorized. After join retention, migrate the remaining attention source owners separately: decisions, interactions, recovery actions, reviews, and other source kinds. Keep the full original server inventory and later aggregate-tick retirement in scope. + +## Slice 16: approval PR-queue membership + +The source audit found automatic seeded-queue materialization hidden inside the +attention retention scan. A PostgreSQL regression failed after approval exclusion: +queue API readers saw no PR queue despite a pending linked approval and PR work +product. This is a separate obligation from retention, not grounds to leave a +new permanent scan or to put queue policy into retention. + +The new approval queue worker uses existing durable records and the queue service's +normalized seed writer. It gets committed signals from all approval/link producers, +issue/company visibility changes, work-product writes, and queue-rule changes. +Missing membership is recovery debt; existing membership and disabled rules are +quiet. Twenty real-database lifecycle cases and 115 existing consumer/startup +tests pass, as do workspace typechecks and the production build. See +[approval queue seeding](../idle-approval-queue-seeding.md). Fresh CI/review +remain required before readiness. This slice is a correctness dependency of +approval retention #17497; do not accept that migration as functionally complete +without preserving queue population. Other source seeding remains to be migrated. + +#17505 now has all latest-head checks accepted at `28a197cdfdab7977eecf591df39612009d725e17` +(Paperclip Review skipped), with Greptile 5/5. #17508 at +`9c5c4fbf65628589b466ce5650e8969388213716` has Greptile 5/5 and no unresolved +threads. Its unchanged login-cleanup consume test failed in CI; all 16 local +login-cleanup tests pass. One justified retry of failed run 40139410144 was +requested. Do not mark that PR green until the retry completes successfully. diff --git a/server/src/__tests__/approval-queue-seeding-work.test.ts b/server/src/__tests__/approval-queue-seeding-work.test.ts new file mode 100644 index 00000000000..93a688b0afb --- /dev/null +++ b/server/src/__tests__/approval-queue-seeding-work.test.ts @@ -0,0 +1,293 @@ +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 { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; +import { approvalRecords } from "../services/approvals.js"; +import { approvalRetentionService } from "../services/approval-retention.js"; +import { approvalQueueSeedingService, registerApprovalQueueSeedingWork } from "../services/approval-queue-seeding-work.js"; +import { attentionService } from "../services/attention.js"; +import { companyService } from "../services/companies.js"; +import { decisionQueueService } from "../services/decision-queues.js"; +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 { idleWorkSnapshot } from "../services/task-admission.js"; + +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_APPROVAL_QUEUE_TEST_DATABASE_URL"); +const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport(); +(support.supported ? describe : describe.skip)("approval queue seeding", () => { + let temporary: Awaited> | undefined; + let db: ReturnType, companyId: string; + let coordinator: ReturnType | undefined; + let draining = false; + const errors: unknown[] = []; + beforeAll(async () => { + if (!external) temporary = await startEmbeddedPostgresTestDatabase("approval-queue-"); + db = createDb(external ?? temporary!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); + }, 30_000); + beforeEach(async () => { + companyId = randomUUID(); draining = false; errors.length = 0; + await db.insert(companies).values({ id: companyId, name: "Approval queue", issuePrefix: `Q${companyId.slice(0, 6)}` }); + await Promise.all(Array.from({ length: 6 }, () => db.execute(sql`select pg_sleep(0.01)`))); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: true }); + }); + afterEach(async () => { + await coordinator?.stop(); coordinator = undefined; + vi.useRealTimers(); vi.restoreAllMocks(); + await db.execute(sql`drop trigger if exists approval_queue_fail on decision_queue_items`); + await db.execute(sql`drop function if exists approval_queue_fail()`); + await db.delete(decisionTriageEvents); await db.delete(decisionQueueItems); await db.delete(decisionQueues); + await db.delete(decisionRetention); await db.delete(activityLog); await db.delete(issueApprovals); + await db.delete(issueWorkProducts); await db.delete(approvals); await db.delete(issues); await db.delete(companies); + expect(errors).toEqual([]); + }); + afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); }); + const authActor = () => ({ type: "board" as const, source: "local_implicit" as const, userId: "board-user", companyIds: [companyId], isInstanceAdmin: true }); + const actor = { actorType: "user" as const, actorId: "board-user", agentId: null, userId: "board-user", runId: null, agentApiKeyId: null, responsibleUserId: "board-user" }; + const mutation = () => ({ companyId, key: "prs", authActor: authActor(), actor }); + const memberships = () => db.select().from(decisionQueueItems).where(eq(decisionQueueItems.companyId, companyId)); + const createApproval = (database: Db = db) => approvalRecords(database).create(companyId, { type: "request_board_approval", payload: {} }); + async function issue(hiddenAt: Date | null = null) { + const [row] = await db.insert(issues).values({ companyId, title: "Review PR", status: "todo", hiddenAt }).returning(); + return row; + } + const product = (issueId: string, type = "pull_request") => workProductService(db).createForIssue(issueId, companyId, { type, provider: "github", title: "PR", status: "open" }); + async function pending(target?: string) { + const targetId = target ?? (await issue()).id; + if (!target) await product(targetId); + const approval = await createApproval(); + await issueApprovalService(db).link(targetId, approval.id); + return approval; + } + async function start(service = approvalQueueSeedingService(db)) { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining, canReconcile: () => true, onError: error => errors.push(error) }); + await registerApprovalQueueSeedingWork(coordinator, service, () => !draining).ready; + return service; + } + async function settle() { + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1_000); + expect(idleWorkSnapshot().active).toBe(0); + }, { timeout: 5000 }); + } + async function seeded(id: string) { + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1_000); + expect(await memberships()).toEqual(expect.arrayContaining([expect.objectContaining({ sourceKind: "approval", sourceId: id })])); + }, { timeout: 5000 }); + } + async function quiet() { + await settle(); expect(coordinator!.nextWakeAt()).toBeNull(); + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction"), executes = vi.spyOn(db, "execute"); + await vi.advanceTimersByTimeAsync(300_000); + expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); expect(executes).not.toHaveBeenCalled(); + reads.mockRestore(); writes.mockRestore(); executes.mockRestore(); + } + + it("populates the PR queue without an attention reader after approvals leave retention maintenance", async () => { + await start(); await quiet(); + const approval = await pending(); + await approvalRetentionService(db).sweep(); + await attentionService(db).list(companyId, { all: true, allowUnscopedAll: true, excludeSources: ["approval", "join_request"] }); + await seeded(approval.id); + expect((await decisionQueueService(db).list(companyId, authActor())).map(queue => queue.key)).toContain("prs"); + expect(await decisionQueueService(db).listItems(companyId, "prs", authActor())).toEqual(expect.arrayContaining([ + expect.objectContaining({ sourceKind: "approval", sourceId: approval.id }), + ])); + await quiet(); + }); + + it("recovers startup projection debt and stays quiet with already seeded and completed history", async () => { + const first = await pending(), completed = await pending(); + await approvalRecords(db).resolveApproval(completed.id, "approved", "board-user", null); + await start(); await seeded(first.id); await quiet(); + expect((await memberships()).map(row => row.sourceId)).toEqual([first.id]); + await coordinator!.stop(); coordinator = undefined; + await start(); await quiet(); + expect(await db.select().from(decisionTriageEvents).where(eq(decisionTriageEvents.action, "queue_item.seeded"))).toHaveLength(1); + }); + + it("wakes when a work product creates the first PR match", async () => { + const target = await issue(), approval = await pending(target.id); + await start(); await quiet(); expect(await memberships()).toEqual([]); + const row = await product(target.id, "artifact"); await quiet(); + await workProductService(db).update(row!.id, { type: "pull_request" }); + await seeded(approval.id); await quiet(); + }); + + it("wakes from batch-imported PR work products", async () => { + const target = await issue(), approval = await pending(target.id); + await start(); await quiet(); + await workProductService(db).createManyForImport([{ companyId, issueId: target.id, type: "pull_request", provider: "github", title: "Imported PR", status: "open", reviewState: "none", isPrimary: false, healthStatus: "unknown" }]); + await seeded(approval.id); await quiet(); + }); + + it("honors hidden targets and wakes when visibility changes", async () => { + const target = await issue(new Date()); await product(target.id); const approval = await pending(target.id); + await start(); await quiet(); expect(await memberships()).toEqual([]); + await issueService(db).update(target.id, { hiddenAt: null }); await seeded(approval.id); await quiet(); + }); + + it("uses the first linked issue and reevaluates after unlink", async () => { + const first = await issue(), second = await issue(); + const [withoutPr, withPr] = [first, second].sort((a, b) => a.id.localeCompare(b.id)); + await product(withPr.id); + const approval = await pending(withoutPr.id); await issueApprovalService(db).link(withPr.id, approval.id); + await start(); await quiet(); expect(await memberships()).toEqual([]); + await issueApprovalService(db).unlink(withoutPr.id, approval.id); await seeded(approval.id); await quiet(); + }); + + it("does not infer PRs from a later link or another company's work product", async () => { + const target = await issue(), approval = await pending(target.id); + const foreignId = randomUUID(); await db.insert(companies).values({ id: foreignId, name: "Other", issuePrefix: "OTHER" }); + await db.insert(issueWorkProducts).values({ companyId: foreignId, issueId: target.id, type: "pull_request", provider: "github", title: "Invalid foreign product", status: "open" }); + await start(); await quiet(); expect(await memberships()).toEqual([]); + await product(target.id); await seeded(approval.id); await quiet(); + }); + + it("keeps disabled seeds quiet and wakes when the queue rule is enabled", async () => { + await db.insert(decisionQueues).values({ companyId, key: "prs", title: "PRs", createdByType: "system", seedRulesEnabled: false }); + const approval = await pending(); await start(); await quiet(); expect(await memberships()).toEqual([]); + await decisionQueueService(db).update({ ...mutation(), patch: { seedRulesEnabled: true } }); + await seeded(approval.id); await quiet(); + }); + + it("waits for company activation", async () => { + const approval = await pending(); await companyService(db).update(companyId, { status: "paused" }); + await start(); await quiet(); expect(await memberships()).toEqual([]); + await companyService(db).update(companyId, { status: "active" }); await seeded(approval.id); await quiet(); + }); + + it("waits for resubmission and never removes retained membership after approval", async () => { + const approval = await pending(); await approvalRecords(db).requestRevision(approval.id, "board-user"); + await start(); await quiet(); expect(await memberships()).toEqual([]); + await approvalRecords(db).resubmit(approval.id); await seeded(approval.id); + await approvalRecords(db).resolveApproval(approval.id, "approved", "board-user", null); await quiet(); + expect(await memberships()).toHaveLength(1); + }); + + it("waits for outer commit and discards rolled-back producers", async () => { + const target = await issue(); await product(target.id); await start(); await quiet(); + await expect(db.transaction(async tx => { + const approval = await createApproval(tx as unknown as Db); + await issueApprovalService(tx as unknown as Db).link(target.id, approval.id); + await vi.advanceTimersByTimeAsync(5_000); expect(await memberships()).toEqual([]); + throw new Error("rollback"); + })).rejects.toThrow("rollback"); + await quiet(); + let id = ""; + await db.transaction(async tx => { + id = (await createApproval(tx as unknown as Db)).id; + await issueApprovalService(tx as unknown as Db).link(target.id, id); + await vi.advanceTimersByTimeAsync(5_000); expect(await memberships()).toEqual([]); + }); + await seeded(id); await quiet(); + }); + + it("retains the first write during drain without passive SQL", async () => { + await start(); await quiet(); draining = true; + const approval = await pending(); + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction"); + await vi.advanceTimersByTimeAsync(60_000); + expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); + reads.mockRestore(); writes.mockRestore(); + expect(await memberships()).toEqual([]); draining = false; + await seeded(approval.id); await quiet(); + }); + + it("recovers a failed projection after drain and stops retrying once it commits", async () => { + const service = approvalQueueSeedingService(db), sweep = service.sweep; + let running = false; + service.sweep = async signal => { running = true; try { await sweep(signal); } finally { running = false; } }; + await start(service); await quiet(); + await db.execute(sql`create function approval_queue_fail() returns trigger language plpgsql as $$ begin raise exception 'projection unavailable'; end $$`); + await db.execute(sql`create trigger approval_queue_fail before insert on decision_queue_items for each row execute function approval_queue_fail()`); + const approval = await pending(); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1_000); expect(errors.length).toBeGreaterThan(0); }, { timeout: 5000 }); + draining = true; + await vi.waitFor(() => expect(running).toBe(false), { timeout: 5000 }); + const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(60_000); expect(reads).not.toHaveBeenCalled(); reads.mockRestore(); + await db.execute(sql`drop trigger approval_queue_fail on decision_queue_items`); errors.length = 0; + draining = false; await seeded(approval.id); await quiet(); + }); + + it("drains multiple pages without duplicating memberships or audit events", async () => { + const target = await issue(); await product(target.id); + const rows = await db.insert(approvals).values(Array.from({ length: 101 }, () => ({ companyId, type: "request_board_approval", payload: {} }))).returning(); + await db.insert(issueApprovals).values(rows.map(row => ({ companyId, issueId: target.id, approvalId: row.id }))); + await start(); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1_000); expect(await memberships()).toHaveLength(101); }, { timeout: 10_000 }); + await quiet(); + await attentionService(db).list(companyId, { all: true, allowUnscopedAll: true }); await quiet(); + expect(await memberships()).toHaveLength(101); + expect(await db.select().from(decisionTriageEvents).where(eq(decisionTriageEvents.action, "queue_item.seeded"))).toHaveLength(101); + }); + + it("preserves the seed rule after manual removal of a still-matching item", async () => { + const approval = await pending(); await start(); await seeded(approval.id); await quiet(); + await decisionQueueService(db).removeItem({ ...mutation(), sourceKind: "approval", sourceId: approval.id }); + await seeded(approval.id); await quiet(); expect(await memberships()).toHaveLength(1); + }); + it("pages past failed projections and revisits them after recovery", async () => { + const target = await issue(); await product(target.id); + const rows = await db.insert(approvals).values(Array.from({ length: 101 }, () => ({ companyId, type: "request_board_approval", payload: {} }))).returning(); + await db.insert(issueApprovals).values(rows.map(row => ({ companyId, issueId: target.id, approvalId: row.id }))); + const last = rows.map(row => row.id).sort().at(-1)!; + await db.execute(sql`create function approval_queue_fail() returns trigger language plpgsql as $$ begin + if new.source_id <> TG_ARGV[0] then raise exception 'projection unavailable'; end if; return new; end $$`); + await db.execute(sql`create trigger approval_queue_fail before insert on decision_queue_items for each row execute function approval_queue_fail(${sql.raw("'" + last + "'")})`); + const service = approvalQueueSeedingService(db); + await expect(service.sweep()).rejects.toThrow("Approval queue seeding failed"); + await service.sweep(); expect((await memberships()).map(row => row.sourceId)).toEqual([last]); + await db.execute(sql`drop trigger approval_queue_fail on decision_queue_items`); + await service.sweep(); expect(await memberships()).toHaveLength(101); expect(await service.next()).toBeNull(); + }); + + it("rechecks the approval after a concurrent resolution commits", async () => { + const approval = await pending(); let release!: () => void, locked!: () => void; + const gate = new Promise(resolve => { release = resolve; }), admitted = new Promise(resolve => { locked = resolve; }); + const write = db.transaction(async tx => { + await tx.execute(sql`select id from approvals where id = ${approval.id} for update`); locked(); await gate; + await approvalRecords(tx as unknown as Db).resolveApproval(approval.id, "approved", "board-user", null); + }); + await admitted; const sweep = approvalQueueSeedingService(db).sweep(); + await vi.advanceTimersByTimeAsync(100); release(); await write; await sweep; + expect(await memberships()).toEqual([]); + }); + + it("preserves startup debt during drain and reconstructs unprocessed work after shutdown", async () => { + const approval = await pending(); draining = true; await start(); + const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(60_000); expect(reads).not.toHaveBeenCalled(); reads.mockRestore(); + await coordinator!.stop(); coordinator = undefined; + expect(await memberships()).toEqual([]); + draining = false; await start(); await seeded(approval.id); await quiet(); + }); + + it("preserves seeding for kept and archived approvals", async () => { + const approval = await pending(); + await db.insert(decisionRetention).values({ companyId, sourceKind: "approval", sourceId: approval.id, + sourceActivityAt: approval.updatedAt, keep: true, archivedAt: new Date() }); + await start(); await seeded(approval.id); await quiet(); + }); + + it("waits for an admitted transaction on shutdown and leaves the next row recoverable", async () => { + await pending(); await pending(); + let release!: () => void, admitted!: () => void; + const gate = new Promise(resolve => { release = resolve; }), entered = new Promise(resolve => { admitted = resolve; }); + const transaction = db.transaction.bind(db); + const intercepted = vi.spyOn(db, "transaction").mockImplementationOnce(async callback => { + admitted(); await gate; return transaction(callback); + }); + const started = start(); await entered; + let stopped = false; + const stopping = coordinator!.stop().then(() => { stopped = true; }); + await vi.advanceTimersByTimeAsync(100); expect(stopped).toBe(false); + release(); await started; await stopping; coordinator = undefined; intercepted.mockRestore(); + expect(await memberships()).toHaveLength(1); + await start(); await quiet(); expect(await memberships()).toHaveLength(2); + }); + +}); diff --git a/server/src/__tests__/decision-queues-routes.test.ts b/server/src/__tests__/decision-queues-routes.test.ts index b1897d06469..6bfb2e50a9f 100644 --- a/server/src/__tests__/decision-queues-routes.test.ts +++ b/server/src/__tests__/decision-queues-routes.test.ts @@ -1,3 +1,4 @@ +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { randomUUID } from "node:crypto"; import express from "express"; import request from "supertest"; @@ -26,7 +27,8 @@ import { errorHandler } from "../middleware/index.js"; import { decisionQueueRoutes } from "../routes/decision-queues.js"; import { decisionQueueService } from "../services/decision-queues.js"; -const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_DECISION_QUEUES_TEST_DATABASE_URL"); +const embeddedPostgresSupport = external ? { supported: true, reason: undefined } : await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; if (!embeddedPostgresSupport.supported) { @@ -40,8 +42,8 @@ describeEmbeddedPostgres("decision queue routes", () => { let tempDb: Awaited> | null = null; beforeAll(async () => { - tempDb = await startEmbeddedPostgresTestDatabase("paperclip-decision-queues-"); - db = createDb(tempDb.connectionString); + if (!external) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-decision-queues-"); + db = createDb(external ?? tempDb!.connectionString); }, 30_000); afterEach(async () => { @@ -59,6 +61,7 @@ describeEmbeddedPostgres("decision queue routes", () => { }); afterAll(async () => { + await db?.$client.end({ timeout: 1 }); await tempDb?.cleanup(); }); diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index c3fc88e4d12..e95f4ca9289 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 { approvalQueueSeedingService } from "../services/approval-queue-seeding-work.js"; import { joinRequestRetentionService } from "../services/join-request-retention.js"; import { approvalRetentionService } from "../services/approval-retention.js"; import { logger } from "../middleware/logger.js"; @@ -328,6 +329,11 @@ vi.mock("../services/join-request-retention.js", async importOriginal => ({ joinRequestRetentionService: vi.fn(() => ({ sweep: vi.fn(async () => ({ archived: 0 })), next: vi.fn(async () => null) })), })); +vi.mock("../services/approval-queue-seeding-work.js", async importOriginal => ({ + ...await importOriginal(), + approvalQueueSeedingService: vi.fn(() => ({ sweep: vi.fn(async () => undefined), next: vi.fn(async () => null) })), +})); + vi.mock("../services/approval-retention.js", async importOriginal => ({ ...await importOriginal(), approvalRetentionService: vi.fn(() => ({ sweep: vi.fn(async () => ({ archived: 0 })), next: vi.fn(async () => null) })), @@ -531,6 +537,18 @@ describe("startServer feedback export wiring", () => { process.env.BETTER_AUTH_SECRET = "test-secret"; }); + it.each([false, true])("registers approval queue seeding only when maintenance is enabled (%s)", async enabled => { + loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled })); + await startServer(); await settleStartupHousekeeping(); + const registration = deliveryWorkRegisterMock.mock.calls.find(([queue]) => queue === DELIVERY_QUEUES.approvalQueueSeeding); + expect(Boolean(registration)).toBe(enabled); + if (enabled) { + const service = vi.mocked(approvalQueueSeedingService).mock.results.at(-1)!.value; + await registration![1].run(new AbortController().signal); + expect(service.sweep).toHaveBeenCalled(); expect(service.next).toHaveBeenCalled(); + } + }); + it.each([false, true])("registers approval retention only when maintenance is enabled (%s)", async enabled => { loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled })); await startServer(); await settleStartupHousekeeping(); diff --git a/server/src/index.ts b/server/src/index.ts index 323bf6f84b3..06ed12a86ef 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1,3 +1,4 @@ +import { approvalQueueSeedingService, registerApprovalQueueSeedingWork } from "./services/approval-queue-seeding-work.js"; import { joinRequestRetentionService, registerJoinRequestRetentionWork } from "./services/join-request-retention.js"; import { approvalRetentionService, registerApprovalRetentionWork } from "./services/approval-retention.js"; import { registerDecisionTargetWork } from "./services/decision-target-work.js"; @@ -1825,6 +1826,8 @@ async function startServerWithDatabaseTeardown( await step("environment_lease_cleanup", startEnvironmentCleanup); await step("archive_notification_worker", () => registerDecisionArchiveDeliveryWork(deliveryWork, retentionExecutor, () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready); + await step("approval_queue_seeding", () => registerApprovalQueueSeedingWork(deliveryWork, approvalQueueSeedingService(db as any), + () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready); await step("approval_retention", () => registerApprovalRetentionWork(deliveryWork, approvalRetentionService(db as any), () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready); await step("join_request_retention", () => registerJoinRequestRetentionWork(deliveryWork, joinRequestRetentionService(db as any), diff --git a/server/src/lib/approval-records.ts b/server/src/lib/approval-records.ts index cbc70c39bac..aaecd65bfda 100644 --- a/server/src/lib/approval-records.ts +++ b/server/src/lib/approval-records.ts @@ -1,4 +1,4 @@ -import { writeApprovalChange } from "../services/retention-work-notifications.js"; +import { writeApprovalChange } from "../services/approval-work-notifications.js"; import { and, eq, inArray } from "drizzle-orm"; import { approvals, type Db } from "@paperclipai/db"; import { notFound, unprocessable } from "../errors.js"; diff --git a/server/src/services/approval-queue-seeding-work.ts b/server/src/services/approval-queue-seeding-work.ts new file mode 100644 index 00000000000..bf0c0358e0a --- /dev/null +++ b/server/src/services/approval-queue-seeding-work.ts @@ -0,0 +1,63 @@ +import { and, asc, eq, gt, sql, type SQL } from "drizzle-orm"; +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 + where linked.company_id = ${approvals.companyId} and linked.approval_id = ${approvals.id} + order by linked.issue_id limit 1)`; + +export function approvalQueueSeedingService(db: Db) { + let cursor: string | undefined; + function candidates(database: Db, 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, + 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} + and product.issue_id = ${issues.id} and product.type = 'pull_request'))`, + sql`not exists (select 1 from decision_queues queue where queue.company_id = ${approvals.companyId} and queue.key = 'prs' + and (not queue.seed_rules_enabled or exists (select 1 from decision_queue_items item where item.queue_id = queue.id + and item.company_id = ${approvals.companyId} and item.source_kind = 'approval' and item.source_id = ${approvals.id}::text)))`, + )).orderBy(asc(approvals.id)); + } + async function sweep(signal?: AbortSignal) { + const rows = await candidates(db, cursor ? gt(approvals.id, 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 }); + 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; + } + 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 { sweep, next }; +} + +export function registerApprovalQueueSeedingWork( + coordinator: ReturnType, + service: ReturnType, + canRun: () => boolean, +) { + return registerDeadlineWork(coordinator, DELIVERY_QUEUES.approvalQueueSeeding, { + run: signal => service.sweep(signal), next: service.next, canRun, retryMs: 1_000, + }); +} diff --git a/server/src/services/approval-work-notifications.ts b/server/src/services/approval-work-notifications.ts new file mode 100644 index 00000000000..39d8e9fd03e --- /dev/null +++ b/server/src/services/approval-work-notifications.ts @@ -0,0 +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 async function notifyApprovalWork(transaction: object) { + await notifyDeliveryWork(transaction, DELIVERY_QUEUES.approvalRetention); + await notifyApprovalQueueWork(transaction); +} + +export function writeApprovalChange(db: Db, write: (transaction: Db) => Promise): Promise { + return db.transaction(async tx => { + await notifyApprovalWork(tx); + return write(tx as unknown as Db); + }); +} diff --git a/server/src/services/approvals.ts b/server/src/services/approvals.ts index 0fbec1a44e2..b806e112368 100644 --- a/server/src/services/approvals.ts +++ b/server/src/services/approvals.ts @@ -1,4 +1,4 @@ -import { writeApprovalChange } from "./retention-work-notifications.js"; +import { writeApprovalChange } from "./approval-work-notifications.js"; import { approvalResolutionRecords, resolvableApprovalStatuses as resolvableStatuses } from "../lib/approval-records.js"; import { notifyHireApproved } from "./hire-hook.js"; diff --git a/server/src/services/budgets.ts b/server/src/services/budgets.ts index aba74d6e0c6..0ce6009fce2 100644 --- a/server/src/services/budgets.ts +++ b/server/src/services/budgets.ts @@ -1,4 +1,5 @@ -import { notifyCompanyRetentionWork, notifyRetentionWork } from "./retention-work-notifications.js"; +import { notifyApprovalWork } from "./approval-work-notifications.js"; +import { notifyCompanyRetentionWork } from "./retention-work-notifications.js"; import { addCents, compareCents, normalizeCents } from "@paperclipai/shared"; import { and, desc, eq, inArray, sql, or, isNull } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; @@ -259,7 +260,7 @@ async function markApprovalStatus( decidedByUserId: string, ) { if (!approvalId) return; - await notifyRetentionWork(db, "approval"); + await notifyApprovalWork(db); await db .update(approvals) .set({ @@ -457,7 +458,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica windowEnd: end, }); - if (thresholdType === "hard") await notifyRetentionWork(db, "approval"); + if (thresholdType === "hard") await notifyApprovalWork(db); const approval = thresholdType === "hard" ? await db .insert(approvals) diff --git a/server/src/services/decision-queues.ts b/server/src/services/decision-queues.ts index d529e1d0a8d..db39daac1a7 100644 --- a/server/src/services/decision-queues.ts +++ b/server/src/services/decision-queues.ts @@ -1,3 +1,4 @@ +import { notifyApprovalQueueWork } from "./approval-work-notifications.js"; import { and, desc, eq, inArray, isNull, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { @@ -43,6 +44,14 @@ export type DecisionMutationActor = { responsibleUserId: string | null; }; +export type DecisionQueueSeedSource = { + sourceKind: AttentionSourceKind; + sourceId: string; + issueId: string | null; + isPlanTarget?: boolean; + interactionKind?: string; +}; + type SeedDefinition = { key: string; title: string; @@ -380,6 +389,104 @@ export function decisionQueueService(db: Db) { return visible; } + async function materializeSeedSources(companyId: string, items: DecisionQueueSeedSource[]) { + if (items.length === 0) return; + const issueIds = [...new Set(items.map(item => item.issueId).filter((id): id is string => Boolean(id)))]; + const prIssueIds = new Set(issueIds.length === 0 ? [] : await db + .select({ issueId: issueWorkProducts.issueId }) + .from(issueWorkProducts) + .where(and( + eq(issueWorkProducts.companyId, companyId), + eq(issueWorkProducts.type, "pull_request"), + inArray(issueWorkProducts.issueId, issueIds), + )).then((rows) => rows.map((row) => row.issueId))); + + const matches = new Map(); + for (const item of items) { + if (prIssueIds.has(item.issueId ?? "")) { + matches.set("prs", [...(matches.get("prs") ?? []), item]); + } + if (item.sourceKind === "issue_thread_interaction" && item.isPlanTarget === true) { + matches.set("plans", [...(matches.get("plans") ?? []), item]); + } + if (item.sourceKind === "issue_thread_interaction" && item.interactionKind === "ask_user_questions") { + matches.set("questions", [...(matches.get("questions") ?? []), item]); + } + } + + for (const seed of DECISION_QUEUE_SEEDS) { + const matchingItems = matches.get(seed.key) ?? []; + if (matchingItems.length === 0) continue; + await db.transaction(async (tx) => { + const txDb = tx as unknown as Db; + const insertedQueue = await txDb.insert(decisionQueues).values({ + companyId, + key: seed.key, + title: seed.title, + description: seed.description, + createdByType: "system", + seedRules: seed.rules, + seedRulesEnabled: true, + }).onConflictDoNothing({ target: [decisionQueues.companyId, decisionQueues.key] }).returning(); + const queue = insertedQueue[0] ?? await txDb.select().from(decisionQueues) + .where(and(eq(decisionQueues.companyId, companyId), eq(decisionQueues.key, seed.key))) + .for("update") + .then((rows) => rows[0] ?? null); + if (!queue || !queue.seedRulesEnabled) return; + if (insertedQueue[0]) { + await txDb.insert(decisionTriageEvents).values({ + companyId, + queueId: queue.id, + action: "queue.seeded", + ...eventActorColumns(SYSTEM_ACTOR), + details: { key: seed.key, rules: seed.rules.map((rule) => rule.key) }, + }); + await recordActivity(txDb, SYSTEM_ACTOR, { + companyId, + action: "decision_queue.seeded", + entityType: "decision_queue", + entityId: queue.id, + details: { key: seed.key }, + }); + } + let insertedAnyItem = false; + for (const item of matchingItems) { + const inserted = await txDb.insert(decisionQueueItems).values({ + companyId, + queueId: queue.id, + sourceKind: item.sourceKind, + sourceId: item.sourceId, + addedByType: "system", + }).onConflictDoNothing({ + target: [decisionQueueItems.queueId, decisionQueueItems.sourceKind, decisionQueueItems.sourceId], + }).returning(); + if (!inserted[0]) continue; + insertedAnyItem = true; + await txDb.insert(decisionTriageEvents).values({ + companyId, + queueId: queue.id, + sourceKind: item.sourceKind, + sourceId: item.sourceId, + action: "queue_item.seeded", + ...eventActorColumns(SYSTEM_ACTOR), + details: { seedKey: seed.key }, + }); + await recordActivity(txDb, SYSTEM_ACTOR, { + companyId, + action: "decision_queue_item.seeded", + entityType: "decision_queue", + entityId: queue.id, + details: { sourceKind: item.sourceKind, sourceId: item.sourceId, seedKey: seed.key }, + }); + } + if (insertedAnyItem) { + await txDb.update(decisionQueues).set({ updatedAt: new Date() }) + .where(and(eq(decisionQueues.companyId, companyId), eq(decisionQueues.id, queue.id))); + } + }); + } + } + return { create: async (input: { companyId: string; @@ -392,6 +499,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); const inserted = await txDb.insert(decisionQueues).values({ companyId: input.companyId, key: input.key, @@ -449,6 +557,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); 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); @@ -556,6 +665,7 @@ export function decisionQueueService(db: Db) { } return db.transaction(async (tx) => { const txDb = tx as unknown as Db; + if (input.key === "prs" && input.sourceKind === "approval") await notifyApprovalQueueWork(txDb); const removed = await txDb.delete(decisionQueueItems).where(and( eq(decisionQueueItems.companyId, input.companyId), eq(decisionQueueItems.queueId, queue.id), @@ -682,101 +792,13 @@ export function decisionQueueService(db: Db) { }); }, - materializeSeededQueues: async (companyId: string, items: AttentionItem[]) => { - if (items.length === 0) return; - const issueIds = [...new Set(items.map(itemIssueId).filter((id): id is string => Boolean(id)))]; - const prIssueIds = new Set(issueIds.length === 0 ? [] : await db - .select({ issueId: issueWorkProducts.issueId }) - .from(issueWorkProducts) - .where(and( - eq(issueWorkProducts.companyId, companyId), - eq(issueWorkProducts.type, "pull_request"), - inArray(issueWorkProducts.issueId, issueIds), - )).then((rows) => rows.map((row) => row.issueId))); - - const matches = new Map(); - for (const item of items) { - if (prIssueIds.has(itemIssueId(item) ?? "")) { - matches.set("prs", [...(matches.get("prs") ?? []), item]); - } - if (item.sourceKind === "issue_thread_interaction" && item.subject.metadata?.isPlanTarget === true) { - matches.set("plans", [...(matches.get("plans") ?? []), item]); - } - if (item.sourceKind === "issue_thread_interaction" && item.subject.metadata?.kind === "ask_user_questions") { - matches.set("questions", [...(matches.get("questions") ?? []), item]); - } - } - - for (const seed of DECISION_QUEUE_SEEDS) { - const matchingItems = matches.get(seed.key) ?? []; - if (matchingItems.length === 0) continue; - await db.transaction(async (tx) => { - const txDb = tx as unknown as Db; - const insertedQueue = await txDb.insert(decisionQueues).values({ - companyId, - key: seed.key, - title: seed.title, - description: seed.description, - createdByType: "system", - seedRules: seed.rules, - seedRulesEnabled: true, - }).onConflictDoNothing({ target: [decisionQueues.companyId, decisionQueues.key] }).returning(); - const queue = insertedQueue[0] ?? await txDb.select().from(decisionQueues) - .where(and(eq(decisionQueues.companyId, companyId), eq(decisionQueues.key, seed.key))) - .then((rows) => rows[0] ?? null); - if (!queue || !queue.seedRulesEnabled) return; - if (insertedQueue[0]) { - await txDb.insert(decisionTriageEvents).values({ - companyId, - queueId: queue.id, - action: "queue.seeded", - ...eventActorColumns(SYSTEM_ACTOR), - details: { key: seed.key, rules: seed.rules.map((rule) => rule.key) }, - }); - await recordActivity(txDb, SYSTEM_ACTOR, { - companyId, - action: "decision_queue.seeded", - entityType: "decision_queue", - entityId: queue.id, - details: { key: seed.key }, - }); - } - let insertedAnyItem = false; - for (const item of matchingItems) { - const inserted = await txDb.insert(decisionQueueItems).values({ - companyId, - queueId: queue.id, - sourceKind: item.sourceKind, - sourceId: item.subject.id, - addedByType: "system", - }).onConflictDoNothing({ - target: [decisionQueueItems.queueId, decisionQueueItems.sourceKind, decisionQueueItems.sourceId], - }).returning(); - if (!inserted[0]) continue; - insertedAnyItem = true; - await txDb.insert(decisionTriageEvents).values({ - companyId, - queueId: queue.id, - sourceKind: item.sourceKind, - sourceId: item.subject.id, - action: "queue_item.seeded", - ...eventActorColumns(SYSTEM_ACTOR), - details: { seedKey: seed.key }, - }); - await recordActivity(txDb, SYSTEM_ACTOR, { - companyId, - action: "decision_queue_item.seeded", - entityType: "decision_queue", - entityId: queue.id, - details: { sourceKind: item.sourceKind, sourceId: item.subject.id, seedKey: seed.key }, - }); - } - if (insertedAnyItem) { - await txDb.update(decisionQueues).set({ updatedAt: new Date() }) - .where(and(eq(decisionQueues.companyId, companyId), eq(decisionQueues.id, queue.id))); - } - }); - } - }, + materializeSeedSources, + materializeSeededQueues: (companyId: string, items: AttentionItem[]) => materializeSeedSources(companyId, items.map(item => ({ + sourceKind: item.sourceKind, + sourceId: item.subject.id, + issueId: itemIssueId(item), + isPlanTarget: item.subject.metadata?.isPlanTarget === true, + interactionKind: typeof item.subject.metadata?.kind === "string" ? item.subject.metadata.kind : undefined, + }))), }; } diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 7355c5d3a6c..641812c6213 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -8,6 +8,7 @@ export const DELIVERY_QUEUES = { executionStatus: "execution-status", executionDisposition: "execution-disposition", environmentCleanup: "environment-cleanup", + approvalQueueSeeding: "approval-queue-seeding", approvalRetention: "approval-retention", joinRequestRetention: "join-request-retention", decisionTargets: "decision-targets", diff --git a/server/src/services/issue-approvals.ts b/server/src/services/issue-approvals.ts index 9d706204522..993355ebaeb 100644 --- a/server/src/services/issue-approvals.ts +++ b/server/src/services/issue-approvals.ts @@ -1,4 +1,4 @@ -import { writeApprovalChange } from "./retention-work-notifications.js"; +import { writeApprovalChange } from "./approval-work-notifications.js"; import { and, desc, eq, inArray } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { approvals, issueApprovals, issues } from "@paperclipai/db"; diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index 50307afaf25..b6318522947 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -1,4 +1,4 @@ -import { notifyRetentionWork } from "./retention-work-notifications.js"; +import { notifyApprovalWork } from "./approval-work-notifications.js"; import { notifyDecisionTargetWork } from "./decision-target-work.js"; import { insertIssueComment } from "./issue-comment-writes.js"; import { measureRunTransaction } from "./execution-transaction-timing.js"; @@ -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 notifyRetentionWork(tx, "approval"); + await notifyApprovalWork(tx); } const updated = await tx .update(issues) @@ -11711,7 +11711,7 @@ export function issueService(db: Db) { try { await notifyTerminalWorkspaceWork(tx, `issue:${id}`); await requestRestorePointDeletion(tx, { sourceIssueId: id }); - await notifyRetentionWork(tx, "approval"); + await notifyApprovalWork(tx); 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 a52ce2de883..e42efd359bd 100644 --- a/server/src/services/retention-work-notifications.ts +++ b/server/src/services/retention-work-notifications.ts @@ -1,3 +1,4 @@ +import { notifyApprovalWork } from "./approval-work-notifications.js"; import type { Db } from "@paperclipai/db"; import type { AttentionSourceKind } from "@paperclipai/shared"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; @@ -7,15 +8,8 @@ export async function notifyRetentionWork(transaction: object, kind: AttentionSo if (kind === "join_request") await notifyDeliveryWork(transaction, DELIVERY_QUEUES.joinRequestRetention); } -export function writeApprovalChange(db: Db, write: (transaction: Db) => Promise): Promise { - return db.transaction(async tx => { - await notifyRetentionWork(tx, "approval"); - return write(tx as unknown as Db); - }); -} - export async function notifyCompanyRetentionWork(transaction: object) { - await notifyRetentionWork(transaction, "approval"); + await notifyApprovalWork(transaction); await notifyRetentionWork(transaction, "join_request"); } diff --git a/server/src/services/routines.ts b/server/src/services/routines.ts index 4608d3b7f19..78a71cbbdc3 100644 --- a/server/src/services/routines.ts +++ b/server/src/services/routines.ts @@ -1,4 +1,4 @@ -import { notifyRetentionWork } from "./retention-work-notifications.js"; +import { notifyApprovalWork } from "./approval-work-notifications.js"; import { notifyDecisionTargetWork } from "./decision-target-work.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { verifyAppWebhook } from "./app-webhook.js"; @@ -1966,7 +1966,7 @@ export function routineService( return updated ?? createdRun; } catch (error) { if (createdIssue) { - await notifyRetentionWork(txDb, "approval"); + await notifyApprovalWork(txDb); 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 fd029ad58c3..1a20945f14c 100644 --- a/server/src/services/tool-gateway.ts +++ b/server/src/services/tool-gateway.ts @@ -1,4 +1,4 @@ -import { writeApprovalChange } from "./retention-work-notifications.js"; +import { writeApprovalChange } from "./approval-work-notifications.js"; import { toolReviewProjectionService } from "./tool-review-projection.js"; import { toolReviewInterruptionService } from "./tool-review-interruption.js"; import { claimToolTestApproval, persistToolTestOutcome, startToolTestInvocation, type ToolTestApprovalClaim } from "./tool-test-approval.js"; diff --git a/server/src/services/work-products.ts b/server/src/services/work-products.ts index e4a8cbb5a10..a041b7a6d8f 100644 --- a/server/src/services/work-products.ts +++ b/server/src/services/work-products.ts @@ -1,3 +1,4 @@ +import { notifyApprovalQueueWork } from "./approval-work-notifications.js"; import { and, desc, eq, inArray } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { executionWorkspaces, heartbeatRunEvents, issueWorkProducts, workspaceRuntimeServices } from "@paperclipai/db"; @@ -273,7 +274,10 @@ export function workProductService( createForIssue: async (issueId: string, companyId: string, data: Omit) => { const row = await db.transaction(async (tx) => { - if (data.type === "pull_request") await notifyTerminalWorkspaceWork(tx, `issue:${issueId}`); + if (data.type === "pull_request") { + await notifyTerminalWorkspaceWork(tx, `issue:${issueId}`); + await notifyApprovalQueueWork(tx); + } await validateExecutionWorkspace(tx, companyId, data.executionWorkspaceId); if (data.isPrimary) { await tx @@ -309,7 +313,10 @@ export function workProductService( .then((rows) => rows[0] ?? null); if (!existing) return null; - if (existing.type === "pull_request" || patch.type === "pull_request") await notifyTerminalWorkspaceWork(tx, `issue:${existing.issueId}`); + if (existing.type === "pull_request" || patch.type === "pull_request") { + await notifyTerminalWorkspaceWork(tx, `issue:${existing.issueId}`); + await notifyApprovalQueueWork(tx); + } await validateExecutionWorkspace(tx, existing.companyId, patch.executionWorkspaceId); @@ -378,6 +385,7 @@ 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 value of values) { if (value.type === "pull_request") await notifyTerminalWorkspaceWork(tx, `issue:${value.issueId}`); } @@ -389,7 +397,10 @@ export function workProductService( const row = await db.transaction(async tx => { const [existing] = await tx.select({ type: issueWorkProducts.type, issueId: issueWorkProducts.issueId }) .from(issueWorkProducts).where(eq(issueWorkProducts.id, id)); - if (existing?.type === "pull_request") await notifyTerminalWorkspaceWork(tx, `issue:${existing.issueId}`); + if (existing?.type === "pull_request") { + await notifyTerminalWorkspaceWork(tx, `issue:${existing.issueId}`); + await notifyApprovalQueueWork(tx); + } return tx.delete(issueWorkProducts).where(eq(issueWorkProducts.id, id)) .returning().then(rows => rows[0] ?? null); });