diff --git a/doc/idle-approval-queue-seeding.md b/doc/idle-approval-queue-seeding.md new file mode 100644 index 00000000000..a741064ca00 --- /dev/null +++ b/doc/idle-approval-queue-seeding.md @@ -0,0 +1,231 @@ +# 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. The candidate lookup and writer share the pull-request product SQL +predicate, so the matching policy has one owner. 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 company-scoped queue seeding before writes in the owning transaction. Approve/reject/cancel retain their explicit tool-review notification; ordinary mutations do not wake those queues. | +| 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 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 and a fixed upper ID per scan cycle. The cursor advances past +failures and finishes that finite cycle before revisiting them; new arrivals cannot +extend it indefinitely. 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. +Unprocessed pages and newer committed signals retain a one-second continuation. +Failed scopes back off exponentially from one second to one minute. A successful +nonempty batch resets consecutive failures; an empty cursor rollover while debt +remains preserves outage backoff. New signals +do not reset a failing scope's backoff; other companies can proceed independently. +The next retry is registered with the shared scheduler, with no SQL during the +wait. A failed scope remains tracked until its owed memberships are reconciled. 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. + +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-one 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. + +Review found a queue/item lock inversion between seeded insertion and manual +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. + +The next CI run passed the security gate and the two originally failing shards. +Its remaining server shard caught a historical cleanup test checking 33 provider +inspections after only 32 had completed. The test advanced fake time before the +previous asynchronous page had necessarily published its next deadline. It now +waits for all 33 inspections, then retains the zero-further-work assertions. All +14 environment cleanup cases pass on real PostgreSQL; production code is unchanged. + + +Further review reproduced 120 failed transactions in two simulated minutes for a +persistently failing projection. Per-scope backoff now bounds that test to at most +eight attempts, even with new signals every second. Removing the fault recovers +the membership. A real coordinator/producer case proves thirty seconds without +SQL during backoff, another company's successful queue seeding during that wait, +and eventual recovery without losing the original work. All 25 worker cases pass. +The first consumer command omitted the external DB setting for queue routes and +skipped those six cases; the corrected command executes and passes all six. +The shared PR predicate is exercised by both those writer tests and worker tests. +Server typecheck and build pass; fresh remote review/checks remain required. + +The final startup, approval-retention, and attention consumer run passes all 93 +cases. With the 25 worker and six queue cases, 124 affected tests pass. + + +## Current retention-parent integration + +The seeding worker reuses `firstApprovalIssueId` from the approval-visibility +owner rather than retaining its own first-link SQL. Company IDs are part of the +approval notification contract because projection reconciliation is scoped. +Generic approval writes notify retention and seeding only; the existing terminal +approve/reject/cancel owners alone notify tool-review consumers. Current precise +issue-deletion target notifications, unseen retention controls, finite retention +cycles and the real source-row lock regressions are preserved. + +All 522 cases in fifteen complete suites pass at `4e725e2055`, with server +typecheck and production build. This includes seeding, queue routes, work products, +issue comments, both retention workers, approvals, attention, retention controls, +budgets, tool gateway, startup, scoped deadlines, environment cleanup and task +watchdogs. Four new disposable fixtures were cloned from the already migrated +parent fixture; this is not a fresh-empty-schema claim. The other suites used +separate existing disposable databases. No database was shared with another +concurrent local test run. The subsequent parent merge changes only two documents; +all tested runtime and acceptance sources remain byte-identical. + +The first integration typecheck caught two incorrect gateway company references. +They now use the existing session company, and typecheck plus the complete gateway +suite pass. Earlier deadlock, cross-company wake, backoff, skipped-fixture and +teardown failures remain in the historical sections above. The parent's attention +cleanup timeout and expiry-test CI assertion are recorded in its document; this +larger current run passes without changing their assertions or timeouts. Fresh +exact-head reviews and CI remain required before this dependent PR can merge. + + +## Final master and review repairs + +Join retention merged as `8406047b4e`. The normal master reconciliation first +proved that landed tree equals the existing parent. Conflict resolution preserved +the complete tested seeding tree; it also removed a duplicate helper introduced +by automatic merge. The PR again differs from master in only its original 25 +seeding paths. Database schema and migration metadata are identical to master. + +Three retry regressions failed before repair: ongoing arrivals skipped a recovered +membership, a partial successful batch retained the old sixty-second failure +penalty, and resetting that counter on an empty cursor rollover weakened outage +backoff. The worker now fixes the upper ID per cycle and resets consecutive +failures only on real successful progress or complete reconciliation. Exact +one-second recovery and two-second consecutive-failure assertions pass; repeated +signals still cannot reset a failing scope's delay. No new table or retry wrapper +is introduced. + +The external seeding, queue-route and watchdog fixtures now apply pending +migrations before creating their client. Each of three empty-database regressions +failed with missing tables before this setup repair. All 55 cases then passed on +a second set of genuinely empty databases, with the suites applying their own +migrations. The final additional rollover guard brings the three complete suites +to 56 passing cases (28 seeding, 6 queue-route, 22 watchdog), using those now +migrated fixtures. Server types/build pass at `7948136ef8`. The earlier 522-case +integration run remains evidence for its stated revision and unchanged consumers; +it is not relabeled as a current full-suite run. + +The first expanded fresh run passed 54 cases and failed the new backoff test's +cleanup expectation: after an entirely failed full page, a separate empty pass +ends the cycle before the next pass revisits rows. The corrected fixture permits +that explicit continuation and still checks complete recovery and the exact retry +deadline. Earlier failures remain recorded. Fresh final-head CI and reviews are +required before merge. 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 b032fcbb481..56ecd467c84 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -1132,3 +1132,94 @@ assertion failure; neither passing rerun is presented as a proven root cause. Server types/build pass. See the join-retention document for the full audit. Fresh exact-head reviews and CI remain required. All original duties remain open until their own accepted merge, including native-harness backup expiration. + +## 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. + + +Slice 16 is [#17510](https://github.com/paperclipai/paperclip/pull/17510). +Greptile identified a queue/item lock-order inversion. A real PostgreSQL test +reproduced a deadlock; the common queue-before-item order repairs manual add/remove +and seeding. The worker and queue route suites pass 27 cases after the fix, with +server typecheck/build passing. Other selected consumers passed on the initial +head (136 unique local cases including the new regression). Fresh-head checks are +required; the initial head also hit an unchanged environment cleanup paging test. +Embedded PostgreSQL is unsupported on this host, so its first local run skipped; +this is not counted as a pass. + +#17508 now has all checks accepted at `9c5c4fbf65628589b466ce5650e8969388213716` +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. + + +Approval queue follow-up: persistent projection failure reproduced 120 failed +transactions in two simulated minutes. Per-scope exponential backoff now caps +retries at one minute without discarding debt or letting repeated signals reset +the budget. A coordinator regression proves SQL silence during backoff and real +producer progress for another company. The candidate and seed writer share the +PR-product predicate. All 124 affected worker/queue/startup/retention/attention +cases pass, along with server types/build. One initial consumer command skipped +six queue tests because its DB variable was missing; the corrected run executes +and passes them. Fresh remote checks and review remain required after publication. + + +Slice 16 current parent: approval seeding integrates published join retention +`b41c51c70f`. Only the original 25 seeding paths differ from the tested parent; +all other files match it. The worker shares the first-linked-issue policy, and +generic writes preserve retention-plus-scoped-seeding signals while terminal +approval outcomes retain tool-review signals. All 522 cases in fifteen complete +suites, server types and production build pass at `4e725e2055`. An initial +integration typecheck caught incorrect gateway company references; those are +repaired with the existing session company. The final parent merge adds only +two documents. Fresh final-head CI/reviews and the parent's merge remain required. + + +Slice 16 now integrates landed join retention `8406047b4e` and targets master. +Its own diff is still the original 25 seeding paths; all database metadata matches +master. Continuing-arrival fairness, post-success backoff reset and outage backoff +through an empty rollover have real failed-then-passing regressions. External +seeding, queue and watchdog tests now migrate their own empty databases: all 55 +cases passed on three fresh fixtures, and the final added guard passes all 56 +cases on those migrated fixtures. Server types/build pass at `7948136ef8`. +The earlier cleanup-expectation failure and 522-case integration evidence retain +their own revisions. Fresh exact-head reviews and CI are the remaining merge gate. 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..9b9b7335d11 --- /dev/null +++ b/server/src/__tests__/approval-queue-seeding-work.test.ts @@ -0,0 +1,534 @@ +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, applyPendingMigrations, 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"; +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 { DELIVERY_QUEUES } from "../services/delivery-work-notifications.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) await applyPendingMigrations(external); + else 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_delete_gate on decision_queue_items`); + await db.execute(sql`drop function if exists approval_queue_delete_gate()`); + 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("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(); + 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("backs off persistent projection failures without forgetting the owed membership", async () => { + const service = approvalQueueSeedingService(db); + await service.sweep(); + const approval = await pending(); + service.request(new Set([companyId])); + 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 transactions = vi.spyOn(db, "transaction"); + try { + for (let second = 0; second < 120; second++) { + service.request(new Set([companyId])); + await service.sweep().catch(() => undefined); + await vi.advanceTimersByTimeAsync(1000); + } + expect(transactions.mock.calls.length).toBeLessThanOrEqual(8); + expect(transactions.mock.calls.length).toBeGreaterThan(1); + expect(await memberships()).toEqual([]); + expect(await service.next()).not.toBeNull(); + } finally { + transactions.mockRestore(); + await db.execute(sql`drop trigger approval_queue_fail on decision_queue_items`); + } + await vi.advanceTimersByTimeAsync(60_000); + await service.sweep(); + expect(await memberships()).toEqual([expect.objectContaining({ sourceId: approval.id })]); + expect(await service.next()).toBeNull(); + }); + + it("resets consecutive failure backoff after a successful batch with remaining debt", 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 service = approvalQueueSeedingService(db); + await db.execute(sql`create function approval_queue_fail() returns trigger language plpgsql as $$ begin raise exception 'projection unavailable'; end $$`); + const fail = () => db.execute(sql`create trigger approval_queue_fail before insert on decision_queue_items for each row execute function approval_queue_fail()`); + await fail(); + for (let attempt = 0; attempt < 7; attempt++) { + await expect(service.sweep()).rejects.toThrow(); + await vi.advanceTimersByTimeAsync(60_000); + } + await db.execute(sql`drop trigger approval_queue_fail on decision_queue_items`); + await service.sweep(); + expect(await memberships()).toHaveLength(1); expect(service.next()).not.toBeNull(); + await fail(); + const at = Date.now(); const clock = vi.spyOn(Date, "now").mockReturnValue(at); + try { + await expect(service.sweep()).rejects.toThrow("Approval queue seeding failed"); + expect(service.next()).toBe(at + 1000); + } finally { clock.mockRestore(); } + await db.execute(sql`drop trigger approval_queue_fail on decision_queue_items`); + await vi.advanceTimersByTimeAsync(60_000); await service.sweep(); + expect(service.next()).not.toBeNull(); await service.sweep(); + expect(await memberships()).toHaveLength(101); expect(service.next()).toBeNull(); + }); + + it("preserves outage backoff across an empty cursor rollover", async () => { + const target = await issue(); await product(target.id); + const rows = await db.insert(approvals).values(Array.from({ length: 100 }, () => ({ + companyId, type: "request_board_approval", payload: {}, + }))).returning(); + await db.insert(issueApprovals).values(rows.map(row => ({ companyId, issueId: target.id, approvalId: row.id }))); + const service = approvalQueueSeedingService(db); + 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()`); + await expect(service.sweep()).rejects.toThrow("Approval queue seeding failed"); + await vi.advanceTimersByTimeAsync(1000); await service.sweep(); + expect(await memberships()).toEqual([]); expect(service.next()).not.toBeNull(); + const at = Date.now(); const clock = vi.spyOn(Date, "now").mockReturnValue(at); + try { + await expect(service.sweep()).rejects.toThrow("Approval queue seeding failed"); + expect(service.next()).toBe(at + 2000); + } finally { clock.mockRestore(); } + await db.execute(sql`drop trigger approval_queue_fail on decision_queue_items`); + await vi.advanceTimersByTimeAsync(60_000); await service.sweep(); await service.sweep(); + expect(await memberships()).toHaveLength(100); expect(service.next()).toBeNull(); + }); + + it("keeps backoff SQL-quiet while another company's committed work proceeds", 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 + if new.company_id::text = 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("'" + companyId + "'")})`); + const approval = await pending(); + for (let count = 1; count <= 7; count++) { + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(errors.length).toBeGreaterThanOrEqual(count); }, { timeout: 5000 }); + await vi.waitFor(() => expect(running).toBe(false)); + } + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction"); + await vi.advanceTimersByTimeAsync(30_000); + expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); + reads.mockRestore(); writes.mockRestore(); + const otherId = randomUUID(); + await db.insert(companies).values({ id: otherId, name: "Healthy tenant", issuePrefix: `H${otherId.slice(0, 6)}` }); + const [otherIssue] = await db.insert(issues).values({ companyId: otherId, title: "Other PR", status: "todo" }).returning(); + await workProductService(db).createForIssue(otherIssue.id, otherId, { type: "pull_request", provider: "github", title: "Other PR", status: "open" }); + const otherApproval = await approvalRecords(db).create(otherId, { type: "request_board_approval", payload: {} }); + await issueApprovalService(db).link(otherIssue.id, otherApproval.id); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(await db.select().from(decisionQueueItems).where(eq(decisionQueueItems.companyId, otherId))) + .toEqual([expect.objectContaining({ sourceId: otherApproval.id })]); + }, { timeout: 5000 }); + expect(await memberships()).toEqual([]); + await db.execute(sql`drop trigger approval_queue_fail on decision_queue_items`); errors.length = 0; + await vi.advanceTimersByTimeAsync(60_000); + 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 vi.advanceTimersByTimeAsync(1000); + 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("retries a failed membership while new matching pages keep arriving", async () => { + const target = await issue(); await product(target.id); + const id = (n: number) => `00000000-0000-4000-8000-${String(n).padStart(12, "0")}`; + const append = async (start: number, count: number) => { + const rows = await db.insert(approvals).values(Array.from({ length: count }, (_, offset) => ({ + id: id(start + offset), companyId, type: "request_board_approval", payload: {}, + }))).returning(); + await db.insert(issueApprovals).values(rows.map(row => ({ companyId, issueId: target.id, approvalId: row.id }))); + }; + const service = approvalQueueSeedingService(db); await service.sweep(); + await append(1, 101); service.request(new Set([companyId])); + 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("'" + id(1) + "'")})`); + await expect(service.sweep()).rejects.toThrow("Failed query"); + await append(102, 100); service.request(new Set([companyId])); + await vi.advanceTimersByTimeAsync(1000); await service.sweep(); + await append(202, 100); service.request(new Set([companyId])); + await db.execute(sql`drop trigger approval_queue_fail on decision_queue_items`); + await service.sweep(); + expect((await memberships()).some(row => row.sourceId === id(1))).toBe(true); + for (let pass = 0; pass < 4 && service.next() !== null; pass++) await service.sweep(); + expect(await memberships()).toHaveLength(301); expect(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); + }); + + it("serializes manual removal and automatic seeding without a queue/item lock cycle", async () => { + const approval = await pending(); + const [link] = await db.select().from(issueApprovals).where(eq(issueApprovals.approvalId, approval.id)); + const queues = decisionQueueService(db); + const seed = () => queues.materializeSeedSources(companyId, [{ sourceKind: "approval", sourceId: approval.id, issueId: link.issueId }]); + await seed(); + await db.execute(sql`create function approval_queue_delete_gate() returns trigger language plpgsql as $$ begin + perform pg_advisory_xact_lock(hashtext(TG_ARGV[0])); return old; end $$`); + await db.execute(sql`create trigger approval_queue_delete_gate after delete on decision_queue_items for each row execute function approval_queue_delete_gate(${sql.raw("'" + companyId + "'")})`); + const blocker = await db.$client.reserve(); + await blocker`select pg_advisory_lock(hashtext(${companyId}))`; + let complete: Promise[]> | undefined; + try { + const removed = queues.removeItem({ ...mutation(), sourceKind: "approval", sourceId: approval.id }); + await vi.waitFor(async () => { + const [row] = await db.execute<{ count: number }>(sql`select count(*)::int as count from pg_stat_activity + where datname = current_database() and state = 'active' and wait_event = 'advisory'`); + expect(row.count).toBeGreaterThan(0); + }, { timeout: 5000 }); + complete = Promise.allSettled([removed, seed()]); + await vi.waitFor(async () => { + const [row] = await db.execute<{ count: number }>(sql`select count(*)::int as count from pg_stat_activity + where datname = current_database() and state = 'active' and wait_event_type = 'Lock' and wait_event <> 'advisory'`); + expect(row.count).toBeGreaterThan(0); + }, { timeout: 5000 }); + } finally { + await blocker`select pg_advisory_unlock_all()`; blocker.release(); + } + const results = await complete!; + expect(results.map(result => result.status), results.filter(result => result.status === "rejected").map(result => String(result.reason?.cause?.message ?? result.reason)).join("; ")).toEqual(["fulfilled", "fulfilled"]); + expect(await memberships()).toHaveLength(1); + }); + +}); diff --git a/server/src/__tests__/decision-queues-routes.test.ts b/server/src/__tests__/decision-queues-routes.test.ts index b1897d06469..9546708b3d4 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"; @@ -8,7 +9,7 @@ import { agents, approvals, companies, - createDb, + createDb, applyPendingMigrations, decisionQueueItems, decisionQueues, decisionTriage, @@ -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,9 @@ describeEmbeddedPostgres("decision queue routes", () => { let tempDb: Awaited> | null = null; beforeAll(async () => { - tempDb = await startEmbeddedPostgresTestDatabase("paperclip-decision-queues-"); - db = createDb(tempDb.connectionString); + if (external) await applyPendingMigrations(external); + else tempDb = await startEmbeddedPostgresTestDatabase("paperclip-decision-queues-"); + db = createDb(external ?? tempDb!.connectionString); }, 30_000); afterEach(async () => { @@ -59,6 +62,7 @@ describeEmbeddedPostgres("decision queue routes", () => { }); afterAll(async () => { + await db?.$client.end({ timeout: 1 }); await tempDb?.cleanup(); }); diff --git a/server/src/__tests__/environment-cleanup-work.test.ts b/server/src/__tests__/environment-cleanup-work.test.ts index 6a251dabdd0..5d71d2d4d42 100644 --- a/server/src/__tests__/environment-cleanup-work.test.ts +++ b/server/src/__tests__/environment-cleanup-work.test.ts @@ -228,7 +228,12 @@ describe.skipIf(!support.supported)("environment cleanup scheduling", () => { for (let i = 0; i < 33; i++) { const lease = await pending(); await environmentService(db).releaseLease(lease.id, "failed", { cleanupStatus: "success" }); } - await start(true); await settle(300_000); await settle(300_000); expect(inspections).toHaveBeenCalledTimes(33); await quiet(); + await start(true); + await vi.waitFor(async () => { + await settle(300_000); + expect(inspections).toHaveBeenCalledTimes(33); + }, { timeout: 5000 }); + await quiet(); }); it("keeps unknown historical resources scheduled until provider confirmation", async () => { const lease = await pending(); await environmentService(db).releaseLease(lease.id, "failed", { cleanupStatus: "success" }); diff --git a/server/src/__tests__/issue-comment-reopen-routes.test.ts b/server/src/__tests__/issue-comment-reopen-routes.test.ts index ba1df3c8b86..28c6900a190 100644 --- a/server/src/__tests__/issue-comment-reopen-routes.test.ts +++ b/server/src/__tests__/issue-comment-reopen-routes.test.ts @@ -1,8 +1,11 @@ import express from "express"; import request from "supertest"; import { getTableName } from "drizzle-orm"; -import { beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; -import { HttpError } from "../errors.js"; +import { beforeEach, describe, expect, it, vi } from "vitest"; +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); } @@ -376,15 +373,6 @@ function createScheduledRetry() { } describe("issue comment reopen routes", () => { - beforeAll(async () => { - // Load the route graph within the setup budget. A first-test import can - // time out and continue into the next test after its mocks are reset. - await Promise.all([ - import("../routes/issues.js"), - import("../middleware/index.js"), - ]); - }); - beforeEach(() => { vi.clearAllMocks(); mockIssueService.getById.mockReset(); @@ -3675,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__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index b08a2ed13e1..d0bb6ab1c33 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 { createSecretProposalsService } from "../services/secret-proposals.js"; @@ -337,6 +338,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(() => null), request: vi.fn() })), +})); + vi.mock("../services/approval-retention.js", async importOriginal => ({ ...await importOriginal(), approvalRetentionService: vi.fn(() => ({ sweep: vi.fn(async () => ({ archived: 0 })), next: vi.fn(async () => null) })), @@ -542,6 +548,21 @@ 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(registration![1].nextRunAt()).toBeNull(); + expect(await registration![1].hasPending()).toBe(false); + 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/__tests__/task-watchdogs-scheduler.test.ts b/server/src/__tests__/task-watchdogs-scheduler.test.ts index e5fbf969ab4..e5bfe047adb 100644 --- a/server/src/__tests__/task-watchdogs-scheduler.test.ts +++ b/server/src/__tests__/task-watchdogs-scheduler.test.ts @@ -7,7 +7,7 @@ import { agents, approvals, companies, - createDb, + createDb, applyPendingMigrations, documents, heartbeatRuns, issueComments, @@ -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,9 @@ describeEmbeddedPostgres("task watchdog scheduler", () => { let tempDb: Awaited> | null = null; beforeAll(async () => { - tempDb = await startEmbeddedPostgresTestDatabase("paperclip-task-watchdogs-scheduler-"); - db = createDb(tempDb.connectionString); + if (externalUrl) await applyPendingMigrations(externalUrl); + else tempDb = await startEmbeddedPostgresTestDatabase("paperclip-task-watchdogs-scheduler-"); + db = createDb(externalUrl ?? tempDb!.connectionString); }, 20_000); afterEach(async () => { @@ -64,6 +68,7 @@ describeEmbeddedPostgres("task watchdog scheduler", () => { }); afterAll(async () => { + await db?.$client.end({ timeout: 1 }); await tempDb?.cleanup(); }); @@ -96,16 +101,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, diff --git a/server/src/index.ts b/server/src/index.ts index a69a95b526d..17680e3d035 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"; @@ -1861,6 +1862,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 7bd14df166a..bfa7a38b3dc 100644 --- a/server/src/lib/approval-records.ts +++ b/server/src/lib/approval-records.ts @@ -1,5 +1,5 @@ import { notifyToolReviewChange } from "../services/tool-review-work-notifications.js"; -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"; @@ -39,7 +39,7 @@ export function approvalResolutionRecords(db: Db) { } const now = new Date(); - const updated = await writeApprovalChange(database, async tx => { + const updated = await writeApprovalChange(database, existing.companyId, async tx => { await notifyToolReviewChange(tx, existing.companyId); return tx.update(approvals) .set({ 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..6a4b3472799 --- /dev/null +++ b/server/src/services/approval-queue-seeding-work.ts @@ -0,0 +1,106 @@ +import { and, asc, desc, eq, gt, lte, sql, type SQL } from "drizzle-orm"; +import { approvals, companies, issues, issueWorkProducts, type Db } from "@paperclipai/db"; +import { firstApprovalIssueId } from "./approval-visibility.js"; +import { executionIssueCondition } from "./issue-visibility.js"; +import { pullRequestSeedProductCondition } from "./decision-queue-seed-rules.js"; +import { decisionQueueService } from "./decision-queues.js"; +import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "./delivery-work-notifications.js"; + +const linkedIssue = firstApprovalIssueId(); + +export function approvalQueueSeedingService(db: Db) { + const pending = new Map([[undefined, { generation: 0, failures: 0, retryAt: 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, failures: 0, retryAt: 0 }); + } + } + function candidates(database: Db, companyId: string | undefined, condition?: SQL, order = asc) { + 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"), 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 ${issueWorkProducts} where ${pullRequestSeedProductCondition(approvals.companyId)} + and ${issueWorkProducts.issueId} = ${issues.id}))`, + 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(order(approvals.id)); + } + async function sweep(signal?: AbortSignal) { + const scope = [...pending.entries()].find(([, state]) => state.retryAt <= Date.now()); + if (!scope) return; + const [companyId, state] = scope; + try { + const generation = state.generation; + state.end ??= (await candidates(db, companyId, undefined, desc).limit(1))[0]?.id; + const rows = state.end ? await candidates(db, companyId, and(lte(approvals.id, state.end), + 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, 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); } + state.cursor = row.id; + } + if (rows.length < 100) { state.cursor = undefined; state.end = undefined; } + const remaining = await candidates(db, companyId).limit(1); + if (!signal?.aborted && generation === state.generation && remaining.length === 0) pending.delete(companyId); + if (errors.length === 1) throw errors[0]; + if (errors.length) throw new AggregateError(errors, "Approval queue seeding failed"); + if (rows.length > 0 || remaining.length === 0) state.failures = 0; + state.retryAt = 0; + } catch (error) { + state.failures = Math.min(state.failures + 1, 7); + state.retryAt = Date.now() + Math.min(60_000, 1_000 * 2 ** (state.failures - 1)); + throw error; + } finally { + if (pending.has(companyId)) { + pending.delete(companyId); + pending.set(companyId, state); + } + } + } + function next() { + let earliest: number | null = null; + for (const state of pending.values()) earliest = earliest === null ? state.retryAt : Math.min(earliest, state.retryAt); + return earliest; + } + return { sweep, next, request }; +} + +export function registerApprovalQueueSeedingWork( + coordinator: ReturnType, + service: ReturnType, + canRun: () => boolean, +) { + let earliest = 0; + return coordinator.register(DELIVERY_QUEUES.approvalQueueSeeding, { + retryMs: 1_000, + pausePassiveMaintenance: true, + onSettled: service.request, + run: async signal => { + if (!canRun()) return; + if (Date.now() < earliest) return; + earliest = Date.now() + 1_000; + await service.sweep(signal); + }, + hasPending: async () => (service.next() ?? Infinity) <= Date.now(), + nextRunAt: () => { + const next = service.next(); + return next === null ? null : Math.max(earliest, next); + }, + }); +} diff --git a/server/src/services/approval-work-notifications.ts b/server/src/services/approval-work-notifications.ts new file mode 100644 index 00000000000..5c914477d15 --- /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, companyId: string) => notifyDeliveryWork(transaction, DELIVERY_QUEUES.approvalQueueSeeding, companyId); + +export async function notifyApprovalWork(transaction: object, companyId: string) { + await notifyDeliveryWork(transaction, DELIVERY_QUEUES.approvalRetention); + await notifyApprovalQueueWork(transaction, companyId); +} + +export function writeApprovalChange(db: Db, companyId: string, write: (transaction: Db) => Promise): Promise { + return db.transaction(async 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 1095766c3ef..cfa7624eee8 100644 --- a/server/src/services/approvals.ts +++ b/server/src/services/approvals.ts @@ -1,5 +1,5 @@ import { notifyToolReviewChange } from "./tool-review-work-notifications.js"; -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"; @@ -55,7 +55,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() @@ -68,7 +68,7 @@ export function approvalRecords(db: Db) { const now = new Date(); const [existing] = await db.select({ companyId: approvals.companyId }).from(approvals).where(eq(approvals.id, id)); if (!existing) return null; - const updated = await writeApprovalChange(db, async tx => { + const updated = await writeApprovalChange(db, existing.companyId, async tx => { await notifyToolReviewChange(tx, existing.companyId); return tx.update(approvals) .set({ @@ -91,7 +91,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", @@ -112,7 +112,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 aba74d6e0c6..ccff957a9ed 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,9 @@ async function markApprovalStatus( decidedByUserId: string, ) { if (!approvalId) return; - await notifyRetentionWork(db, "approval"); + 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({ @@ -297,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 @@ -354,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) @@ -457,7 +460,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica windowEnd: end, }); - if (thresholdType === "hard") await notifyRetentionWork(db, "approval"); + 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 e6f716faca3..ba9d8f758ac 100644 --- a/server/src/services/company-deletion.ts +++ b/server/src/services/company-deletion.ts @@ -97,7 +97,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-queue-seed-rules.ts b/server/src/services/decision-queue-seed-rules.ts new file mode 100644 index 00000000000..c46b0cde464 --- /dev/null +++ b/server/src/services/decision-queue-seed-rules.ts @@ -0,0 +1,6 @@ +import { and, eq, type SQLWrapper } from "drizzle-orm"; +import { issueWorkProducts } from "@paperclipai/db"; + +export function pullRequestSeedProductCondition(companyId: string | SQLWrapper) { + return and(eq(issueWorkProducts.companyId, companyId), eq(issueWorkProducts.type, "pull_request")); +} diff --git a/server/src/services/decision-queues.ts b/server/src/services/decision-queues.ts index d529e1d0a8d..b0c871aa085 100644 --- a/server/src/services/decision-queues.ts +++ b/server/src/services/decision-queues.ts @@ -1,3 +1,5 @@ +import { pullRequestSeedProductCondition } from "./decision-queue-seed-rules.js"; +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 +45,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; @@ -367,6 +377,12 @@ export function decisionQueueService(db: Db) { .then((rows) => rows[0] ?? null); } + async function lockQueue(database: Db, companyId: string, id: string) { + const [queue] = await database.select({ id: decisionQueues.id }).from(decisionQueues) + .where(and(eq(decisionQueues.companyId, companyId), eq(decisionQueues.id, id))).for("update"); + if (!queue) throw notFound("Decision queue not found"); + } + async function visibleItems(companyId: string, queueId: string, authActor: AuthorizationActor) { const rows = await db.select().from(decisionQueueItems) .where(and(eq(decisionQueueItems.companyId, companyId), eq(decisionQueueItems.queueId, queueId))) @@ -380,6 +396,103 @@ 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( + pullRequestSeedProductCondition(companyId), + 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 +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, input.companyId); const inserted = await txDb.insert(decisionQueues).values({ companyId: input.companyId, key: input.key, @@ -449,6 +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, 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); @@ -491,6 +606,7 @@ export function decisionQueueService(db: Db) { await requireSourceRead(db, input.authActor, input.companyId, input.sourceKind, input.sourceId); return db.transaction(async (tx) => { const txDb = tx as unknown as Db; + await lockQueue(txDb, input.companyId, queue.id); const inserted = await txDb.insert(decisionQueueItems).values({ companyId: input.companyId, queueId: queue.id, @@ -556,6 +672,8 @@ 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, input.companyId); const removed = await txDb.delete(decisionQueueItems).where(and( eq(decisionQueueItems.companyId, input.companyId), eq(decisionQueueItems.queueId, queue.id), @@ -682,101 +800,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 f6d1d6bb5f7..9a42db08f16 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -9,6 +9,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 b00133898b0..682262ca899 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"; @@ -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, @@ -128,7 +128,7 @@ export function issueApprovalService(db: Db) { unlink: async (issueId: string, approvalId: string) => { const { issue } = await assertIssueAndApprovalSameCompany(issueId, approvalId); - await writeApprovalChange(db, tx => tx + 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 53f3cff74a6..5c559559a70 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, notifyDecisionTargetIssueRemoval } from "./decision-targets.js"; import { assertOnboardingFirstTaskCreationAllowed } from "./onboarding-first-task-admission.js"; @@ -11391,7 +11391,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, receiptExisting.companyId); } const updated = await tx .update(issues) @@ -11800,7 +11800,8 @@ export function issueService(db: Db) { try { await notifyTerminalWorkspaceWork(tx, `issue:${id}`); await requestRestorePointDeletion(tx, { sourceIssueId: id }); - await notifyRetentionWork(tx, "approval"); + const [owner] = await tx.select({ companyId: issues.companyId }).from(issues).where(eq(issues.id, id)); + if (owner) await notifyApprovalWork(tx, owner.companyId); await notifyDecisionTargetIssueRemoval(tx, id); 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..ffedb0b1a4d 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"); +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 372c8d4f45d..9ea1884369c 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 { notifyDecisionTargetIssueRemoval } from "./decision-targets.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, createdIssue.companyId); await notifyDecisionTargetIssueRemoval(txDb, createdIssue.id); 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 ae81998e947..c5c9cdb07b6 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 { bindVerifiedToolApprovalArguments, claimToolTestApproval, releaseToolApproval, persistToolTestOutcome, startToolInvocation, type ToolApprovalClaim } from "./tool-approval-execution.js"; @@ -2650,7 +2650,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, @@ -2676,7 +2676,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 e4a8cbb5a10..4aa44ef7f49 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, companyId); + } 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, existing.companyId); + } await validateExecutionWorkspace(tx, existing.companyId, patch.executionWorkspaceId); @@ -378,6 +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) => { + 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}`); } @@ -387,9 +397,12 @@ 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}`); + if (existing?.type === "pull_request") { + await notifyTerminalWorkspaceWork(tx, `issue:${existing.issueId}`); + await notifyApprovalQueueWork(tx, existing.companyId); + } return tx.delete(issueWorkProducts).where(eq(issueWorkProducts.id, id)) .returning().then(rows => rows[0] ?? null); });