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

Co-Authored-By: Paperclip <noreply@paperclip.ing>

doc/idle-approval-queue-seeding.md

@@ -0,0 +1,114 @@
+# Approval queue seeding without retention polling
+
+## Obligation and stack dependency
+
+The attention maintenance scan does two different jobs: retention and automatic
+queue membership. Its call to `materializeSeededQueues` adds pending approvals to
+the PR queue when the first linked issue has a pull-request work product. Queue
+API readers use saved membership; they do not first build the attention feed.
+Removing approvals from maintenance therefore lost automatic queue population.
+A real PostgreSQL regression reproduced an empty queue with a pending linked
+approval and PR work product, without an attention reader.
+
+This slice owns automatic PR-queue membership for approvals. It depends on the
+approval producer boundaries introduced in #17497 and is required with that
+retention migration before treating the combined behavior as complete. Join
+requests have no issue association or matching seed rule. Other attention-source
+seed discovery remains in the existing feed scan and must migrate with an
+explicit owner before those sources leave that scan.
+
+## Source of truth and ownership
+
+Existing approvals, issue approvals, issue work products, queues, and queue items
+are the only durable state. No new table, outbox, external API, or provider branch
+is introduced. `approvalQueueSeedingService` selects missing memberships for:
+
+- A pending approval in an active company.
+- Its first linked issue, ordered by issue ID, in the same company and eligible
+  under the attention feed's execution-issue visibility predicate.
+- A pull-request work product belonging to that issue and company.
+- A missing PR queue or an enabled seed rule without this approval's membership.
+
+Disabled rules, nonmatching approvals, completed approvals, and existing seeded
+history are passive. A pending approval itself prevents its linked Slack/Speko
+conversation from projecting as idle, so this source does not depend on every
+other conversation-state producer. Hidden/harness/conversation-container issue
+changes retain their explicit signals.
+
+The queue service owns seed matching, creation, membership uniqueness, and audit
+writes. The feed and worker use the same writer through a small normalized source
+contract. A worker does not construct a presentation-layer AttentionItem. Queue
+reads keep their existing source authorization checks. A queue-rule update and
+seeding serialize on the existing queue row. Existing memberships are not pruned
+when an approval resolves. This preserves current behavior; pruning is a separate
+change, including the proposal in #13364.
+
+## Producer audit
+
+| Producer | Commit boundary and effect |
+| --- | --- |
+| Approval create, cancel, revision, resubmit, resolve | `writeApprovalChange` notifies retention and queue seeding before writes in the owning transaction. |
+| Shared approval records, budget approvals, formal tool approvals | The same approval notification covers their existing transactions. |
+| Issue approval link, unlink, link-many, tool-created links | Existing approval transaction helper. Link removal can expose a different first issue. |
+| Issue visibility changes and issue deletion | Existing approval notification before the update or cascading delete. Routine failure cleanup retains this signal. |
+| Company activation, pause, archive, deletion and budget pause/resume | Existing company notification wakes both approval consumers. |
+| Work-product create, type/issue update, import batch, delete | The owning work-product transaction notifies seeding when a pull-request row is involved. |
+| PR queue creation, rule update and manual approval membership removal | The queue mutation transaction notifies seeding. An enabled rule can re-add a still-matching manually removed item, as the old scan did. |
+
+Other direct work-product writes only create artifacts or update title, metadata,
+or trust. They cannot create a pull-request match. Supported writes use these
+in-process service boundaries. Independent SQL writers and multiple server
+replicas are not covered by local transaction signals.
+
+Retention-only keep/archive/revive changes do not create queue projection debt.
+The old seed writer ran before retention filtering, so archived or kept approvals
+still follow the same seeding rule. User-specific dismissals do not filter the
+unscoped maintenance feed. No new deadline is inferred from those display states.
+
+## Worker lifecycle
+
+Startup reconstructs missing memberships. A committed producer wakes the shared
+coordinator. Each sweep processes at most 100 approvals, with an ID cursor that
+advances past failures and rotates to revisit them. Each row is rechecked under
+an approval lock before the queue writer runs. One bad row cannot permanently
+hide later candidates. Atomic existing membership uniqueness prevents duplicate
+rows and duplicate insertion audit events.
+
+After a sweep, another lookup decides whether actual projection debt remains.
+Only that debt or a failed attempt retains a one-second retry. Once reconciled,
+there is no deadline and no recurring worker SQL. The shared deadline worker
+preserves requested work during drain, does not run passive domain SQL while
+paused, and reconstructs from rows after restart. Shutdown aborts between rows
+and waits for an admitted transaction; unprocessed rows remain recoverable.
+Registration follows the existing heartbeat-maintenance enablement switch.
+
+This slice does not extend Cloud alarm coverage or loosen host sleep admission.
+It proves a worker's quiet/reactivation behavior, not deployed host or database
+suspension. The remaining attention maintenance interval is still present for
+other source owners.
+
+## Verification
+
+- Before the fix, the PostgreSQL regression found no PR queue after the normal
+  maintenance path excluded approvals. After the fix, the queue API finds it
+  without a full attention read.
+- Twenty real-database worker tests cover empty-to-active-to-quiet behavior,
+  startup and completed history, actual approval/link/work-product producers,
+  batch import, first-link selection, company isolation, hidden targets, queue
+  rule enablement, company activation, resubmission, outer commit and rollback,
+  first write during drain, database failure through drain, multiple pages,
+  failed-page progress and retry, concurrent resolution, archived/kept history, shutdown/restart, and an admitted transaction at shutdown.
+- 115 existing tests pass across queue routes, attention, approval retention,
+  approval service, work products, and startup wiring. The shared writer still
+  passes existing plan/question/PR seed tests and authorization tests.
+- Workspace typechecks and the production build pass. Fresh PR checks are
+  tracked in the PR and must pass before readiness is claimed.
+- The actual pending lookup on 40,000 completed approvals plus one pending PR
+  approval uses the existing partial pending-retention index. The probe runs in
+  a rolled-back transaction; execution took 0.094 ms on this local fixture. No new index or migration is needed for this case.
+
+The first expanded test run caught two fixture errors: a hand-inserted queue
+omitted required creator attribution, and a failure test expected no idle hold
+while committed projection work was still unresolved. The corrected test waits
+for the active attempt to finish and checks passive SQL directly while preserving
+that required hold. Neither required weakening the worker's recovery contract.

doc/plans/2026-10-24-recurring-server-sleep-stack.md

@@ -888,3 +888,29 @@ claimed ready based only on local tests, and no merge is authorized.
 After join retention, migrate the remaining attention source owners separately:
 decisions, interactions, recovery actions, reviews, and other source kinds. Keep
 the full original server inventory and later aggregate-tick retirement in scope.
+
+## Slice 16: approval PR-queue membership
+
+The source audit found automatic seeded-queue materialization hidden inside the
+attention retention scan. A PostgreSQL regression failed after approval exclusion:
+queue API readers saw no PR queue despite a pending linked approval and PR work
+product. This is a separate obligation from retention, not grounds to leave a
+new permanent scan or to put queue policy into retention.
+
+The new approval queue worker uses existing durable records and the queue service's
+normalized seed writer. It gets committed signals from all approval/link producers,
+issue/company visibility changes, work-product writes, and queue-rule changes.
+Missing membership is recovery debt; existing membership and disabled rules are
+quiet. Twenty real-database lifecycle cases and 115 existing consumer/startup
+tests pass, as do workspace typechecks and the production build. See
+[approval queue seeding](../idle-approval-queue-seeding.md). Fresh CI/review
+remain required before readiness. This slice is a correctness dependency of
+approval retention #17497; do not accept that migration as functionally complete
+without preserving queue population. Other source seeding remains to be migrated.
+
+#17505 now has all latest-head checks accepted at `28a197cdfdab7977eecf591df39612009d725e17`
+(Paperclip Review skipped), with Greptile 5/5. #17508 at
+`9c5c4fbf65628589b466ce5650e8969388213716` has Greptile 5/5 and no unresolved
+threads. Its unchanged login-cleanup consume test failed in CI; all 16 local
+login-cleanup tests pass. One justified retry of failed run 40139410144 was
+requested. Do not mark that PR green until the retry completes successfully.

server/src/__tests__/approval-queue-seeding-work.test.ts

@@ -0,0 +1,293 @@
+import { randomUUID } from "node:crypto";
+import { eq, sql } from "drizzle-orm";
+import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
+import { activityLog, approvals, companies, createDb, decisionQueueItems, decisionQueues, decisionRetention, decisionTriageEvents, issueApprovals, issues, issueWorkProducts, type Db } from "@paperclipai/db";
+import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
+import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
+import { approvalRecords } from "../services/approvals.js";
+import { approvalRetentionService } from "../services/approval-retention.js";
+import { approvalQueueSeedingService, registerApprovalQueueSeedingWork } from "../services/approval-queue-seeding-work.js";
+import { attentionService } from "../services/attention.js";
+import { companyService } from "../services/companies.js";
+import { decisionQueueService } from "../services/decision-queues.js";
+import { issueApprovalService } from "../services/issue-approvals.js";
+import { issueService } from "../services/issues.js";
+import { workProductService } from "../services/work-products.js";
+import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
+import { idleWorkSnapshot } from "../services/task-admission.js";
+
+const external = resolveExternalTestDatabaseUrl("PAPERCLIP_APPROVAL_QUEUE_TEST_DATABASE_URL");
+const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport();
+(support.supported ? describe : describe.skip)("approval queue seeding", () => {
+  let temporary: Awaited<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) temporary = await startEmbeddedPostgresTestDatabase("approval-queue-");
+    db = createDb(external ?? temporary!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 });
+  }, 30_000);
+  beforeEach(async () => {
+    companyId = randomUUID(); draining = false; errors.length = 0;
+    await db.insert(companies).values({ id: companyId, name: "Approval queue", issuePrefix: `Q${companyId.slice(0, 6)}` });
+    await Promise.all(Array.from({ length: 6 }, () => db.execute(sql`select pg_sleep(0.01)`)));
+    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: true });
+  });
+  afterEach(async () => {
+    await coordinator?.stop(); coordinator = undefined;
+    vi.useRealTimers(); vi.restoreAllMocks();
+    await db.execute(sql`drop trigger if exists approval_queue_fail on decision_queue_items`);
+    await db.execute(sql`drop function if exists approval_queue_fail()`);
+    await db.delete(decisionTriageEvents); await db.delete(decisionQueueItems); await db.delete(decisionQueues);
+    await db.delete(decisionRetention); await db.delete(activityLog); await db.delete(issueApprovals);
+    await db.delete(issueWorkProducts); await db.delete(approvals); await db.delete(issues); await db.delete(companies);
+    expect(errors).toEqual([]);
+  });
+  afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); });
+  const authActor = () => ({ type: "board" as const, source: "local_implicit" as const, userId: "board-user", companyIds: [companyId], isInstanceAdmin: true });
+  const actor = { actorType: "user" as const, actorId: "board-user", agentId: null, userId: "board-user", runId: null, agentApiKeyId: null, responsibleUserId: "board-user" };
+  const mutation = () => ({ companyId, key: "prs", authActor: authActor(), actor });
+  const memberships = () => db.select().from(decisionQueueItems).where(eq(decisionQueueItems.companyId, companyId));
+  const createApproval = (database: Db = db) => approvalRecords(database).create(companyId, { type: "request_board_approval", payload: {} });
+  async function issue(hiddenAt: Date | null = null) {
+    const [row] = await db.insert(issues).values({ companyId, title: "Review PR", status: "todo", hiddenAt }).returning();
+    return row;
+  }
+  const product = (issueId: string, type = "pull_request") => workProductService(db).createForIssue(issueId, companyId, { type, provider: "github", title: "PR", status: "open" });
+  async function pending(target?: string) {
+    const targetId = target ?? (await issue()).id;
+    if (!target) await product(targetId);
+    const approval = await createApproval();
+    await issueApprovalService(db).link(targetId, approval.id);
+    return approval;
+  }
+  async function start(service = approvalQueueSeedingService(db)) {
+    coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining, canReconcile: () => true, onError: error => errors.push(error) });
+    await registerApprovalQueueSeedingWork(coordinator, service, () => !draining).ready;
+    return service;
+  }
+  async function settle() {
+    await vi.waitFor(async () => {
+      await vi.advanceTimersByTimeAsync(1_000);
+      expect(idleWorkSnapshot().active).toBe(0);
+    }, { timeout: 5000 });
+  }
+  async function seeded(id: string) {
+    await vi.waitFor(async () => {
+      await vi.advanceTimersByTimeAsync(1_000);
+      expect(await memberships()).toEqual(expect.arrayContaining([expect.objectContaining({ sourceKind: "approval", sourceId: id })]));
+    }, { timeout: 5000 });
+  }
+  async function quiet() {
+    await settle(); expect(coordinator!.nextWakeAt()).toBeNull();
+    const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction"), executes = vi.spyOn(db, "execute");
+    await vi.advanceTimersByTimeAsync(300_000);
+    expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); expect(executes).not.toHaveBeenCalled();
+    reads.mockRestore(); writes.mockRestore(); executes.mockRestore();
+  }
+
+  it("populates the PR queue without an attention reader after approvals leave retention maintenance", async () => {
+    await start(); await quiet();
+    const approval = await pending();
+    await approvalRetentionService(db).sweep();
+    await attentionService(db).list(companyId, { all: true, allowUnscopedAll: true, excludeSources: ["approval", "join_request"] });
+    await seeded(approval.id);
+    expect((await decisionQueueService(db).list(companyId, authActor())).map(queue => queue.key)).toContain("prs");
+    expect(await decisionQueueService(db).listItems(companyId, "prs", authActor())).toEqual(expect.arrayContaining([
+      expect.objectContaining({ sourceKind: "approval", sourceId: approval.id }),
+    ]));
+    await quiet();
+  });
+
+  it("recovers startup projection debt and stays quiet with already seeded and completed history", async () => {
+    const first = await pending(), completed = await pending();
+    await approvalRecords(db).resolveApproval(completed.id, "approved", "board-user", null);
+    await start(); await seeded(first.id); await quiet();
+    expect((await memberships()).map(row => row.sourceId)).toEqual([first.id]);
+    await coordinator!.stop(); coordinator = undefined;
+    await start(); await quiet();
+    expect(await db.select().from(decisionTriageEvents).where(eq(decisionTriageEvents.action, "queue_item.seeded"))).toHaveLength(1);
+  });
+
+  it("wakes when a work product creates the first PR match", async () => {
+    const target = await issue(), approval = await pending(target.id);
+    await start(); await quiet(); expect(await memberships()).toEqual([]);
+    const row = await product(target.id, "artifact"); await quiet();
+    await workProductService(db).update(row!.id, { type: "pull_request" });
+    await seeded(approval.id); await quiet();
+  });
+
+  it("wakes from batch-imported PR work products", async () => {
+    const target = await issue(), approval = await pending(target.id);
+    await start(); await quiet();
+    await workProductService(db).createManyForImport([{ companyId, issueId: target.id, type: "pull_request", provider: "github", title: "Imported PR", status: "open", reviewState: "none", isPrimary: false, healthStatus: "unknown" }]);
+    await seeded(approval.id); await quiet();
+  });
+
+  it("honors hidden targets and wakes when visibility changes", async () => {
+    const target = await issue(new Date()); await product(target.id); const approval = await pending(target.id);
+    await start(); await quiet(); expect(await memberships()).toEqual([]);
+    await issueService(db).update(target.id, { hiddenAt: null }); await seeded(approval.id); await quiet();
+  });
+
+  it("uses the first linked issue and reevaluates after unlink", async () => {
+    const first = await issue(), second = await issue();
+    const [withoutPr, withPr] = [first, second].sort((a, b) => a.id.localeCompare(b.id));
+    await product(withPr.id);
+    const approval = await pending(withoutPr.id); await issueApprovalService(db).link(withPr.id, approval.id);
+    await start(); await quiet(); expect(await memberships()).toEqual([]);
+    await issueApprovalService(db).unlink(withoutPr.id, approval.id); await seeded(approval.id); await quiet();
+  });
+
+  it("does not infer PRs from a later link or another company's work product", async () => {
+    const target = await issue(), approval = await pending(target.id);
+    const foreignId = randomUUID(); await db.insert(companies).values({ id: foreignId, name: "Other", issuePrefix: "OTHER" });
+    await db.insert(issueWorkProducts).values({ companyId: foreignId, issueId: target.id, type: "pull_request", provider: "github", title: "Invalid foreign product", status: "open" });
+    await start(); await quiet(); expect(await memberships()).toEqual([]);
+    await product(target.id); await seeded(approval.id); await quiet();
+  });
+
+  it("keeps disabled seeds quiet and wakes when the queue rule is enabled", async () => {
+    await db.insert(decisionQueues).values({ companyId, key: "prs", title: "PRs", createdByType: "system", seedRulesEnabled: false });
+    const approval = await pending(); await start(); await quiet(); expect(await memberships()).toEqual([]);
+    await decisionQueueService(db).update({ ...mutation(), patch: { seedRulesEnabled: true } });
+    await seeded(approval.id); await quiet();
+  });
+
+  it("waits for company activation", async () => {
+    const approval = await pending(); await companyService(db).update(companyId, { status: "paused" });
+    await start(); await quiet(); expect(await memberships()).toEqual([]);
+    await companyService(db).update(companyId, { status: "active" }); await seeded(approval.id); await quiet();
+  });
+
+  it("waits for resubmission and never removes retained membership after approval", async () => {
+    const approval = await pending(); await approvalRecords(db).requestRevision(approval.id, "board-user");
+    await start(); await quiet(); expect(await memberships()).toEqual([]);
+    await approvalRecords(db).resubmit(approval.id); await seeded(approval.id);
+    await approvalRecords(db).resolveApproval(approval.id, "approved", "board-user", null); await quiet();
+    expect(await memberships()).toHaveLength(1);
+  });
+
+  it("waits for outer commit and discards rolled-back producers", async () => {
+    const target = await issue(); await product(target.id); await start(); await quiet();
+    await expect(db.transaction(async tx => {
+      const approval = await createApproval(tx as unknown as Db);
+      await issueApprovalService(tx as unknown as Db).link(target.id, approval.id);
+      await vi.advanceTimersByTimeAsync(5_000); expect(await memberships()).toEqual([]);
+      throw new Error("rollback");
+    })).rejects.toThrow("rollback");
+    await quiet();
+    let id = "";
+    await db.transaction(async tx => {
+      id = (await createApproval(tx as unknown as Db)).id;
+      await issueApprovalService(tx as unknown as Db).link(target.id, id);
+      await vi.advanceTimersByTimeAsync(5_000); expect(await memberships()).toEqual([]);
+    });
+    await seeded(id); await quiet();
+  });
+
+  it("retains the first write during drain without passive SQL", async () => {
+    await start(); await quiet(); draining = true;
+    const approval = await pending();
+    const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction");
+    await vi.advanceTimersByTimeAsync(60_000);
+    expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled();
+    reads.mockRestore(); writes.mockRestore();
+    expect(await memberships()).toEqual([]); draining = false;
+    await seeded(approval.id); await quiet();
+  });
+
+  it("recovers a failed projection after drain and stops retrying once it commits", async () => {
+    const service = approvalQueueSeedingService(db), sweep = service.sweep;
+    let running = false;
+    service.sweep = async signal => { running = true; try { await sweep(signal); } finally { running = false; } };
+    await start(service); await quiet();
+    await db.execute(sql`create function approval_queue_fail() returns trigger language plpgsql as $$ begin raise exception 'projection unavailable'; end $$`);
+    await db.execute(sql`create trigger approval_queue_fail before insert on decision_queue_items for each row execute function approval_queue_fail()`);
+    const approval = await pending();
+    await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1_000); expect(errors.length).toBeGreaterThan(0); }, { timeout: 5000 });
+    draining = true;
+    await vi.waitFor(() => expect(running).toBe(false), { timeout: 5000 });
+    const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(60_000); expect(reads).not.toHaveBeenCalled(); reads.mockRestore();
+    await db.execute(sql`drop trigger approval_queue_fail on decision_queue_items`); errors.length = 0;
+    draining = false; await seeded(approval.id); await quiet();
+  });
+
+  it("drains multiple pages without duplicating memberships or audit events", async () => {
+    const target = await issue(); await product(target.id);
+    const rows = await db.insert(approvals).values(Array.from({ length: 101 }, () => ({ companyId, type: "request_board_approval", payload: {} }))).returning();
+    await db.insert(issueApprovals).values(rows.map(row => ({ companyId, issueId: target.id, approvalId: row.id })));
+    await start();
+    await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1_000); expect(await memberships()).toHaveLength(101); }, { timeout: 10_000 });
+    await quiet();
+    await attentionService(db).list(companyId, { all: true, allowUnscopedAll: true }); await quiet();
+    expect(await memberships()).toHaveLength(101);
+    expect(await db.select().from(decisionTriageEvents).where(eq(decisionTriageEvents.action, "queue_item.seeded"))).toHaveLength(101);
+  });
+
+  it("preserves the seed rule after manual removal of a still-matching item", async () => {
+    const approval = await pending(); await start(); await seeded(approval.id); await quiet();
+    await decisionQueueService(db).removeItem({ ...mutation(), sourceKind: "approval", sourceId: approval.id });
+    await seeded(approval.id); await quiet(); expect(await memberships()).toHaveLength(1);
+  });
+  it("pages past failed projections and revisits them after recovery", async () => {
+    const target = await issue(); await product(target.id);
+    const rows = await db.insert(approvals).values(Array.from({ length: 101 }, () => ({ companyId, type: "request_board_approval", payload: {} }))).returning();
+    await db.insert(issueApprovals).values(rows.map(row => ({ companyId, issueId: target.id, approvalId: row.id })));
+    const last = rows.map(row => row.id).sort().at(-1)!;
+    await db.execute(sql`create function approval_queue_fail() returns trigger language plpgsql as $$ begin
+      if new.source_id <> TG_ARGV[0] then raise exception 'projection unavailable'; end if; return new; end $$`);
+    await db.execute(sql`create trigger approval_queue_fail before insert on decision_queue_items for each row execute function approval_queue_fail(${sql.raw("'" + last + "'")})`);
+    const service = approvalQueueSeedingService(db);
+    await expect(service.sweep()).rejects.toThrow("Approval queue seeding failed");
+    await service.sweep(); expect((await memberships()).map(row => row.sourceId)).toEqual([last]);
+    await db.execute(sql`drop trigger approval_queue_fail on decision_queue_items`);
+    await service.sweep(); expect(await memberships()).toHaveLength(101); expect(await service.next()).toBeNull();
+  });
+
+  it("rechecks the approval after a concurrent resolution commits", async () => {
+    const approval = await pending(); let release!: () => void, locked!: () => void;
+    const gate = new Promise<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);
+  });
+
+});

server/src/__tests__/decision-queues-routes.test.ts

@@ -1,3 +1,4 @@
+import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
 import { randomUUID } from "node:crypto";
 import express from "express";
 import request from "supertest";
@@ -26,7 +27,8 @@ import { errorHandler } from "../middleware/index.js";
 import { decisionQueueRoutes } from "../routes/decision-queues.js";
 import { decisionQueueService } from "../services/decision-queues.js";
 
-const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
+const external = resolveExternalTestDatabaseUrl("PAPERCLIP_DECISION_QUEUES_TEST_DATABASE_URL");
+const embeddedPostgresSupport = external ? { supported: true, reason: undefined } : await getEmbeddedPostgresTestSupport();
 const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
 
 if (!embeddedPostgresSupport.supported) {
@@ -40,8 +42,8 @@ describeEmbeddedPostgres("decision queue routes", () => {
   let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
 
   beforeAll(async () => {
-    tempDb = await startEmbeddedPostgresTestDatabase("paperclip-decision-queues-");
-    db = createDb(tempDb.connectionString);
+    if (!external) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-decision-queues-");
+    db = createDb(external ?? tempDb!.connectionString);
   }, 30_000);
 
   afterEach(async () => {
@@ -59,6 +61,7 @@ describeEmbeddedPostgres("decision queue routes", () => {
   });
 
   afterAll(async () => {
+    await db?.$client.end({ timeout: 1 });
     await tempDb?.cleanup();
   });
 

server/src/__tests__/server-startup-feedback-export.test.ts

@@ -1,3 +1,4 @@
+import { approvalQueueSeedingService } from "../services/approval-queue-seeding-work.js";
 import { joinRequestRetentionService } from "../services/join-request-retention.js";
 import { approvalRetentionService } from "../services/approval-retention.js";
 import { logger } from "../middleware/logger.js";
@@ -328,6 +329,11 @@ vi.mock("../services/join-request-retention.js", async importOriginal => ({
   joinRequestRetentionService: vi.fn(() => ({ sweep: vi.fn(async () => ({ archived: 0 })), next: vi.fn(async () => null) })),
 }));
 
+vi.mock("../services/approval-queue-seeding-work.js", async importOriginal => ({
+  ...await importOriginal<typeof import("../services/approval-queue-seeding-work.js")>(),
+  approvalQueueSeedingService: vi.fn(() => ({ sweep: vi.fn(async () => undefined), next: vi.fn(async () => null) })),
+}));
+
 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) })),
@@ -531,6 +537,18 @@ describe("startServer feedback export wiring", () => {
     process.env.BETTER_AUTH_SECRET = "test-secret";
   });
 
+  it.each([false, true])("registers approval queue seeding only when maintenance is enabled (%s)", async enabled => {
+    loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled }));
+    await startServer(); await settleStartupHousekeeping();
+    const registration = deliveryWorkRegisterMock.mock.calls.find(([queue]) => queue === DELIVERY_QUEUES.approvalQueueSeeding);
+    expect(Boolean(registration)).toBe(enabled);
+    if (enabled) {
+      const service = vi.mocked(approvalQueueSeedingService).mock.results.at(-1)!.value;
+      await registration![1].run(new AbortController().signal);
+      expect(service.sweep).toHaveBeenCalled(); expect(service.next).toHaveBeenCalled();
+    }
+  });
+
   it.each([false, true])("registers approval retention only when maintenance is enabled (%s)", async enabled => {
     loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled }));
     await startServer(); await settleStartupHousekeeping();

server/src/index.ts

@@ -1,3 +1,4 @@
+import { approvalQueueSeedingService, registerApprovalQueueSeedingWork } from "./services/approval-queue-seeding-work.js";
 import { joinRequestRetentionService, registerJoinRequestRetentionWork } from "./services/join-request-retention.js";
 import { approvalRetentionService, registerApprovalRetentionWork } from "./services/approval-retention.js";
 import { registerDecisionTargetWork } from "./services/decision-target-work.js";
@@ -1825,6 +1826,8 @@ async function startServerWithDatabaseTeardown(
       await step("environment_lease_cleanup", startEnvironmentCleanup);
       await step("archive_notification_worker", () => registerDecisionArchiveDeliveryWork(deliveryWork, retentionExecutor,
         () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready);
+      await step("approval_queue_seeding", () => registerApprovalQueueSeedingWork(deliveryWork, approvalQueueSeedingService(db as any),
+        () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready);
       await step("approval_retention", () => registerApprovalRetentionWork(deliveryWork, approvalRetentionService(db as any),
         () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready);
       await step("join_request_retention", () => registerJoinRequestRetentionWork(deliveryWork, joinRequestRetentionService(db as any),

server/src/lib/approval-records.ts

@@ -1,4 +1,4 @@
-import { writeApprovalChange } from "../services/retention-work-notifications.js";
+import { writeApprovalChange } from "../services/approval-work-notifications.js";
 import { and, eq, inArray } from "drizzle-orm";
 import { approvals, type Db } from "@paperclipai/db";
 import { notFound, unprocessable } from "../errors.js";

server/src/services/approval-queue-seeding-work.ts

@@ -0,0 +1,63 @@
+import { and, asc, eq, gt, sql, type SQL } from "drizzle-orm";
+import { approvals, companies, issues, type Db } from "@paperclipai/db";
+import { executionIssueCondition } from "./issue-visibility.js";
+import { decisionQueueService } from "./decision-queues.js";
+import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js";
+import { registerDeadlineWork } from "./deadline-work.js";
+import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
+
+const linkedIssue = sql<string>`(select linked.issue_id from issue_approvals linked
+  where linked.company_id = ${approvals.companyId} and linked.approval_id = ${approvals.id}
+  order by linked.issue_id limit 1)`;
+
+export function approvalQueueSeedingService(db: Db) {
+  let cursor: string | undefined;
+  function candidates(database: Db, condition?: SQL) {
+    return database.select({ id: approvals.id, companyId: approvals.companyId, issueId: linkedIssue }).from(approvals)
+      .innerJoin(companies, eq(companies.id, approvals.companyId))
+      .where(and(eq(approvals.status, "pending"), eq(companies.status, "active"), condition,
+        sql`exists (select 1 from ${issues} where ${issues.id} = ${linkedIssue}
+          and ${issues.companyId} = ${approvals.companyId} and ${executionIssueCondition()}
+          and exists (select 1 from issue_work_products product where product.company_id = ${approvals.companyId}
+            and product.issue_id = ${issues.id} and product.type = 'pull_request'))`,
+        sql`not exists (select 1 from decision_queues queue where queue.company_id = ${approvals.companyId} and queue.key = 'prs'
+          and (not queue.seed_rules_enabled or exists (select 1 from decision_queue_items item where item.queue_id = queue.id
+            and item.company_id = ${approvals.companyId} and item.source_kind = 'approval' and item.source_id = ${approvals.id}::text)))`,
+      )).orderBy(asc(approvals.id));
+  }
+  async function sweep(signal?: AbortSignal) {
+    const rows = await candidates(db, cursor ? gt(approvals.id, cursor) : undefined).limit(100);
+    const errors: unknown[] = [];
+    for (const row of rows) {
+      if (signal?.aborted) break;
+      try {
+        await db.transaction(async tx => {
+          const database = tx as unknown as Db;
+          const [current] = await candidates(database, eq(approvals.id, row.id)).for("update", { of: approvals });
+          if (!current) return;
+          await decisionQueueService(database).materializeSeedSources(current.companyId, [{
+            sourceKind: "approval", sourceId: current.id, issueId: current.issueId,
+          }]);
+        });
+      } catch (error) { errors.push(error); }
+      cursor = row.id;
+    }
+    if (rows.length < 100) cursor = undefined;
+    if (errors.length === 1) throw errors[0];
+    if (errors.length) throw new AggregateError(errors, "Approval queue seeding failed");
+  }
+  async function next() {
+    return (await candidates(db).limit(1)).length ? Date.now() : null;
+  }
+  return { sweep, next };
+}
+
+export function registerApprovalQueueSeedingWork(
+  coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
+  service: ReturnType<typeof approvalQueueSeedingService>,
+  canRun: () => boolean,
+) {
+  return registerDeadlineWork(coordinator, DELIVERY_QUEUES.approvalQueueSeeding, {
+    run: signal => service.sweep(signal), next: service.next, canRun, retryMs: 1_000,
+  });
+}

server/src/services/approval-work-notifications.ts

@@ -0,0 +1,16 @@
+import type { Db } from "@paperclipai/db";
+import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
+
+export const notifyApprovalQueueWork = (transaction: object) => notifyDeliveryWork(transaction, DELIVERY_QUEUES.approvalQueueSeeding);
+
+export async function notifyApprovalWork(transaction: object) {
+  await notifyDeliveryWork(transaction, DELIVERY_QUEUES.approvalRetention);
+  await notifyApprovalQueueWork(transaction);
+}
+
+export function writeApprovalChange<T>(db: Db, write: (transaction: Db) => Promise<T>): Promise<T> {
+  return db.transaction(async tx => {
+    await notifyApprovalWork(tx);
+    return write(tx as unknown as Db);
+  });
+}

server/src/services/approvals.ts

@@ -1,4 +1,4 @@
-import { writeApprovalChange } from "./retention-work-notifications.js";
+import { writeApprovalChange } from "./approval-work-notifications.js";
 import { approvalResolutionRecords, resolvableApprovalStatuses as resolvableStatuses } from "../lib/approval-records.js";
 import { notifyHireApproved } from "./hire-hook.js";
 

server/src/services/budgets.ts

@@ -1,4 +1,5 @@
-import { notifyCompanyRetentionWork, notifyRetentionWork } from "./retention-work-notifications.js";
+import { notifyApprovalWork } from "./approval-work-notifications.js";
+import { notifyCompanyRetentionWork } from "./retention-work-notifications.js";
 import { addCents, compareCents, normalizeCents } from "@paperclipai/shared";
 import { and, desc, eq, inArray, sql, or, isNull } from "drizzle-orm";
 import type { Db } from "@paperclipai/db";
@@ -259,7 +260,7 @@ async function markApprovalStatus(
   decidedByUserId: string,
 ) {
   if (!approvalId) return;
-  await notifyRetentionWork(db, "approval");
+  await notifyApprovalWork(db);
   await db
     .update(approvals)
     .set({
@@ -457,7 +458,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica
       windowEnd: end,
     });
 
-    if (thresholdType === "hard") await notifyRetentionWork(db, "approval");
+    if (thresholdType === "hard") await notifyApprovalWork(db);
     const approval = thresholdType === "hard"
       ? await db
         .insert(approvals)

server/src/services/decision-queues.ts

@@ -1,3 +1,4 @@
+import { notifyApprovalQueueWork } from "./approval-work-notifications.js";
 import { and, desc, eq, inArray, isNull, sql } from "drizzle-orm";
 import type { Db } from "@paperclipai/db";
 import {
@@ -43,6 +44,14 @@ export type DecisionMutationActor = {
   responsibleUserId: string | null;
 };
 
+export type DecisionQueueSeedSource = {
+  sourceKind: AttentionSourceKind;
+  sourceId: string;
+  issueId: string | null;
+  isPlanTarget?: boolean;
+  interactionKind?: string;
+};
+
 type SeedDefinition = {
   key: string;
   title: string;
@@ -380,6 +389,104 @@ export function decisionQueueService(db: Db) {
     return visible;
   }
 
+  async function materializeSeedSources(companyId: string, items: DecisionQueueSeedSource[]) {
+    if (items.length === 0) return;
+    const issueIds = [...new Set(items.map(item => item.issueId).filter((id): id is string => Boolean(id)))];
+    const prIssueIds = new Set(issueIds.length === 0 ? [] : await db
+      .select({ issueId: issueWorkProducts.issueId })
+      .from(issueWorkProducts)
+      .where(and(
+        eq(issueWorkProducts.companyId, companyId),
+        eq(issueWorkProducts.type, "pull_request"),
+        inArray(issueWorkProducts.issueId, issueIds),
+      )).then((rows) => rows.map((row) => row.issueId)));
+
+    const matches = new Map<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 +499,7 @@ export function decisionQueueService(db: Db) {
     }) => {
       const result = await db.transaction(async (tx) => {
         const txDb = tx as unknown as Db;
+        if (input.key === "prs") await notifyApprovalQueueWork(txDb);
         const inserted = await txDb.insert(decisionQueues).values({
           companyId: input.companyId,
           key: input.key,
@@ -449,6 +557,7 @@ export function decisionQueueService(db: Db) {
     }) => {
       const row = await db.transaction(async (tx) => {
         const txDb = tx as unknown as Db;
+        if (input.key === "prs") await notifyApprovalQueueWork(txDb);
         const updated = await txDb.update(decisionQueues).set({ ...input.patch, updatedAt: new Date() })
           .where(and(eq(decisionQueues.companyId, input.companyId), eq(decisionQueues.key, input.key)))
           .returning().then((rows) => rows[0] ?? null);
@@ -556,6 +665,7 @@ export function decisionQueueService(db: Db) {
       }
       return db.transaction(async (tx) => {
         const txDb = tx as unknown as Db;
+        if (input.key === "prs" && input.sourceKind === "approval") await notifyApprovalQueueWork(txDb);
         const removed = await txDb.delete(decisionQueueItems).where(and(
           eq(decisionQueueItems.companyId, input.companyId),
           eq(decisionQueueItems.queueId, queue.id),
@@ -682,101 +792,13 @@ export function decisionQueueService(db: Db) {
       });
     },
 
-    materializeSeededQueues: async (companyId: string, items: AttentionItem[]) => {
-      if (items.length === 0) return;
-      const issueIds = [...new Set(items.map(itemIssueId).filter((id): id is string => Boolean(id)))];
-      const prIssueIds = new Set(issueIds.length === 0 ? [] : await db
-        .select({ issueId: issueWorkProducts.issueId })
-        .from(issueWorkProducts)
-        .where(and(
-          eq(issueWorkProducts.companyId, companyId),
-          eq(issueWorkProducts.type, "pull_request"),
-          inArray(issueWorkProducts.issueId, issueIds),
-        )).then((rows) => rows.map((row) => row.issueId)));
-
-      const matches = new Map<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

@@ -8,6 +8,7 @@ export const DELIVERY_QUEUES = {
   executionStatus: "execution-status",
   executionDisposition: "execution-disposition",
   environmentCleanup: "environment-cleanup",
+  approvalQueueSeeding: "approval-queue-seeding",
   approvalRetention: "approval-retention",
   joinRequestRetention: "join-request-retention",
   decisionTargets: "decision-targets",

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

server/src/services/issues.ts

@@ -1,4 +1,4 @@
-import { notifyRetentionWork } from "./retention-work-notifications.js";
+import { notifyApprovalWork } from "./approval-work-notifications.js";
 import { notifyDecisionTargetWork } from "./decision-target-work.js";
 import { insertIssueComment } from "./issue-comment-writes.js";
 import { measureRunTransaction } from "./execution-transaction-timing.js";
@@ -11302,7 +11302,7 @@ export function issueService(db: Db) {
           await notifyDecisionTargetWork(tx, `issue:${id}`);
         }
         if (patch.hiddenAt !== undefined || patch.harnessKind !== undefined || patch.conversationAgentId !== undefined) {
-          await notifyRetentionWork(tx, "approval");
+          await notifyApprovalWork(tx);
         }
         const updated = await tx
           .update(issues)
@@ -11711,7 +11711,7 @@ export function issueService(db: Db) {
         try {
           await notifyTerminalWorkspaceWork(tx, `issue:${id}`);
           await requestRestorePointDeletion(tx, { sourceIssueId: id });
-          await notifyRetentionWork(tx, "approval");
+          await notifyApprovalWork(tx);
           await notifyDecisionTargetWork(tx);
           removedIssue = await tx
             .delete(issues)

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");
+  await notifyApprovalWork(transaction);
   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 { notifyDecisionTargetWork } from "./decision-target-work.js";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
 import { verifyAppWebhook } from "./app-webhook.js";
@@ -1966,7 +1966,7 @@ export function routineService(
         return updated ?? createdRun;
       } catch (error) {
         if (createdIssue) {
-          await notifyRetentionWork(txDb, "approval");
+          await notifyApprovalWork(txDb);
           await notifyDecisionTargetWork(txDb);
           await txDb.delete(issues).where(eq(issues.id, createdIssue.id));
         }

server/src/services/tool-gateway.ts

@@ -1,4 +1,4 @@
-import { writeApprovalChange } from "./retention-work-notifications.js";
+import { writeApprovalChange } from "./approval-work-notifications.js";
 import { toolReviewProjectionService } from "./tool-review-projection.js";
 import { toolReviewInterruptionService } from "./tool-review-interruption.js";
 import { claimToolTestApproval, persistToolTestOutcome, startToolTestInvocation, type ToolTestApprovalClaim } from "./tool-test-approval.js";

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);
+        }
         await validateExecutionWorkspace(tx, companyId, data.executionWorkspaceId);
         if (data.isPrimary) {
           await tx
@@ -309,7 +313,10 @@ export function workProductService(
           .then((rows) => rows[0] ?? null);
         if (!existing) return null;
 
-        if (existing.type === "pull_request" || patch.type === "pull_request") await notifyTerminalWorkspaceWork(tx, `issue:${existing.issueId}`);
+        if (existing.type === "pull_request" || patch.type === "pull_request") {
+          await notifyTerminalWorkspaceWork(tx, `issue:${existing.issueId}`);
+          await notifyApprovalQueueWork(tx);
+        }
 
         await validateExecutionWorkspace(tx, existing.companyId, patch.executionWorkspaceId);
 
@@ -378,6 +385,7 @@ export function workProductService(
       // prefix behind (which a retry would then duplicate). Mirrors the
       // per-writer transaction the batched issue/document writers use.
       await db.transaction(async (tx) => {
+        if (values.some(value => value.type === "pull_request")) await notifyApprovalQueueWork(tx);
         for (const value of values) {
           if (value.type === "pull_request") await notifyTerminalWorkspaceWork(tx, `issue:${value.issueId}`);
         }
@@ -389,7 +397,10 @@ export function workProductService(
       const row = await db.transaction(async tx => {
         const [existing] = await tx.select({ type: issueWorkProducts.type, issueId: issueWorkProducts.issueId })
           .from(issueWorkProducts).where(eq(issueWorkProducts.id, id));
-        if (existing?.type === "pull_request") await notifyTerminalWorkspaceWork(tx, `issue:${existing.issueId}`);
+        if (existing?.type === "pull_request") {
+          await notifyTerminalWorkspaceWork(tx, `issue:${existing.issueId}`);
+          await notifyApprovalQueueWork(tx);
+        }
         return tx.delete(issueWorkProducts).where(eq(issueWorkProducts.id, id))
           .returning().then(rows => rows[0] ?? null);
       });