fix(server): wake approval queue seeding from committed changes (#17510)

## Thinking Path

> - Paperclip manages work and human decisions for AI-agent companies.
> - Automatic decision queues store membership for pending attention
sources.
> - The old attention maintenance scan created that membership as a side
effect.
> - Moving approval retention out of the scan also removed approval
queue discovery.
> - Queue API consumers do not build the attention feed first, so they
can miss pending PR approvals.
> - This pull request gives approval queue discovery its own worker and
commit signals.
> - The worker becomes quiet after it saves every missing membership.

## Linked Issues or Issue Description

Refs #17497. Stacked on #17508, following #17505.

This is a correctness dependency for the approval retention migration in
#17497. Do not treat that migration as complete without this
queue-population path. A real PostgreSQL regression found no PR queue
when a pending approval had a linked PR work product and the maintenance
feed excluded approvals.

Related: #13364 proposes pruning stale seeded memberships. This PR
preserves existing membership retention and does not implement pruning.
#15149 documents the decision-queue APIs.

## What Changed

- Add an approval queue worker that selects missing PR-queue memberships
from existing durable records. Bound each pass to 100 rows. Continue
past failed rows and revisit them on retry.
- Reuse the queue service's seed writer through a normalized source
contract. Keep queue rules, uniqueness, and audit writes in that
service. Keep feed readers compatible.
- Wake from committed approval and link changes, issue/company
visibility, PR work-product changes and imports, and queue-rule changes.
Reuse the existing transaction boundaries.
- Recheck approval status under a row lock. Serialize queue-rule changes
with the seed writer. Preserve required recovery during drain and
shutdown.
- Register startup reconstruction under the existing maintenance switch.
Leave other attention-source seeding with its current owner until its
own migration.
- Document the complete producer audit and stack dependency. Add a
validated external PostgreSQL option to the existing queue route suite
so it runs on hosts without embedded database support.

## Verification

Current tested revision: `7948136ef8`. This PR targets master after
#17508 merged as `8406047b4e`. Normal ancestry reconciliation preserved
the complete tested integration tree, before the review repairs below.
Only the original 25 seeding paths differ from master, and database
schema/migration metadata is identical.

- **56 cases pass in the three complete affected suites:** 28 approval
seeding, 6 queue routes and 22 task-watchdog cases. Server typecheck and
production build pass.
- Three real PostgreSQL retry regressions failed then passed: new
matching arrivals cannot postpone a recovered failure indefinitely;
successful partial progress resets old failure penalties; an empty
cursor rollover does not reset outage backoff. Exact one- and two-second
retry assertions, eventual complete recovery and existing
capped-backoff/company isolation tests pass.
- Three genuinely empty external databases failed before setup repair
because tables were missing. Each suite now applies migrations itself.
All 55 cases passed on a second set of fresh databases; adding the final
rollover guard then passed all 56 cases using those now-migrated
fixtures. No production migration or table is added.
- The initial expanded fresh run passed 54 cases but failed a cleanup
expectation that omitted a required cursor rollover pass. The fixture
now permits that explicit continuation while preserving exact deadline
and complete-membership assertions. All failure records remain in
`doc/idle-approval-queue-seeding.md`.
- The prior integrated revision `4e725e2055` passed **522 cases in
fifteen complete producer/consumer suites**, plus types/build. Its
unchanged consumer code and historical deadlock, cross-company
amplification, backoff, gateway integration and setup evidence remain
documented; those counts are not relabeled as current acceptance.

Commands: server Vitest for approval queue seeding, queue routes and
task watchdogs; server typecheck/build. Fresh exact-head
Greptile/Ponytail reviews and all CI checks remain required. No full
local repository-suite, deployed suspension or cost claim is made.

## Risks

- Signals cover the supported in-process service writers. Independent
SQL writers and multiple server replicas need a separate notification
contract.
- The source predicate must stay aligned with unscoped attention
maintenance. The first linked issue, company scope, pending status, and
execution visibility are tested. A pending approval itself keeps its
linked external conversation active.
- Existing enabled seed rules can re-add manually removed matching
memberships. This preserves the old scan's behavior. Completed or
archived membership is not pruned.
- Queue seeding and retention are separate owners. Other attention
sources still need their own event-driven discovery before the aggregate
scan can be retired.
- This PR does not extend Cloud wake coverage, relax sleep admission, or
claim measured host/database suspension.

## Model Used

OpenAI Codex, GPT-6 family, with reasoning and code execution. The exact
serving model identifier and context window were not exposed.

## Checklist

- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either (a) linked existing issues with `Fixes: #` / `Closes
#` / `Refs #` OR (b) described the issue in-PR following the relevant
issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip`
URLs)
- [x] My branch name describes the change (e.g. `docs/...`, `fix/...`)
and contains no internal Paperclip ticket id or instance-derived details
- [x] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [ ] All Paperclip CI gates are green
- [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge

---------

Co-authored-by: Paperclip <noreply@paperclip.ing>

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.

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.

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<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
+  let db: ReturnType<typeof createDb>, companyId: string;
+  let coordinator: ReturnType<typeof createDeliveryWorkCoordinator> | 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<void>(resolve => { observed = resolve; });
+    const barrier = new Promise<void>(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<void>(resolve => { release = resolve; }), admitted = new Promise<void>(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<void>(resolve => { release = resolve; }), entered = new Promise<void>(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<PromiseSettledResult<unknown>[]> | 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);
+  });
+
+});

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<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | 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();
   });
 

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" });

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<string, unknown>,
 ) {
-  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"),
     );

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<typeof import("../services/approval-queue-seeding-work.js")>(),
+  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<typeof import("../services/approval-retention.js")>(),
   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();

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<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | 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<typeof issues.$inferInsert> = {}) {
     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,

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),

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({

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<string | undefined, { cursor?: string; end?: string; generation: number; failures: number; retryAt: number }>([[undefined, { generation: 0, failures: 0, retryAt: 0 }]]);
+  function request(companyIds: ReadonlySet<string> | 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<typeof createDeliveryWorkCoordinator>,
+  service: ReturnType<typeof approvalQueueSeedingService>,
+  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);
+    },
+  });
+}

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<T>(db: Db, companyId: string, write: (transaction: Db) => Promise<T>): Promise<T> {
+  return db.transaction(async tx => {
+    await notifyApprovalWork(tx, companyId);
+    return write(tx as unknown as Db);
+  });
+}

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<typeof approvals.$inferInsert, "companyId">) =>
-      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",

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)

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)

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));

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"));
+}

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<string, DecisionQueueSeedSource[]>();
+    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<string, AttentionItem[]>();
-      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,
+    }))),
   };
 }

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",

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) => ({

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)

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<T>(db: Db, write: (transaction: Db) => Promise<T>): Promise<T> {
-  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");
 }
 

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));
         }

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,

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<typeof issueWorkProducts.$inferInsert, "issueId" | "companyId">) => {
       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);
       });