fix(server): schedule join request retention from durable deadlines

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

doc/idle-join-request-retention.md

@@ -0,0 +1,99 @@
+# Join request retention while idle
+
+Pending join requests already have durable activity timestamps. Retention now
+uses their next 90-day archive deadline instead of including them in the main
+maintenance tick's attention-feed scan. The regular attention UI still includes
+join requests. This changes automatic retention discovery, not access approval.
+
+## Owner and eligibility
+
+`join-request-retention.ts` owns the candidate query, bounded scan, and deadline.
+A candidate is a pending_approval request belonging to an active company, with
+an invite whose company matches the request. Kept and already archived items do
+not schedule work. Invite expiration or revocation does not remove the existing
+attention item and therefore does not disable its retention. No provider policy
+is interpreted by the worker or shared scheduler.
+
+The next deadline is the earliest eligible updatedAt plus 90 days. The partial
+pending index orders updated_at and id. Resolved request history is excluded by
+the index predicate. Each pass attempts at most 100 due rows. The private cursor
+preserves database timestamp precision so a failed page cannot starve later work.
+Errors keep the shared 30-second retry; successful full pages continue until
+work is exhausted. No eligible rows means no scheduled SQL or network scan.
+
+Each candidate is re-read under its request row lock before retention changes.
+This preserves a concurrent request update or status transition. The shared
+retention owner still controls keep, archive/revive, version checks, activity
+logging, and notification creation. Archiving does not approve or reject a join.
+Join requests have no origin agent, so their archival creates no agent delivery.
+Company and invite joins preserve existing visibility policy; this does not add
+cross-table serializability beyond the existing retention contract.
+
+## Producer audit
+
+| Producer | Notification boundary |
+| --- | --- |
+| Invite acceptance creates a pending request | Existing acceptance transaction, before the request insert |
+| Supported gateway invite replay changes activity | Small writeJoinRequestChange transaction around the existing SQL update |
+| Automatic human approval after acceptance | Same helper around the status update |
+| Explicit approval and rejection | Same helper around the status update |
+| Slack self-service access request | Existing platform transaction, before the request insert |
+| Company pause, resume, archive, or deletion | Company-level retention notification inside the existing transaction |
+| Budget-driven company pause or resume | Same company notification in the accounting transaction |
+| Keep/unkeep, manual archive/revive, accepted archive proposal | Existing typed retention notification now routes join_request to its worker |
+
+Provider calls, membership grants, and unrelated route behavior stay outside the
+small SQL mutation helper. Nested callers retain the existing outer transaction
+settlement and uncertain-outcome protocol. The access router itself requires a
+root database connection; it is not wrapped in a caller transaction.
+
+API-key consumption only permits an already-approved request. It cannot create
+pending retention work and does not acquire a new wrapper or signal. Invite
+acceptance that reuses a human request does not update that request. Revocation,
+expiry, and acceptedAt changes do not change the feed predicate. Invite/company
+identity is immutable in supported routes; company deletion is the invite delete
+owner. A future producer that changes these facts must join the same contract.
+Process-local signals do not cover independent SQL writers or another instance.
+
+## Lifecycle and limits
+
+Startup reconstructs pending deadlines. The worker is registered only when the
+existing heartbeat maintenance switch is enabled. Writes during drain retain a
+wake without passive SQL. Failed work retains its retry across drain. Shutdown
+joins an admitted archive and stops before subsequent candidates; restart picks
+up the durable remainder. There is no new queue table or permanent DB listener.
+The in-process deadline is not a Cloud alarm. Existing unsupported-future-work
+sleep blockers remain in place.
+
+## Verification
+
+Seventeen new real-PostgreSQL cases pass. They cover actual invite acceptance,
+rejection, human approval, gateway replay, and Slack access requests; empty and
+resolved-history quiet; saved deadlines; startup reconstruction; keep and manual
+archival controls; company pause/resume; unchanged expired/revoked invite policy;
+normal versus maintenance feeds; nested commit/rollback; first writes and failed
+work during drain; concurrent source updates; 101-row microsecond paging with a
+full failed page; and shutdown/restart.
+
+Another 139 consumer cases pass: 71 invitation/startup cases, 21 approval worker
+cases, and 47 shared retention/attention/budget cases. This is 156 selected tests,
+not a full local suite claim. The first consumer command used a nonexistent
+retention test filename and ran only the approval suite; the corrected selection
+ran the actual retention suite against its dedicated external database.
+
+Initial test failures exposed fixture assumptions: the access router cannot run
+on a transaction handle; unresolved recovery cannot be expected to become idle;
+Slack links require actual endpoint/principal ownership rows; and shutdown tests
+must establish admission and deterministic candidate order. Those fixtures were
+corrected without weakening assertions or production checks. The first replay
+fixture lacked required gateway credentials. The final fixture supplies test-only
+credentials. A reused transaction spy retained admission calls during the quiet
+assertion; restoring it before the quiet measurement fixed the measurement.
+
+Migration 0387 applies, the full migration set replays, and the concurrent index
+statement replays idempotently. Migration safety and numbering checks pass; the
+five newest schema snapshots are retained. The actual next-deadline query with
+40,000 resolved requests and one pending request uses the pending index and
+executes in 0.017 ms locally. The probe rolls back its data. This is query-plan
+evidence, not production latency or measured savings. Workspace typechecking and the production build pass. Fresh PR CI/review remain
+required before readiness.

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

@@ -866,3 +866,25 @@ preserves cursor precision in the owning query. All 87 workspace service/worker
 cases and server typechecking pass. See
 [workspace pagination](../idle-terminal-workspace-pagination.md). Publication and
 fresh CI/review are next. Join retention remains the next new migration duty.
+
+## Join request retention implementation
+
+Terminal workspace pagination is [#17505](https://github.com/paperclipai/paperclip/pull/17505),
+head `28a197cdfd`, stacked on #17497. It has fresh 5/5 review without unresolved
+findings. The full CI matrix is running. Workspace typecheck/build and all 87
+selected service/worker tests passed locally.
+
+Join retention is implemented on `codex/sleep-15-join-retention`, based on #17505.
+The worker preserves pending request/invite/company eligibility, manual controls,
+transaction ownership, microsecond paging, and bounded recovery. All supported
+writers are audited; approved-only API-key consumption is deliberately exempt.
+See [join retention](../idle-join-request-retention.md). Seventeen new real-DB
+lifecycle cases and 139 consumers pass (156 total). Initial fixture and command
+failures are recorded in that document. Index generation, apply, replay, safety,
+numbering, and an actual-query plan with 40,000 resolved requests pass. Workspace
+typecheck/build and remote publication/CI/review remain in progress. No slice is
+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.

packages/db/src/client.ts

@@ -512,6 +512,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([
   "0384_many_reaper.sql",
   "0385_normal_blindfold.sql",
   "0386_omniscient_luke_cage.sql",
+  "0387_huge_skrulls.sql",
   "0380_funny_adam_warlock.sql",
 ]);
 

packages/db/src/migrations/0387_huge_skrulls.sql

@@ -0,0 +1 @@
+CREATE INDEX CONCURRENTLY IF NOT EXISTS "join_requests_pending_retention_idx" ON "join_requests" USING btree ("updated_at","id") WHERE "join_requests"."status" = 'pending_approval';

packages/db/src/migrations/meta/0387_snapshot.json

@@ -1,6 +1,6 @@
 {
-  "id": "5e8d042e-d697-4753-9582-7d0c042ef893",
-  "prevId": "afdc5e70-19b3-416d-917a-beb9fdc1e2e1",
+  "id": "4c86af49-02a0-496f-b028-b468d9e4e960",
+  "prevId": "02c5ad8a-cdb9-41ad-acf1-5ab4f7d5bd67",
   "version": "7",
   "dialect": "postgresql",
   "tables": {
@@ -5721,6 +5721,28 @@
         }
       },
       "indexes": {
+        "approvals_pending_retention_idx": {
+          "name": "approvals_pending_retention_idx",
+          "columns": [
+            {
+              "expression": "updated_at",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            },
+            {
+              "expression": "id",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            }
+          ],
+          "isUnique": false,
+          "where": "\"approvals\".\"status\" = 'pending'",
+          "concurrently": false,
+          "method": "btree",
+          "with": {}
+        },
         "approvals_company_status_type_idx": {
           "name": "approvals_company_status_type_idx",
           "columns": [
@@ -24030,6 +24052,50 @@
           "method": "btree",
           "with": {}
         },
+        "decisions_running_recovery_idx": {
+          "name": "decisions_running_recovery_idx",
+          "columns": [
+            {
+              "expression": "updated_at",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            },
+            {
+              "expression": "id",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            }
+          ],
+          "isUnique": false,
+          "where": "\"decisions\".\"status\" = 'decided' AND \"decisions\".\"execution_status\" = 'running'",
+          "concurrently": false,
+          "method": "btree",
+          "with": {}
+        },
+        "decisions_open_expiry_idx": {
+          "name": "decisions_open_expiry_idx",
+          "columns": [
+            {
+              "expression": "expires_at",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            },
+            {
+              "expression": "id",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            }
+          ],
+          "isUnique": false,
+          "where": "\"decisions\".\"status\" = 'open'",
+          "concurrently": false,
+          "method": "btree",
+          "with": {}
+        },
         "decisions_pending_continuation_idx": {
           "name": "decisions_pending_continuation_idx",
           "columns": [
@@ -32035,6 +32101,28 @@
           "method": "btree",
           "with": {}
         },
+        "grok_dispatch_prepared_idx": {
+          "name": "grok_dispatch_prepared_idx",
+          "columns": [
+            {
+              "expression": "created_at",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            },
+            {
+              "expression": "id",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            }
+          ],
+          "isUnique": false,
+          "where": "\"grok_bot_dispatches\".\"status\" = 'prepared'",
+          "concurrently": false,
+          "method": "btree",
+          "with": {}
+        },
         "grok_dispatch_company_status_idx": {
           "name": "grok_dispatch_company_status_idx",
           "columns": [
@@ -41829,6 +41917,28 @@
         }
       },
       "indexes": {
+        "join_requests_pending_retention_idx": {
+          "name": "join_requests_pending_retention_idx",
+          "columns": [
+            {
+              "expression": "updated_at",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            },
+            {
+              "expression": "id",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            }
+          ],
+          "isUnique": false,
+          "where": "\"join_requests\".\"status\" = 'pending_approval'",
+          "concurrently": false,
+          "method": "btree",
+          "with": {}
+        },
         "join_requests_invite_unique_idx": {
           "name": "join_requests_invite_unique_idx",
           "columns": [

packages/db/src/migrations/meta/_journal.json

@@ -2696,6 +2696,13 @@
       "when": 1792877330859,
       "tag": "0386_omniscient_luke_cage",
       "breakpoints": true
+    },
+    {
+      "idx": 387,
+      "version": "7",
+      "when": 1792880971454,
+      "tag": "0387_huge_skrulls",
+      "breakpoints": true
     }
   ]
 }

packages/db/src/schema/join_requests.ts

@@ -31,6 +31,8 @@ export const joinRequests = pgTable(
     updatedAt: timestamp("updated_at", { withTimezone: true }).notNull().defaultNow(),
   },
   (table) => ({
+    pendingRetentionIdx: index("join_requests_pending_retention_idx").on(table.updatedAt, table.id)
+      .where(sql`${table.status} = 'pending_approval'`),
     inviteUniqueIdx: uniqueIndex("join_requests_invite_unique_idx").on(table.inviteId),
     companyStatusTypeCreatedIdx: index("join_requests_company_status_type_created_idx").on(
       table.companyId,

server/src/__tests__/join-request-retention-work.test.ts

@@ -0,0 +1,308 @@
+import { createSlackIdentityLinks } from "../services/chat-channels/slack/identity-links.js";
+import { createHash, randomUUID } from "node:crypto";
+import express from "express";
+import request from "supertest";
+import { eq, sql } from "drizzle-orm";
+import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
+import { activityLog, agents, authUsers, chatEndpoints, chatExternalPrincipals, chatIdentityLinks, companies, companyMemberships, companySecretProviderConfigs, companySecrets, companySecretVersions, createDb, decisionArchiveNotificationOutbox,
+  decisionRetention, invites, joinRequests, principalPermissionGrants, toolApplications, toolConnections, type Db } from "@paperclipai/db";
+import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
+import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
+import { joinRequestRetentionService, registerJoinRequestRetentionWork } from "../services/join-request-retention.js";
+import { decisionRetentionService, DEFAULT_DECISION_ARCHIVE_DAYS } from "../services/decision-retention.js";
+import { attentionService } from "../services/attention.js";
+import { companyService } from "../services/companies.js";
+import { writeJoinRequestChange } from "../services/retention-work-notifications.js";
+import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
+import { idleWorkSnapshot } from "../services/task-admission.js";
+
+const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_JOIN_RETENTION_TEST_DATABASE_URL");
+const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTestSupport();
+const delay = DEFAULT_DECISION_ARCHIVE_DAYS * 86_400_000;
+(support.supported ? describe : describe.skip)("join request retention worker", () => {
+  let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
+  let db: ReturnType<typeof createDb>, companyId: string, userId: string;
+  let coordinator: ReturnType<typeof createDeliveryWorkCoordinator> | undefined;
+  let draining: boolean;
+  let accessRoutes: typeof import("../routes/access.js").accessRoutes;
+  const errors: unknown[] = [];
+  beforeAll(async () => {
+    ({ accessRoutes } = await import("../routes/access.js"));
+    if (!externalUrl) temporary = await startEmbeddedPostgresTestDatabase("join-retention-");
+    db = createDb(externalUrl ?? temporary!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 });
+  }, 30_000);
+  beforeEach(async () => {
+    vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "0123456789abcdef0123456789abcdef");
+    companyId = randomUUID(); userId = randomUUID(); draining = false; errors.length = 0;
+    await db.insert(companies).values({ id: companyId, name: "Join retention", issuePrefix: `J${companyId.slice(0, 6)}` });
+    await db.insert(authUsers).values({ id: userId, name: "Owner", email: `${userId}@example.test`, createdAt: new Date(), updatedAt: new Date() });
+    await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: userId, membershipRole: "owner", status: "active" });
+    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(); vi.unstubAllEnvs();
+    await db.execute(sql`drop trigger if exists join_retention_fail on decision_retention`);
+    await db.execute(sql`drop function if exists join_retention_fail()`);
+    await db.delete(decisionArchiveNotificationOutbox); await db.delete(decisionRetention); await db.delete(activityLog);
+    await db.delete(chatIdentityLinks); await db.delete(chatExternalPrincipals); await db.delete(chatEndpoints);
+    await db.delete(toolConnections); await db.delete(toolApplications); await db.delete(agents);
+    await db.delete(joinRequests); await db.delete(invites); await db.delete(principalPermissionGrants);
+    await db.delete(companySecretVersions); await db.delete(companySecrets); await db.delete(companySecretProviderConfigs);
+    await db.delete(companyMemberships); await db.delete(authUsers); await db.delete(companies);
+    expect(errors).toEqual([]);
+  });
+  afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); });
+  const old = () => new Date(Date.now() - delay - 60_000);
+  const state = (id: string) => decisionRetentionService(db).getState(companyId, "join_request", id);
+  const authActor = () => ({ type: "board" as const, source: "local_implicit" as const, userId, companyIds: [companyId], isInstanceAdmin: true,
+    memberships: [{ companyId, membershipRole: "owner", status: "active" }] });
+  const mutation = (id: string) => ({ companyId, sourceKind: "join_request" as const, sourceId: id, authActor: authActor(),
+    actor: { actorType: "user" as const, actorId: userId, agentId: null, userId, runId: null, agentApiKeyId: null, responsibleUserId: userId } });
+  async function invite(database: Db = db) {
+    const token = randomUUID();
+    const [row] = await database.insert(invites).values({ companyId, tokenHash: createHash("sha256").update(token).digest("hex"),
+      allowedJoinTypes: "both", expiresAt: new Date(Date.now() + 86_400_000), invitedByUserId: userId }).returning();
+    return { ...row, token };
+  }
+  async function create(at = new Date(), database: Db = db) {
+    const invitation = await invite(database);
+    const [row] = await writeJoinRequestChange(database, tx => tx.insert(joinRequests).values({ companyId, inviteId: invitation.id,
+      requestType: "agent", requestIp: "127.0.0.1", agentName: "Pending agent", createdAt: at, updatedAt: at }).returning());
+    return row;
+  }
+  function app(database: Db = db) {
+    const server = express(); server.use(express.json());
+    server.use((req, _res, next) => { req.actor = authActor(); next(); });
+    server.use("/api", accessRoutes(database, { deploymentMode: "authenticated", deploymentExposure: "private",
+      bindHost: "127.0.0.1", allowedHostnames: [] }));
+    server.use((error: { status?: number; message?: string }, _req: express.Request, res: express.Response, _next: express.NextFunction) => {
+      res.status(error.status ?? 500).json({ error: error.message });
+    });
+    return server;
+  }
+  async function accept(token: string, database: Db = db) {
+    const response = await request(app(database)).post(`/api/invites/${token}/accept`)
+      .send({ requestType: "agent", agentName: "New agent", adapterType: "claude_local" });
+    expect(response.status, JSON.stringify(response.body)).toBe(202);
+    const [row] = await database.select().from(joinRequests).where(eq(joinRequests.companyId, companyId));
+    return row;
+  }
+  async function start(service = joinRequestRetentionService(db)) {
+    coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining, canReconcile: () => true, onError: error => errors.push(error) });
+    await registerJoinRequestRetentionWork(coordinator, service, () => !draining).ready;
+    return service;
+  }
+  async function settle() {
+    await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(30_000); expect(idleWorkSnapshot().active).toBe(0); }, { timeout: 5000 });
+  }
+  async function quiet() {
+    await settle(); expect(coordinator!.nextWakeAt()).toBeNull();
+    const reads = vi.spyOn(db, "select"), executes = vi.spyOn(db, "execute"), writes = vi.spyOn(db, "transaction");
+    await vi.advanceTimersByTimeAsync(300_000);
+    expect(reads).not.toHaveBeenCalled(); expect(executes).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled();
+    reads.mockRestore(); executes.mockRestore(); writes.mockRestore();
+  }
+  async function archived(id: string) {
+    await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(30_000); expect((await state(id))?.archivedAt).toBeInstanceOf(Date); }, { timeout: 5000 });
+  }
+
+  it("wakes from real invite acceptance, waits for the saved deadline, and disarms on rejection", async () => {
+    await start(); await quiet();
+    const invitation = await invite(); const row = await accept(invitation.token);
+    await settle(); expect(coordinator!.nextWakeAt()).toBe(row.updatedAt.getTime() + delay);
+    const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(300_000);
+    expect(reads).not.toHaveBeenCalled(); reads.mockRestore();
+    const response = await request(app()).post(`/api/companies/${companyId}/join-requests/${row.id}/reject`).send({});
+    expect(response.status, JSON.stringify(response.body)).toBe(200); await quiet();
+    expect(await state(row.id)).toBeNull();
+  });
+
+  it("archives unseen pending requests at the actual deadline without changing approval status", async () => {
+    await start(); await quiet(); const deadline = Date.now() + 600_000;
+    const row = await create(new Date(deadline - delay)); await settle();
+    expect(coordinator!.nextWakeAt()).toBe(deadline); expect(await state(row.id)).toBeNull();
+    const window = deadline - Date.now() - 30_000; expect(window).toBeGreaterThan(0);
+    const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(window);
+    expect(reads).not.toHaveBeenCalled(); reads.mockRestore();
+    await vi.advanceTimersByTimeAsync(60_000); await archived(row.id); await quiet();
+    expect((await db.select().from(joinRequests))[0].status).toBe("pending_approval");
+    expect(await db.select().from(decisionArchiveNotificationOutbox)).toHaveLength(0);
+  });
+
+  it("reconstructs pending requests at startup and ignores resolved history", async () => {
+    const due = await create(old()), resolved = await create(old());
+    await db.update(joinRequests).set({ status: "approved" }).where(eq(joinRequests.id, resolved.id));
+    await start(); await archived(due.id); await quiet(); expect(await state(resolved.id)).toBeNull();
+  });
+
+  it("keeps revoked and expired invite requests visible under the existing retention policy", async () => {
+    const row = await create(old()); await db.update(invites).set({ expiresAt: old(), revokedAt: old() });
+    const feed = await attentionService(db).list(companyId, { all: true, allowUnscopedAll: true });
+    expect(feed.items.some(item => item.subject.id === row.id)).toBe(true);
+    await start(); await archived(row.id); await quiet();
+  });
+
+  it("waits for company activation and disarms when the company pauses", async () => {
+    const row = await create(old()); await companyService(db).update(companyId, { status: "paused" });
+    await start(); await quiet(); expect(await state(row.id)).toBeNull();
+    await companyService(db).update(companyId, { status: "active" }); await archived(row.id); await quiet();
+    await create(); await settle(); expect(coordinator!.nextWakeAt()).not.toBeNull();
+    await companyService(db).update(companyId, { status: "paused" }); await quiet();
+  });
+
+  it("honors keep and schedules an old request again when unkept", async () => {
+    const row = await create(old()); await attentionService(db).list(companyId, { all: true, allowUnscopedAll: true });
+    await decisionRetentionService(db).setKeep({ ...mutation(row.id), keep: true });
+    await start(); await quiet();
+    await decisionRetentionService(db).setKeep({ ...mutation(row.id), keep: false }); await archived(row.id); await quiet();
+  });
+
+  it("honors manual archival and reactivates on revive", async () => {
+    const row = await create(old()); await attentionService(db).list(companyId, { all: true, allowUnscopedAll: true });
+    await decisionRetentionService(db).archive(mutation(row.id)); await start(); await quiet();
+    await decisionRetentionService(db).revive(mutation(row.id)); await archived(row.id); await quiet();
+  });
+
+  it("preserves normal attention reads but omits join requests from maintenance", async () => {
+    const row = await create(old());
+    const excluded = await attentionService(db).list(companyId, { all: true, allowUnscopedAll: true, excludeSources: ["join_request"] });
+    expect(excluded.items.some(item => item.sourceKind === "join_request")).toBe(false); expect(await state(row.id)).toBeNull();
+    const feed = await attentionService(db).list(companyId, { all: true, allowUnscopedAll: true });
+    expect(feed.items.some(item => item.subject.id === row.id)).toBe(true);
+  });
+
+  it("defers nested request writes until outer commit and releases rolled-back writes", async () => {
+    await start(); await quiet();
+    await expect(db.transaction(async tx => {
+      await create(new Date(), tx as unknown as Db);
+      await vi.advanceTimersByTimeAsync(120_000); expect(coordinator!.nextWakeAt()).toBeNull(); throw new Error("rollback");
+    })).rejects.toThrow("rollback");
+    await quiet(); expect(await db.select().from(joinRequests)).toHaveLength(0);
+    let row: typeof joinRequests.$inferSelect;
+    await db.transaction(async tx => {
+      row = await create(new Date(), tx as unknown as Db);
+      await vi.advanceTimersByTimeAsync(120_000); expect(coordinator!.nextWakeAt()).toBeNull();
+    });
+    await settle(); expect(coordinator!.nextWakeAt()).toBe(row!.updatedAt.getTime() + delay);
+  });
+
+  it("retains a first write during drain without passive SQL", async () => {
+    await start(); await quiet(); draining = true; const row = await create(old());
+    const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(120_000);
+    expect(reads).not.toHaveBeenCalled(); reads.mockRestore();
+    draining = false; await archived(row.id); await quiet();
+  });
+
+  it("retains a failed archival through drain until the database recovers", async () => {
+    const row = await create(old());
+    await db.execute(sql`create function join_retention_fail() returns trigger language plpgsql as $$ begin
+      raise exception 'retention unavailable'; end $$`);
+    await db.execute(sql`create trigger join_retention_fail before insert or update on decision_retention for each row execute function join_retention_fail()`);
+    const service = joinRequestRetentionService(db); const sweep = vi.spyOn(service, "sweep");
+    await start(service); await vi.waitFor(() => expect(errors.length).toBeGreaterThan(0)); draining = true;
+    await Promise.allSettled(sweep.mock.results.map(result => result.value));
+    await vi.advanceTimersByTimeAsync(30_000);
+    const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(120_000);
+    expect(reads).not.toHaveBeenCalled(); reads.mockRestore();
+    await db.execute(sql`drop trigger join_retention_fail on decision_retention`); draining = false;
+    await archived(row.id); await quiet();
+    for (const error of errors) expect(error).toMatchObject({ cause: { message: "retention unavailable" } }); errors.length = 0;
+  });
+
+  it("wakes from the Slack access-request producer and deduplicates repeated requests", async () => {
+    await start(); await quiet(); const token = randomUUID(), endpointId = randomUUID();
+    const [agent] = await db.insert(agents).values({ companyId, name: "Chat agent", role: "engineer", status: "active", adapterType: "codex_local" }).returning();
+    const [application] = await db.insert(toolApplications).values({ companyId, name: "Slack", type: "chat" }).returning();
+    const [connection] = await db.insert(toolConnections).values({ companyId, applicationId: application.id, name: "Slack", uid: randomUUID(), transport: "chat_sdk", connectionPurpose: "channel" }).returning();
+    await db.insert(chatEndpoints).values({ id: endpointId, companyId, connectionId: connection.id, provider: "slack", publicId: randomUUID(), assignedAgentId: agent.id, status: "active" });
+    const [principal] = await db.insert(chatExternalPrincipals).values({ companyId, provider: "slack", providerAccountId: "team", externalId: randomUUID() }).returning();
+    await db.insert(chatIdentityLinks).values({ companyId, endpointId, principalId: principal.id,
+      confirmationTokenHash: createHash("sha256").update(token).digest("hex"), expiresAt: new Date(Date.now() + 600_000) });
+    const links = createSlackIdentityLinks(db, {
+      previewIdentityLink: async () => ({ selfService: true, canConfirm: false, companyId, endpointId }),
+      stageProviderEffect: vi.fn(async () => null),
+    });
+    expect(await links.requestIdentityAccess(token, userId, "127.0.0.1")).toEqual({ status: "pending_approval" });
+    await settle(); const [row] = await db.select().from(joinRequests);
+    expect(coordinator!.nextWakeAt()).toBe(row.updatedAt.getTime() + delay);
+    await links.requestIdentityAccess(token, userId, "127.0.0.1"); await settle();
+    expect(await db.select().from(joinRequests)).toHaveLength(1);
+  });
+
+  it("disarms after the real approval route resolves a human request", async () => {
+    const requester = randomUUID();
+    await db.insert(authUsers).values({ id: requester, name: "Joiner", email: `${requester}@example.test`, createdAt: new Date(), updatedAt: new Date() });
+    const row = await create();
+    await db.update(joinRequests).set({ requestType: "human", requestingUserId: requester }).where(eq(joinRequests.id, row.id));
+    await start(); await settle(); expect(coordinator!.nextWakeAt()).not.toBeNull();
+    const response = await request(app()).post(`/api/companies/${companyId}/join-requests/${row.id}/approve`).send({});
+    expect(response.status, JSON.stringify(response.body)).toBe(200); await quiet();
+    expect((await db.select().from(joinRequests))[0].status).toBe("approved");
+  });
+
+  it("rechecks activity after waiting for a concurrent request writer", async () => {
+    const row = await create(old()); let release!: () => void, locked!: () => void;
+    const gate = new Promise<void>(resolve => { release = resolve; });
+    const admitted = new Promise<void>(resolve => { locked = resolve; });
+    const write = db.transaction(async tx => {
+      await tx.execute(sql`select id from join_requests where id = ${row.id} for update`); locked(); await gate;
+      await writeJoinRequestChange(tx as unknown as Db, owner => owner.update(joinRequests).set({ updatedAt: new Date() }).where(eq(joinRequests.id, row.id)));
+    });
+    await admitted; const sweep = joinRequestRetentionService(db).sweep();
+    await vi.advanceTimersByTimeAsync(100); release(); await write; await sweep;
+    expect(await state(row.id)).toBeNull();
+  });
+
+  it("pages past failed requests with database timestamp precision and retries them later", async () => {
+    const ids: string[] = [];
+    for (let i = 0; i < 101; i++) ids.push((await create(old())).id);
+    ids.sort(); const last = ids.at(-1)!;
+    await db.execute(sql`update join_requests set updated_at = date_trunc('milliseconds', now() - interval '91 days') + interval '456 microseconds'`);
+    await db.execute(sql`create function join_retention_fail() returns trigger language plpgsql as $$ begin
+      if new.archived_at is not null and new.source_id <> TG_ARGV[0] then raise exception 'retention unavailable'; end if;
+      return new; end $$`);
+    await db.execute(sql`create trigger join_retention_fail before update on decision_retention for each row execute function join_retention_fail(${sql.raw("'" + last + "'")})`);
+    const service = joinRequestRetentionService(db);
+    await expect(service.sweep()).rejects.toThrow("Join request retention failed");
+    expect(await service.sweep()).toEqual({ archived: 1 }); expect((await state(last))?.archivedAt).toBeInstanceOf(Date);
+    await db.execute(sql`drop trigger join_retention_fail on decision_retention`);
+    expect(await service.sweep()).toEqual({ archived: 100 }); expect(await service.sweep()).toEqual({ archived: 0 });
+    expect(await service.next()).toBeNull();
+  });
+
+  it("joins an admitted archive on shutdown and leaves later candidates for restart", async () => {
+    const first = await create(new Date(old().getTime() - 1000)), second = await create(old());
+    let release!: () => void, locked!: () => void;
+    const gate = new Promise<void>(resolve => { release = resolve; });
+    const acquired = new Promise<void>(resolve => { locked = resolve; });
+    const transaction = db.transaction.bind(db);
+    const transactionSpy = vi.spyOn(db, "transaction").mockImplementationOnce((write, options) => transaction(async tx => {
+      locked(); await gate; return write(tx);
+    }, options));
+    const starting = start(); await acquired;
+    let stopped = false; const stopping = coordinator!.stop().then(() => { stopped = true; });
+    try { await vi.advanceTimersByTimeAsync(500); expect(stopped).toBe(false); } finally { release(); }
+    await starting; await stopping; coordinator = undefined; transactionSpy.mockRestore();
+    expect((await state(first.id))?.archivedAt).toBeInstanceOf(Date); expect(await state(second.id)).toBeNull();
+    await start(); await archived(second.id); await quiet();
+  });
+
+
+  it("moves the saved deadline after a supported gateway invite replay", async () => {
+    const invitation = await invite(); await start(); await quiet();
+    const body = { requestType: "agent", agentName: "Gateway", adapterType: "openclaw_gateway",
+      agentDefaultsPayload: { url: "ws://127.0.0.1:18789", disableDeviceAuth: true, headers: { "x-openclaw-token": "fixture-gateway-token-1234567890" } } };
+    const first = await request(app()).post(`/api/invites/${invitation.token}/accept`).send(body);
+    expect(first.status, JSON.stringify(first.body)).toBe(202); await settle();
+    const initial = coordinator!.nextWakeAt();
+    await vi.advanceTimersByTimeAsync(60_000);
+    const replay = await request(app()).post(`/api/invites/${invitation.token}/accept`).send({ ...body, agentName: "Renamed gateway" });
+    expect(replay.status, JSON.stringify(replay.body)).toBe(202); await settle();
+    const rows = await db.select().from(joinRequests); expect(rows).toHaveLength(1);
+    expect(coordinator!.nextWakeAt()).toBe(rows[0].updatedAt.getTime() + delay);
+    expect(coordinator!.nextWakeAt()).toBeGreaterThan(initial!);
+  });
+
+});

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

@@ -1,3 +1,4 @@
+import { joinRequestRetentionService } from "../services/join-request-retention.js";
 import { approvalRetentionService } from "../services/approval-retention.js";
 import { logger } from "../middleware/logger.js";
 import { getStartupRecoveryState } from "../startup-recovery-state.js";
@@ -322,6 +323,11 @@ vi.mock("../realtime/live-events-ws.js", () => ({
   setupLiveEventsWebSocketServer: vi.fn(),
 }));
 
+vi.mock("../services/join-request-retention.js", async importOriginal => ({
+  ...await importOriginal<typeof import("../services/join-request-retention.js")>(),
+  joinRequestRetentionService: vi.fn(() => ({ sweep: vi.fn(async () => ({ archived: 0 })), 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) })),
@@ -537,6 +543,18 @@ describe("startServer feedback export wiring", () => {
     }
   });
 
+  it.each([false, true])("registers join request retention 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.joinRequestRetention);
+    expect(Boolean(registration)).toBe(enabled);
+    if (enabled) {
+      const service = vi.mocked(joinRequestRetentionService).mock.results.at(-1)!.value;
+      await registration![1].run(new AbortController().signal);
+      expect(service.sweep).toHaveBeenCalled(); expect(service.next).toHaveBeenCalled();
+    }
+  });
+
   it("registers durable case handoffs independently of the heartbeat scheduler", async () => {
     await startServer();
     expect(caseHandoffDeliveryServiceFactoryMock).toHaveBeenCalledWith(createDbMock.mock.results[0]!.value, { heartbeat: heartbeatServiceMock });

server/src/index.ts

@@ -1,3 +1,4 @@
+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";
 import { registerDecisionEffectRecoveryWork } from "./services/decision-effect-recovery-work.js";
@@ -1776,7 +1777,7 @@ async function startServerWithDatabaseTeardown(
         // per company per tick.
         const page = await attentionService(db as any).list(company.id, {
           includeDismissed: true,
-          excludeSources: ["approval"],
+          excludeSources: ["approval", "join_request"],
           all: true,
           allowUnscopedAll: true,
         });
@@ -1826,6 +1827,8 @@ async function startServerWithDatabaseTeardown(
         () => !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),
+        () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready);
       await step("retention_sweep", runRetentionSweep);
     };
 

server/src/routes/access.ts

@@ -1,3 +1,4 @@
+import { notifyRetentionWork, writeJoinRequestChange } from "../services/retention-work-notifications.js";
 import { createAgentLifecycle } from "../services/agent-lifecycle.js";
 import {
   createHash,
@@ -3136,7 +3137,7 @@ export function accessRoutes(
     const approvedAt = new Date();
     const approvedByUserId =
       input.invite.invitedByUserId ?? (isLocalImplicit(input.req) ? "local-board" : null);
-    const approved = await db
+    const approved = await writeJoinRequestChange(db, tx => tx
       .update(joinRequests)
       .set({
         status: "approved",
@@ -3146,7 +3147,7 @@ export function accessRoutes(
       })
       .where(eq(joinRequests.id, input.joinRequest.id))
       .returning()
-      .then((rows) => rows[0] ?? null);
+      .then((rows) => rows[0] ?? null));
 
     await logActivity(db, {
       companyId: input.companyId,
@@ -3882,6 +3883,7 @@ export function accessRoutes(
                   )
                 );
 
+              await notifyRetentionWork(tx, "join_request");
               const row = await tx
                 .insert(joinRequests)
                 .values({
@@ -3912,7 +3914,7 @@ export function accessRoutes(
                 .then((rows) => rows[0]);
               return row;
             })
-        : await db
+        : await writeJoinRequestChange(db, tx => tx
             .update(joinRequests)
             .set({
               requestIp: requestIp(req),
@@ -3935,7 +3937,7 @@ export function accessRoutes(
             })
             .where(eq(joinRequests.id, replayJoinRequestId as string))
             .returning()
-            .then((rows) => rows[0]);
+            .then((rows) => rows[0]));
 
       if (!created) {
         throw conflict("Join request not found");
@@ -4279,7 +4281,7 @@ export function accessRoutes(
         );
       }
 
-      const approved = await db
+      const approved = await writeJoinRequestChange(db, tx => tx
         .update(joinRequests)
         .set({
           status: "approved",
@@ -4291,7 +4293,7 @@ export function accessRoutes(
         })
         .where(eq(joinRequests.id, requestId))
         .returning()
-        .then((rows) => rows[0]);
+        .then((rows) => rows[0]));
 
       await logActivity(db, {
         companyId,
@@ -4338,7 +4340,7 @@ export function accessRoutes(
       if (existing.status !== "pending_approval")
         throw conflict("Join request is not pending");
 
-      const rejected = await db
+      const rejected = await writeJoinRequestChange(db, tx => tx
         .update(joinRequests)
         .set({
           status: "rejected",
@@ -4349,7 +4351,7 @@ export function accessRoutes(
         })
         .where(eq(joinRequests.id, requestId))
         .returning()
-        .then((rows) => rows[0]);
+        .then((rows) => rows[0]));
 
       await logActivity(db, {
         companyId,

server/src/services/attention.ts

@@ -1375,6 +1375,7 @@ export function attentionService(db: Db, serviceOptions: AttentionServiceOptions
         }));
       }
 
+      if (!options.excludeSources?.includes("join_request")) {
       const pendingJoins = await db
         .select({
           id: joinRequests.id,
@@ -1436,6 +1437,8 @@ export function attentionService(db: Db, serviceOptions: AttentionServiceOptions
         }));
       }
 
+      }
+
       const recoveryRows = await db
         .select()
         .from(issueRecoveryActions)

server/src/services/budgets.ts

@@ -1,4 +1,4 @@
-import { notifyRetentionWork } from "./retention-work-notifications.js";
+import { notifyCompanyRetentionWork, notifyRetentionWork } 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";
@@ -297,7 +297,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica
 
     const scope = await resolveScopeRecord(db, "company", policy.scopeId);
     if (scope.paused) return;
-    await notifyRetentionWork(db, "approval");
+    await notifyCompanyRetentionWork(db);
     await notifyDeliveryWork(db, DELIVERY_QUEUES.agentLifecycle, policy.companyId);
     await notifyDeliveryWork(db, DELIVERY_QUEUES.skillSourceSync);
     await db
@@ -354,7 +354,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 notifyRetentionWork(db, "approval");
+    await notifyCompanyRetentionWork(db);
     await notifyDeliveryWork(db, DELIVERY_QUEUES.agentLifecycle, policy.companyId);
     await db
       .update(companies)

server/src/services/chat-channels/slack/identity-links.ts

@@ -1,3 +1,4 @@
+import { notifyRetentionWork } from "../../retention-work-notifications.js";
 import { createHash, randomBytes } from "node:crypto";
 import { and, desc, eq, gt, or, sql } from "drizzle-orm";
 
@@ -97,6 +98,7 @@ export function createSlackIdentityLinks(db: Db, { previewIdentityLink, stagePro
         inviteType: "company_join", allowedJoinTypes: "human", acceptedAt: now, expiresAt: now,
         defaultsPayload: { human: { role: "operator" }, source: "slack_identity_link" },
       }).returning({ id: invites.id });
+      await notifyRetentionWork(tx, "join_request");
       const [request] = await tx.insert(joinRequests).values({
         inviteId: invite.id, companyId: preview.companyId, requestType: "human", status: "pending_approval",
         requestIp, requestingUserId: userId, requestEmailSnapshot: user.email,

server/src/services/companies.ts

@@ -1,4 +1,4 @@
-import { notifyRetentionWork } from "./retention-work-notifications.js";
+import { notifyCompanyRetentionWork } from "./retention-work-notifications.js";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
 import { isRunnerApiResponseAsset } from "./asset-content-access.js";
 import { deleteCompany } from "./company-deletion.js";
@@ -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 notifyRetentionWork(tx, "approval");
+          await notifyCompanyRetentionWork(tx);
           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 notifyRetentionWork(tx, "approval");
+          await notifyCompanyRetentionWork(tx);
           await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentLifecycle, id);
           await tx
             .update(companies)

server/src/services/company-deletion.ts

@@ -1,4 +1,4 @@
-import { notifyRetentionWork } from "./retention-work-notifications.js";
+import { notifyCompanyRetentionWork } from "./retention-work-notifications.js";
 import { notifyDecisionTargetWork } from "./decision-target-work.js";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
 import { agentObservabilityService } from "./agent-observability.js";
@@ -95,7 +95,7 @@ export async function deleteCompany(db: Db, id: string) {
         await tx.delete(agentRuntimeState).where(eq(agentRuntimeState.companyId, id));
         await tx.delete(issueComments).where(eq(issueComments.companyId, id));
         await tx.delete(approvalComments).where(eq(approvalComments.companyId, id));
-        await notifyRetentionWork(tx, "approval");
+        await notifyCompanyRetentionWork(tx);
         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/delivery-work-notifications.ts

@@ -9,6 +9,7 @@ export const DELIVERY_QUEUES = {
   executionDisposition: "execution-disposition",
   environmentCleanup: "environment-cleanup",
   approvalRetention: "approval-retention",
+  joinRequestRetention: "join-request-retention",
   decisionTargets: "decision-targets",
   decisionEffectRecovery: "decision-effect-recovery",
   decisionExpiry: "decision-expiry",

server/src/services/join-request-retention.ts

@@ -0,0 +1,68 @@
+import { and, asc, eq, isNull, lte, or, sql, type SQL } from "drizzle-orm";
+import { companies, decisionRetention, invites, joinRequests, type Db } from "@paperclipai/db";
+import { decisionRetentionService, DEFAULT_DECISION_ARCHIVE_DAYS } from "./decision-retention.js";
+import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js";
+import { registerDeadlineWork } from "./deadline-work.js";
+import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
+
+const archiveDelay = DEFAULT_DECISION_ARCHIVE_DAYS * 86_400_000;
+
+export function joinRequestRetentionService(db: Db) {
+  let cursor: { at: string; id: string } | undefined;
+  function candidates(database: Db, condition?: SQL) {
+    return database.select({ id: joinRequests.id, companyId: joinRequests.companyId,
+      at: joinRequests.updatedAt, cursorAt: sql<string>`${joinRequests.updatedAt}::text` }).from(joinRequests)
+      .innerJoin(companies, eq(companies.id, joinRequests.companyId))
+      .innerJoin(invites, and(eq(invites.id, joinRequests.inviteId), eq(invites.companyId, joinRequests.companyId)))
+      .leftJoin(decisionRetention, and(eq(decisionRetention.companyId, joinRequests.companyId),
+        eq(decisionRetention.sourceKind, "join_request"), sql`${decisionRetention.sourceId} = ${joinRequests.id}::text`))
+      .where(and(sql`${joinRequests.status} = 'pending_approval'`, eq(companies.status, "active"),
+        or(isNull(decisionRetention.id), eq(decisionRetention.keep, false)), isNull(decisionRetention.archivedAt), condition))
+      .orderBy(asc(joinRequests.updatedAt), asc(joinRequests.id));
+  }
+  async function sweep(now = new Date(), signal?: AbortSignal) {
+    const cutoff = new Date(now.getTime() - archiveDelay);
+    const rows = await candidates(db, and(lte(joinRequests.updatedAt, cutoff), cursor
+      ? sql`(${joinRequests.updatedAt}, ${joinRequests.id}) > (${cursor.at}::timestamptz, ${cursor.id}::uuid)` : undefined)).limit(100);
+    let archived = 0;
+    const errors: unknown[] = [];
+    for (const candidate of rows) {
+      if (signal?.aborted) break;
+      try {
+        archived += await db.transaction(async tx => {
+          const database = tx as unknown as Db;
+          const [current] = await candidates(database, and(eq(joinRequests.id, candidate.id), lte(joinRequests.updatedAt, cutoff)))
+            .for("update", { of: joinRequests });
+          if (!current) return 0;
+          const retention = decisionRetentionService(database);
+          const source = { sourceKind: "join_request" as const, subject: { id: current.id }, activityAt: current.at.toISOString() };
+          const states = await retention.syncItems(current.companyId, [source]);
+          const state = states.get(`join_request:${current.id}`)!;
+          return retention.autoArchive({ companyId: current.companyId, now, items: [{ ...source,
+            keep: state.keep, archivedAt: state.archivedAt?.toISOString() ?? null, retentionVersion: state.version,
+          }] });
+        });
+      } catch (error) { errors.push(error); }
+      cursor = { at: candidate.cursorAt, id: candidate.id };
+    }
+    if (rows.length < 100) cursor = undefined;
+    if (errors.length === 1) throw errors[0];
+    if (errors.length) throw new AggregateError(errors, "Join request retention failed");
+    return { archived };
+  }
+  async function next() {
+    const [row] = await candidates(db).limit(1);
+    return row ? row.at.getTime() + archiveDelay : null;
+  }
+  return { sweep, next };
+}
+
+export function registerJoinRequestRetentionWork(
+  coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
+  service: ReturnType<typeof joinRequestRetentionService>,
+  canRun: () => boolean,
+) {
+  return registerDeadlineWork(coordinator, DELIVERY_QUEUES.joinRequestRetention, {
+    run: signal => service.sweep(new Date(), signal), next: service.next, canRun, retryMs: 30_000,
+  });
+}

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

@@ -4,6 +4,7 @@ import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notificatio
 
 export async function notifyRetentionWork(transaction: object, kind: AttentionSourceKind) {
   if (kind === "approval") await notifyDeliveryWork(transaction, DELIVERY_QUEUES.approvalRetention);
+  if (kind === "join_request") await notifyDeliveryWork(transaction, DELIVERY_QUEUES.joinRequestRetention);
 }
 
 export function writeApprovalChange<T>(db: Db, write: (transaction: Db) => Promise<T>): Promise<T> {
@@ -12,3 +13,15 @@ export function writeApprovalChange<T>(db: Db, write: (transaction: Db) => Promi
     return write(tx as unknown as Db);
   });
 }
+
+export async function notifyCompanyRetentionWork(transaction: object) {
+  await notifyRetentionWork(transaction, "approval");
+  await notifyRetentionWork(transaction, "join_request");
+}
+
+export function writeJoinRequestChange<T>(db: Db, write: (transaction: Db) => Promise<T>): Promise<T> {
+  return db.transaction(async tx => {
+    await notifyRetentionWork(tx, "join_request");
+    return write(tx as unknown as Db);
+  });
+}