diff --git a/docs/catalog.json b/docs/catalog.json
index 976406a4..47b25070 100644
--- a/docs/catalog.json
+++ b/docs/catalog.json
@@ -686,7 +686,8 @@
         "packages/core/src/compliance/plugin.ts",
         "packages/core/src/engagement/notifications/plugin.ts",
         "packages/core/src/pam/tag/plugin.ts",
-        "packages/core/src/server/runtime/create-app.ts"
+        "packages/core/src/server/runtime/create-app.ts",
+        "packages/core/src/wallet/plugin.ts"
       ]
     },
     {
diff --git a/packages/core/src/contracts/adapters/audit.ts b/packages/core/src/contracts/adapters/audit.ts
index 08295193..76bae753 100644
--- a/packages/core/src/contracts/adapters/audit.ts
+++ b/packages/core/src/contracts/adapters/audit.ts
@@ -24,7 +24,8 @@ export type DirectAuditAction =
   | 'wallet.auto_withdrawal_rule.set'
   | 'wallet.auto_withdrawal_rule.deleted'
   | 'wallet.auto_withdrawal_config.set'
-  | 'wallet.manual_adjustment.created';
+  | 'wallet.manual_adjustment.created'
+  | 'wallet.custody.sweep_cycle';
 
 /**
  * Every value the audit `action` column legitimately holds: a cross-module domain
diff --git a/packages/core/src/wallet/__tests__/custody-sweep-gate.test.ts b/packages/core/src/wallet/__tests__/custody-sweep-gate.test.ts
new file mode 100644
index 00000000..59843218
--- /dev/null
+++ b/packages/core/src/wallet/__tests__/custody-sweep-gate.test.ts
@@ -0,0 +1,90 @@
+import { describe, it, expect } from 'vitest';
+import { gateSweepBalance } from '../service/custody-sweep.service.js';
+
+const base = {
+  amount: '100',
+  estimatedFee: '1',
+  minDeposit: '10',
+  feeMultiple: '5',
+  sweepFeeCeiling: null as string | null,
+  poolLiquidityFloor: null as string | null,
+  poolBalance: null as string | null,
+};
+
+describe('gateSweepBalance', () => {
+  it('sweeps a balance that clears every gate', () => {
+    expect(gateSweepBalance(base)).toBe('sweep');
+  });
+
+  describe('dust floor (amount >= minDeposit)', () => {
+    it('skips strictly below the floor', () => {
+      expect(gateSweepBalance({ ...base, amount: '9.999999999999999999' })).toBe('dust');
+    });
+
+    it('sweeps exactly at the floor', () => {
+      expect(gateSweepBalance({ ...base, amount: '10' })).toBe('sweep');
+    });
+  });
+
+  describe('fee-multiple floor (amount >= estimatedFee * feeMultiple)', () => {
+    it('skips strictly below the floor', () => {
+      // fee 1 * multiple 5 = 5; amount must be >= 5. Use 4.999... below it, but still
+      // above the dust floor (10) so dust never masks this gate.
+      expect(gateSweepBalance({ ...base, minDeposit: '1', amount: '4.999999999999999999' })).toBe(
+        'fee',
+      );
+    });
+
+    it('sweeps exactly at the floor', () => {
+      expect(gateSweepBalance({ ...base, minDeposit: '1', amount: '5' })).toBe('sweep');
+    });
+  });
+
+  describe('fee ceiling (skip when estimatedFee > sweepFeeCeiling)', () => {
+    it('sweeps when the fee is exactly at the ceiling', () => {
+      expect(gateSweepBalance({ ...base, estimatedFee: '2', sweepFeeCeiling: '2' })).toBe('sweep');
+    });
+
+    it('skips when the fee exceeds the ceiling and there is no liquidity floor', () => {
+      expect(gateSweepBalance({ ...base, estimatedFee: '3', sweepFeeCeiling: '2' })).toBe(
+        'ceiling',
+      );
+    });
+
+    it('skips when the fee exceeds the ceiling and the pool is at or above the floor', () => {
+      expect(
+        gateSweepBalance({
+          ...base,
+          estimatedFee: '3',
+          sweepFeeCeiling: '2',
+          poolLiquidityFloor: '1000',
+          poolBalance: '1000',
+        }),
+      ).toBe('ceiling');
+    });
+
+    it('overrides the ceiling only when the pool is strictly below the liquidity floor', () => {
+      expect(
+        gateSweepBalance({
+          ...base,
+          estimatedFee: '3',
+          sweepFeeCeiling: '2',
+          poolLiquidityFloor: '1000',
+          poolBalance: '999.999999999999999999',
+        }),
+      ).toBe('sweep');
+    });
+
+    it('never overrides on a null poolBalance even with a floor configured', () => {
+      expect(
+        gateSweepBalance({
+          ...base,
+          estimatedFee: '3',
+          sweepFeeCeiling: '2',
+          poolLiquidityFloor: '1000',
+          poolBalance: null,
+        }),
+      ).toBe('ceiling');
+    });
+  });
+});
diff --git a/packages/core/src/wallet/__tests__/custody-sweep.service.int.test.ts b/packages/core/src/wallet/__tests__/custody-sweep.service.int.test.ts
new file mode 100644
index 00000000..56ed40dc
--- /dev/null
+++ b/packages/core/src/wallet/__tests__/custody-sweep.service.int.test.ts
@@ -0,0 +1,317 @@
+import { randomUUID } from 'node:crypto';
+import { describe, it, expect, vi, beforeAll, afterAll, beforeEach } from 'vitest';
+import { eq, sql } from 'drizzle-orm';
+import { createTestDb, type TestDb } from '@openora/core/testing';
+import {
+  definePlatformConfig,
+  type CustodyBalance,
+  type PaymentAdapter,
+} from '@openora/core/contracts';
+import { makeAuditWriter, makePaymentProviderRegistry, mock } from '../../testing/mock.js';
+import { migrate } from '../migrate.js';
+import {
+  wallet,
+  walletAsset,
+  walletBalance,
+  walletCustodySweep,
+  walletJobRun,
+  walletReconciliationFinding,
+  walletTransaction,
+} from '../schema/index.js';
+import { CustodySweepService, CUSTODY_SWEEP_JOB_NAME } from '../service/custody-sweep.service.js';
+
+let db: TestDb;
+
+beforeAll(async () => {
+  db = await createTestDb([migrate]);
+});
+
+afterAll(async () => {
+  await db.drop();
+});
+
+beforeEach(async () => {
+  await db.drizzle.db.execute(
+    sql`TRUNCATE ${walletReconciliationFinding}, ${walletCustodySweep}, ${walletJobRun}, ${walletAsset}, ${walletTransaction}, ${walletBalance}, ${wallet} RESTART IDENTITY CASCADE`,
+  );
+});
+
+const platformConfig = (overrides: Partial<Record<string, unknown>> = {}) =>
+  definePlatformConfig({
+    wallet: {
+      sweep: {
+        cron: '*/15 * * * *',
+        feeMultiple: '1',
+        batchSize: 200,
+        concurrency: 4,
+        unknownAfterMinutes: 60,
+        staleRunAfterMinutes: 30,
+        ...overrides,
+      },
+    },
+  });
+
+async function seedAsset(overrides: Partial<typeof walletAsset.$inferInsert> = {}) {
+  await db.drizzle.db.insert(walletAsset).values({
+    currency: 'USDT',
+    network: 'TRC20',
+    providerAssetId: 'usdt-trc20',
+    minDeposit: '1',
+    minWithdrawal: '1',
+    withdrawalFee: '0.1',
+    ...overrides,
+  });
+}
+
+function makeBalance(overrides: Partial<CustodyBalance> = {}): CustodyBalance {
+  return {
+    userId: randomUUID(),
+    currency: 'USDT',
+    network: 'TRC20',
+    amount: '100',
+    estimatedFee: '1',
+    ...overrides,
+  };
+}
+
+function serviceWith(adapter: PaymentAdapter, config = platformConfig()) {
+  const paymentProviders = makePaymentProviderRegistry({ adapter });
+  const audit = makeAuditWriter();
+  const service = new CustodySweepService({
+    drizzle: db.drizzle,
+    paymentProviders,
+    audit,
+    platformConfig: config,
+  });
+  return { service, audit };
+}
+
+async function sweepRows(userId: string) {
+  return db.drizzle.db
+    .select()
+    .from(walletCustodySweep)
+    .where(eq(walletCustodySweep.userId, userId));
+}
+
+describe('CustodySweepService (real PG)', () => {
+  it('no-ops without touching the database when sweep policy is unconfigured', async () => {
+    const listSweepableBalances = vi.fn().mockResolvedValue([makeBalance()]);
+    const { service } = serviceWith(
+      mock<PaymentAdapter>({ listSweepableBalances, sweepToPool: vi.fn() }),
+      definePlatformConfig({}),
+    );
+
+    const result = await service.runCycle();
+
+    expect(result).toBeNull();
+    expect(listSweepableBalances).not.toHaveBeenCalled();
+    const runs = await db.drizzle.db.select().from(walletJobRun);
+    expect(runs).toHaveLength(0);
+  });
+
+  it('a second cycle does not re-sweep an in-flight balance', async () => {
+    await seedAsset();
+    const b = makeBalance();
+    const listSweepableBalances = vi.fn().mockResolvedValue([b]);
+    const sweepToPool = vi
+      .fn()
+      .mockResolvedValue({ externalId: 'vendor-ref-1', poolRef: 'pool-players-1' });
+    const { service } = serviceWith(mock<PaymentAdapter>({ listSweepableBalances, sweepToPool }));
+
+    const first = await service.runCycle();
+    expect(first?.summary.swept).toBe(1);
+
+    const second = await service.runCycle();
+    expect(second?.summary.swept).toBe(0);
+    expect(second?.summary.inFlight).toBe(1);
+
+    expect(sweepToPool).toHaveBeenCalledTimes(1);
+    const rows = await sweepRows(b.userId);
+    expect(rows).toHaveLength(1);
+    expect(rows[0]?.status).toBe('processing');
+    // Which pool received the funds, not merely that a transfer happened - the
+    // separation of player money from operator money has to be evidenced.
+    expect(rows[0]?.poolRef).toBe('pool-players-1');
+  });
+
+  it('sweepToPool throws -> the guard is still held and the next cycle does not re-sweep', async () => {
+    await seedAsset();
+    const b = makeBalance();
+    const listSweepableBalances = vi.fn().mockResolvedValue([b]);
+    const sweepToPool = vi.fn().mockRejectedValue(new Error('vendor timed out'));
+    const { service } = serviceWith(mock<PaymentAdapter>({ listSweepableBalances, sweepToPool }));
+
+    const first = await service.runCycle();
+    expect(first?.summary.unknown).toBe(1);
+    expect(first?.summary.swept).toBe(0);
+
+    const rowsAfterFirst = await sweepRows(b.userId);
+    expect(rowsAfterFirst).toHaveLength(1);
+    expect(rowsAfterFirst[0]?.status).toBe('unknown');
+
+    const second = await service.runCycle();
+    expect(second?.summary.inFlight).toBe(1);
+    expect(second?.summary.swept).toBe(0);
+
+    // Still exactly the one row from the first cycle - the throw never released the guard.
+    expect(sweepToPool).toHaveBeenCalledTimes(1);
+    const rowsAfterSecond = await sweepRows(b.userId);
+    expect(rowsAfterSecond).toHaveLength(1);
+    expect(rowsAfterSecond[0]?.status).toBe('unknown');
+  });
+
+  it('a swept balance changes no wallet_balance row and creates no wallet_transaction', async () => {
+    await seedAsset();
+    const b = makeBalance();
+    const adapter = mock<PaymentAdapter>({
+      listSweepableBalances: vi.fn().mockResolvedValue([b]),
+      sweepToPool: vi.fn().mockResolvedValue({ externalId: 'vendor-ref-2' }),
+    });
+    const { service } = serviceWith(adapter);
+
+    const result = await service.runCycle();
+    expect(result?.summary.swept).toBe(1);
+
+    const balances = await db.drizzle.db.select().from(walletBalance);
+    const transactions = await db.drizzle.db.select().from(walletTransaction);
+    expect(balances).toHaveLength(0);
+    expect(transactions).toHaveLength(0);
+  });
+
+  it('two cycles started concurrently: the second returns immediately and listSweepableBalances is called exactly once', async () => {
+    await seedAsset();
+    // The vendor call has to be slow enough to hold the first cycle's claim open while
+    // the second one attempts its own. Resolved immediately, a cycle can finish - and
+    // clear its claim - before the sibling's insert is even issued, at which point both
+    // legitimately succeed as two SEQUENTIAL runs and the test fails without anything
+    // being wrong. That is a race in the test, not in the claim.
+    const balances = [makeBalance()];
+    const listSweepableBalances = vi.fn(
+      () => new Promise((resolve) => setTimeout(() => resolve(balances), 100)),
+    );
+    const sweepToPool = vi.fn().mockResolvedValue({ externalId: randomUUID() });
+    const { service } = serviceWith(mock<PaymentAdapter>({ listSweepableBalances, sweepToPool }));
+
+    const [a, b] = await Promise.all([service.runCycle(), service.runCycle()]);
+    const results = [a, b];
+    const claimed = results.filter((r) => r !== null);
+    const skipped = results.filter((r) => r === null);
+
+    expect(claimed).toHaveLength(1);
+    expect(skipped).toHaveLength(1);
+    expect(listSweepableBalances).toHaveBeenCalledTimes(1);
+  });
+
+  it('a missing wallet_asset row produces an unconfigured_asset finding', async () => {
+    // Deliberately no seedAsset() call - (currency, network) is unconfigured.
+    const b = makeBalance({ currency: 'DOGE', network: 'MAINNET' });
+    const listSweepableBalances = vi.fn().mockResolvedValue([b]);
+    const sweepToPool = vi.fn().mockResolvedValue({ externalId: 'never-called' });
+    const { service } = serviceWith(mock<PaymentAdapter>({ listSweepableBalances, sweepToPool }));
+
+    const result = await service.runCycle();
+    expect(result?.summary.swept).toBe(0);
+    expect(result?.summary.skippedDust).toBe(0);
+
+    const findings = await db.drizzle.db.select().from(walletReconciliationFinding);
+    expect(findings).toHaveLength(1);
+    expect(findings[0]).toMatchObject({
+      kind: 'unconfigured_asset',
+      currency: 'DOGE',
+      network: 'MAINNET',
+      amount: '100.000000000000000000',
+      runId: result?.runId,
+    });
+
+    const sweeps = await sweepRows(b.userId);
+    expect(sweeps).toHaveLength(0);
+  });
+
+  it('writes exactly one audit entry per cycle regardless of how many balances were swept', async () => {
+    await seedAsset();
+    const balances = [makeBalance(), makeBalance(), makeBalance()];
+    const adapter = mock<PaymentAdapter>({
+      listSweepableBalances: vi.fn().mockResolvedValue(balances),
+      sweepToPool: vi.fn().mockImplementation(async () => ({ externalId: randomUUID() })),
+    });
+    const { service, audit } = serviceWith(adapter);
+
+    const result = await service.runCycle();
+    expect(result?.summary.swept).toBe(3);
+
+    expect(audit.recordInTransaction).toHaveBeenCalledTimes(1);
+    const [, entry] = (audit.recordInTransaction as ReturnType<typeof vi.fn>).mock.calls[0] as [
+      unknown,
+      { action: string; after: Record<string, unknown> },
+    ];
+    expect(entry.action).toBe('wallet.custody.sweep_cycle');
+    expect(entry.after).toMatchObject({ runId: result?.runId, swept: 3 });
+  });
+
+  it('a repeated unconfigured asset files exactly one finding, not one per cycle', async () => {
+    // Without a stable dedup key this condition - which recurs every single tick until
+    // an operator configures the asset - files a fresh finding per cron tick and pins
+    // the reconciliation alert threshold permanently over the line.
+    const b = makeBalance({ currency: 'DOGE', network: 'MAINNET' });
+    const listSweepableBalances = vi.fn().mockResolvedValue([b]);
+    const sweepToPool = vi.fn();
+    const { service } = serviceWith(mock<PaymentAdapter>({ listSweepableBalances, sweepToPool }));
+
+    await service.runCycle();
+    await service.runCycle();
+
+    const findings = await db.drizzle.db.select().from(walletReconciliationFinding);
+    expect(findings).toHaveLength(1);
+  });
+
+  it('a throwing adapter fails the run and frees the claim for the next cycle', async () => {
+    await seedAsset();
+    const listSweepableBalances = vi
+      .fn()
+      .mockRejectedValueOnce(new Error('vendor 503'))
+      .mockResolvedValue([]);
+    const { service } = serviceWith(
+      mock<PaymentAdapter>({ listSweepableBalances, sweepToPool: vi.fn() }),
+    );
+
+    await expect(service.runCycle()).rejects.toThrow('vendor 503');
+
+    const [failed] = await db.drizzle.db
+      .select()
+      .from(walletJobRun)
+      .where(eq(walletJobRun.status, 'failed'));
+    expect(failed?.finishedAt).not.toBeNull();
+
+    // The whole point: a vendor blip must not hold the single live-run slot until the
+    // staleness window elapses.
+    const second = await service.runCycle();
+    expect(second).not.toBeNull();
+  });
+
+  it('takes over a run whose startedAt is older than the staleness threshold', async () => {
+    const staleRunId = randomUUID();
+    await db.drizzle.db.insert(walletJobRun).values({
+      jobName: CUSTODY_SWEEP_JOB_NAME,
+      runId: staleRunId,
+      startedAt: new Date(Date.now() - 2 * 60_000),
+    });
+    const listSweepableBalances = vi.fn().mockResolvedValue([]);
+    const { service } = serviceWith(
+      mock<PaymentAdapter>({ listSweepableBalances, sweepToPool: vi.fn() }),
+      platformConfig({ staleRunAfterMinutes: 1 }),
+    );
+
+    const result = await service.runCycle();
+
+    expect(result).not.toBeNull();
+    expect(result?.runId).not.toBe(staleRunId);
+    const runs = await db.drizzle.db
+      .select()
+      .from(walletJobRun)
+      .where(eq(walletJobRun.jobName, CUSTODY_SWEEP_JOB_NAME));
+    const stale = runs.find((r) => r.runId === staleRunId);
+    const fresh = runs.find((r) => r.runId === result?.runId);
+    expect(stale?.status).toBe('abandoned');
+    expect(fresh?.status).toBe('completed');
+  });
+});
diff --git a/packages/core/src/wallet/__tests__/wallet-asset.router.int.test.ts b/packages/core/src/wallet/__tests__/wallet-asset.router.int.test.ts
index dc13f2c6..34d8d8f3 100644
--- a/packages/core/src/wallet/__tests__/wallet-asset.router.int.test.ts
+++ b/packages/core/src/wallet/__tests__/wallet-asset.router.int.test.ts
@@ -79,7 +79,12 @@ function routerWith(
     audit,
     identityReader: makeIdentityReader(),
   });
-  const router = createWalletRouter(service, guard, audit, paymentProviders);
+  const router = createWalletRouter({
+    wallet: service,
+    adminGuard: guard,
+    audit,
+    paymentProviders,
+  });
   return { router, audit, service };
 }
 
diff --git a/packages/core/src/wallet/__tests__/wallet-auto-rule.router.int.test.ts b/packages/core/src/wallet/__tests__/wallet-auto-rule.router.int.test.ts
index 168036eb..d8e24d13 100644
--- a/packages/core/src/wallet/__tests__/wallet-auto-rule.router.int.test.ts
+++ b/packages/core/src/wallet/__tests__/wallet-auto-rule.router.int.test.ts
@@ -55,7 +55,7 @@ function routerWith(adminGuard: AdminGuard) {
     audit,
     identityReader: makeIdentityReader(),
   });
-  const router = createWalletRouter(service, adminGuard, audit, paymentProviders);
+  const router = createWalletRouter({ wallet: service, adminGuard, audit, paymentProviders });
   return { router, audit };
 }
 
diff --git a/packages/core/src/wallet/__tests__/wallet-auto-withdrawal-config.router.int.test.ts b/packages/core/src/wallet/__tests__/wallet-auto-withdrawal-config.router.int.test.ts
index f3420dbc..ee2c022a 100644
--- a/packages/core/src/wallet/__tests__/wallet-auto-withdrawal-config.router.int.test.ts
+++ b/packages/core/src/wallet/__tests__/wallet-auto-withdrawal-config.router.int.test.ts
@@ -106,7 +106,7 @@ function routerWith(adminGuard: AdminGuard, platformConfig?: Partial<PlatformCon
     platformConfig: platformConfig ? mock<PlatformConfig>(platformConfig) : undefined,
     riskTags,
   });
-  const router = createWalletRouter(service, adminGuard, audit, paymentProviders);
+  const router = createWalletRouter({ wallet: service, adminGuard, audit, paymentProviders });
   return { router, audit, service };
 }
 
diff --git a/packages/core/src/wallet/__tests__/wallet-custody-stubs.router.test.ts b/packages/core/src/wallet/__tests__/wallet-custody-stubs.router.test.ts
index f713471c..e489eb92 100644
--- a/packages/core/src/wallet/__tests__/wallet-custody-stubs.router.test.ts
+++ b/packages/core/src/wallet/__tests__/wallet-custody-stubs.router.test.ts
@@ -1,7 +1,7 @@
-import { describe, it, expect } from 'vitest';
+import { describe, it, expect, vi } from 'vitest';
 import { call, ORPCError } from '@orpc/server';
 import type { AdminGuard } from '@openora/core/server';
-import type { PaymentAdapter } from '@openora/core/contracts';
+import type { JobQueueAdapter, PaymentAdapter } from '@openora/core/contracts';
 import {
   mock,
   makeDrizzle,
@@ -18,9 +18,10 @@ import { WalletService } from '../service/wallet.service.js';
 const CTX = testContext();
 const CALLER_ID = '9a2f7c11-0000-4000-8000-0000000000dd';
 
-// These stub routes only ever run adminGuard.assert() then throw NOT_IMPLEMENTED - no DB
-// row is ever read or written, so a mocked DrizzleService is enough (no real Postgres).
-function routerWith(guard: AdminGuard) {
+// The reconciliation routes below only ever run adminGuard.assert() then throw
+// NOT_IMPLEMENTED - no DB row is ever read or written, so a mocked DrizzleService is
+// enough (no real Postgres). The custody sweep route now enqueues instead of throwing.
+function routerWith(guard: AdminGuard, jobQueue?: JobQueueAdapter) {
   const paymentProviders = makePaymentProviderRegistry();
   const service = new WalletService({
     drizzle: makeDrizzle(),
@@ -30,7 +31,13 @@ function routerWith(guard: AdminGuard) {
     audit: makeAuditWriter(),
     identityReader: makeIdentityReader(),
   });
-  return createWalletRouter(service, guard, makeAuditWriter(), paymentProviders);
+  return createWalletRouter({
+    wallet: service,
+    adminGuard: guard,
+    audit: makeAuditWriter(),
+    paymentProviders,
+    jobQueue,
+  });
 }
 
 const authorizedGuard = () => makeAdminGuard({ caller: { userId: CALLER_ID, role: 'admin' } });
@@ -40,12 +47,19 @@ const deniedGuard = (deny: readonly string[]) =>
 
 describe('wallet custody/reconciliation stub routes', () => {
   describe('POST /wallet/custody/sweep/run', () => {
-    it('501s for an authorized caller', async () => {
-      const router = routerWith(authorizedGuard());
+    it('enqueues a sweep cycle and returns its runId for an authorized caller', async () => {
+      const enqueue = vi.fn().mockResolvedValue({ id: 'job-1' });
+      const jobQueue = mock<JobQueueAdapter>({ enqueue });
+      const router = routerWith(authorizedGuard(), jobQueue);
 
-      await expect(call(router.custody.sweep.run, {}, { context: CTX })).rejects.toMatchObject({
-        code: 'NOT_IMPLEMENTED',
-      });
+      const result = await call(router.custody.sweep.run, {}, { context: CTX });
+
+      expect(result.runId).toMatch(
+        /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/,
+      );
+      expect(enqueue).toHaveBeenCalledTimes(1);
+      const [, payload] = enqueue.mock.calls[0] as [unknown, { runId: string }];
+      expect(payload).toEqual({ runId: result.runId });
     });
 
     it('403s for a caller missing wallet-custody:run', async () => {
diff --git a/packages/core/src/wallet/__tests__/wallet-webhook.router.int.test.ts b/packages/core/src/wallet/__tests__/wallet-webhook.router.int.test.ts
index 37bdf66d..793275e4 100644
--- a/packages/core/src/wallet/__tests__/wallet-webhook.router.int.test.ts
+++ b/packages/core/src/wallet/__tests__/wallet-webhook.router.int.test.ts
@@ -72,13 +72,13 @@ function routerWithProviders(
     audit: makeAuditWriter(),
     identityReader: makeIdentityReader(),
   });
-  return createWalletRouter(
-    service,
-    mock<AdminGuard>({ assert: vi.fn() }),
-    makeAuditWriter(),
+  return createWalletRouter({
+    wallet: service,
+    adminGuard: mock<AdminGuard>({ assert: vi.fn() }),
+    audit: makeAuditWriter(),
     paymentProviders,
     limiter,
-  );
+  });
 }
 
 // A registry with two DISTINCTLY-behaving named providers, to test that a webhook is
diff --git a/packages/core/src/wallet/__tests__/wallet.router.int.test.ts b/packages/core/src/wallet/__tests__/wallet.router.int.test.ts
index 99fac8a0..e3ad8f8c 100644
--- a/packages/core/src/wallet/__tests__/wallet.router.int.test.ts
+++ b/packages/core/src/wallet/__tests__/wallet.router.int.test.ts
@@ -49,12 +49,12 @@ function realWalletService() {
 }
 
 function routerWith(adminGuard: AdminGuard) {
-  return createWalletRouter(
-    realWalletService(),
+  return createWalletRouter({
+    wallet: realWalletService(),
     adminGuard,
-    makeAuditWriter(),
-    makePaymentProviderRegistry(),
-  );
+    audit: makeAuditWriter(),
+    paymentProviders: makePaymentProviderRegistry(),
+  });
 }
 
 const transactionDenyingGuard = () =>
diff --git a/packages/core/src/wallet/contract/index.ts b/packages/core/src/wallet/contract/index.ts
index e12b0707..9a4fdd9a 100644
--- a/packages/core/src/wallet/contract/index.ts
+++ b/packages/core/src/wallet/contract/index.ts
@@ -512,7 +512,7 @@ export const walletContract = {
         .route({
           method: 'POST',
           path: '/wallet/custody/sweep/run',
-          summary: NOT_IMPLEMENTED_YET,
+          summary: 'Enqueue a custody sweep cycle',
         })
         .output(JobRunResultSchema),
     },
diff --git a/packages/core/src/wallet/plugin.ts b/packages/core/src/wallet/plugin.ts
index 9fad2922..481b8615 100644
--- a/packages/core/src/wallet/plugin.ts
+++ b/packages/core/src/wallet/plugin.ts
@@ -1,5 +1,5 @@
-import { ADMIN_GUARD, EVENT_BUS, DRIZZLE } from '@openora/core/server';
-import type { CoreTokenCatalog, Plugin } from '@openora/core/server';
+import { ADMIN_GUARD, EVENT_BUS, DRIZZLE, createLogger } from '@openora/core/server';
+import type { CoreTokenCatalog, Plugin, TypedContainer } from '@openora/core/server';
 import * as z from 'zod';
 import {
   ADMIN_USER_DIRECTORY,
@@ -18,9 +18,15 @@ import {
   TAG_EVALUATION_COMMANDS,
   PLAY_ELIGIBILITY,
   AUDIT_WRITER,
+  JOB_QUEUE,
 } from '@openora/core/contracts';
 import { WalletService } from './service/wallet.service.js';
 import { WalletCommandsService } from './service/wallet-commands.service.js';
+import {
+  CustodySweepService,
+  CUSTODY_SWEEP_QUEUE,
+  CustodySweepJobPayloadSchema,
+} from './service/custody-sweep.service.js';
 import { WalletReaderService } from './adapters/wallet-reader.service.js';
 import { WalletAssetCatalogService } from './adapters/wallet-asset-catalog.service.js';
 import { DrizzleAdminWalletReporting } from './admin-reporting.js';
@@ -28,6 +34,13 @@ import { createWalletRouter } from './router/index.js';
 import { MockPaymentAdapter } from './adapters/mock/mock-payment-adapter.js';
 import { HmacPaymentWebhookVerifier } from './adapters/hmac-payment-webhook-verifier.js';
 
+const logger = createLogger('wallet');
+
+// Mirrors WalletConfigSchema's wallet.sweep.cron default - used only when the operator
+// hasn't configured platformConfig.wallet.sweep at all, so the job is still scheduled
+// (CustodySweepService.runCycle no-ops every tick until sweep policy is configured).
+const DEFAULT_SWEEP_CRON = '*/15 * * * *';
+
 export default {
   // NOT dependsOn 'tag': that would cycle (tag hard-depends on wallet's WALLET_READER).
   // wallet's use of tag's PLAYER_TAGS / TAG_EVALUATION_COMMANDS is optional and resolved
@@ -73,9 +86,41 @@ export default {
     // Operator-editable currency/network config, readable by a payment adapter without
     // importing wallet tables. Overlay-rebindable, but bound here so it always works.
     ctx.provide(WALLET_ASSET_CATALOG, (c) => new WalletAssetCatalogService(c.get(DRIZZLE)));
-    ctx.routers.add('wallet', (c) =>
-      createWalletRouter(
-        new WalletService({
+
+    // One memoized instance backs the cron worker and the router factory below - lazily
+    // constructed (subscriptions/workers wire before router factories run), matching
+    // pam/tag's and compliance's job-worker shape.
+    let sweepSvc: CustodySweepService | null = null;
+    const custodySweepService = (c: TypedContainer<CoreTokenCatalog>) =>
+      (sweepSvc ??= new CustodySweepService({
+        drizzle: c.get(DRIZZLE),
+        paymentProviders: c.get(PAYMENT_PROVIDERS),
+        audit: c.get(AUDIT_WRITER),
+        platformConfig: c.has(PLATFORM_CONFIG) ? c.get(PLATFORM_CONFIG) : undefined,
+      }));
+
+    ctx.jobs.worker({
+      queue: CUSTODY_SWEEP_QUEUE,
+      schema: CustodySweepJobPayloadSchema,
+      handler: async ({ payload }) => {
+        await sweepSvc?.runCycle(payload.runId);
+      },
+    });
+
+    ctx.routers.add('wallet', (c) => {
+      const jobQueue = c.get(JOB_QUEUE);
+      custodySweepService(c); // ensure sweepSvc is constructed for the job worker above
+      const sweepCron =
+        (c.has(PLATFORM_CONFIG) ? c.get(PLATFORM_CONFIG) : undefined)?.wallet?.sweep?.cron ??
+        DEFAULT_SWEEP_CRON;
+      // Idempotent schedule (keyed by scheduleId). If platformConfig.wallet.sweep is
+      // absent, this still registers the tick - the handler just no-ops every time.
+      void jobQueue
+        .schedule(CUSTODY_SWEEP_QUEUE, 'wallet-custody-sweep.cron', {}, { cron: sweepCron })
+        .catch((err) => logger.error({ err }, 'wallet-custody-sweep schedule failed'));
+
+      return createWalletRouter({
+        wallet: new WalletService({
           drizzle: c.get(DRIZZLE),
           events: c.get(EVENT_BUS),
           payment: c.get(PAYMENT_ADAPTER),
@@ -90,11 +135,12 @@ export default {
             : undefined,
           audit: c.get(AUDIT_WRITER),
         }),
-        c.get(ADMIN_GUARD),
-        c.get(AUDIT_WRITER),
-        c.get(PAYMENT_PROVIDERS),
-        c.get(RATE_LIMITER),
-      ),
-    );
+        adminGuard: c.get(ADMIN_GUARD),
+        audit: c.get(AUDIT_WRITER),
+        paymentProviders: c.get(PAYMENT_PROVIDERS),
+        limiter: c.get(RATE_LIMITER),
+        jobQueue,
+      });
+    });
   },
 } as const satisfies Plugin<CoreTokenCatalog>;
diff --git a/packages/core/src/wallet/router/index.ts b/packages/core/src/wallet/router/index.ts
index 9ebc3ccd..af76795a 100644
--- a/packages/core/src/wallet/router/index.ts
+++ b/packages/core/src/wallet/router/index.ts
@@ -1,3 +1,4 @@
+import { randomUUID } from 'node:crypto';
 import { implement, ORPCError } from '@orpc/server';
 import {
   getUserId,
@@ -11,12 +12,14 @@ import {
   RATE_LIMIT_KEYS,
   makeRateLimitKey,
   type AuditWritePort,
+  type JobQueueAdapter,
   type PaymentProviderRegistry,
   type PaymentWebhookEvent,
   type RateLimiterAdapter,
   type RateLimitKey,
 } from '@openora/core/contracts';
 import { walletContract } from '../contract/index.js';
+import { CUSTODY_SWEEP_QUEUE } from '../service/custody-sweep.service.js';
 import {
   WalletService,
   WalletNotFoundError,
@@ -88,13 +91,29 @@ async function dispatchWebhook(
   return { ok: true as const };
 }
 
-export function createWalletRouter(
-  wallet: WalletService,
-  adminGuard: AdminGuard,
-  audit: AuditWritePort,
-  paymentProviders: PaymentProviderRegistry,
-  limiter?: RateLimiterAdapter<RateLimitKey>,
-) {
+/**
+ * Named rather than positional. Half of these are structurally similar ports, so a
+ * positional slip type-checks; and the sweep and reconciliation branches each add their
+ * own dependencies here, which as positional parameters git merges into a signature that
+ * declares one twice without ever reporting a conflict.
+ */
+export type WalletRouterDeps = {
+  wallet: WalletService;
+  adminGuard: AdminGuard;
+  audit: AuditWritePort;
+  paymentProviders: PaymentProviderRegistry;
+  limiter?: RateLimiterAdapter<RateLimitKey>;
+  jobQueue?: JobQueueAdapter;
+};
+
+export function createWalletRouter({
+  wallet,
+  adminGuard,
+  audit,
+  paymentProviders,
+  limiter,
+  jobQueue,
+}: WalletRouterDeps) {
   const os = implement(walletContract).$context<OssContext>();
 
   const throttleWebhook = (context: OssContext) =>
@@ -394,7 +413,18 @@ export function createWalletRouter(
       sweep: {
         run: os.custody.sweep.run.handler(async ({ context }) => {
           await adminGuard.assert(context, 'wallet-custody', 'run');
-          return notImplemented();
+          if (!jobQueue) {
+            throw new ORPCError('INTERNAL_SERVER_ERROR', {
+              message: 'custody sweep job queue is not wired',
+            });
+          }
+          // The run claim (a unique-index insert in wallet_job_run) is the sole
+          // concurrency authority, not this request - enqueue and return immediately so
+          // an HTTP timeout can never orphan a cycle. runId is minted here so the caller
+          // gets it back synchronously, before the job actually runs.
+          const runId = randomUUID();
+          await jobQueue.enqueue(CUSTODY_SWEEP_QUEUE, { runId });
+          return { runId };
         }),
       },
     },
diff --git a/packages/core/src/wallet/service/custody-sweep.service.ts b/packages/core/src/wallet/service/custody-sweep.service.ts
new file mode 100644
index 00000000..6688cdb9
--- /dev/null
+++ b/packages/core/src/wallet/service/custody-sweep.service.ts
@@ -0,0 +1,502 @@
+import { randomUUID } from 'node:crypto';
+import * as z from 'zod';
+import {
+  type DrizzleService,
+  moneyCompare,
+  moneyScaleBy,
+  mapConcurrent,
+  createLogger,
+} from '@openora/core/server';
+import {
+  queue,
+  UuidSchema,
+  type AuditWritePort,
+  type CustodyBalance,
+  type PaymentAdapter,
+  type PaymentProviderRegistry,
+  type PlatformConfig,
+  type User,
+  type Uuid,
+} from '@openora/core/contracts';
+import { and, eq, inArray, isNotNull, isNull } from 'drizzle-orm';
+import {
+  walletAsset,
+  walletCustodySweep,
+  walletJobRun,
+  walletReconciliationFinding,
+  type WalletAssetRow,
+  type WalletJobRun,
+} from '../schema/index.js';
+
+const logger = createLogger('wallet-custody-sweep');
+
+export const CUSTODY_SWEEP_JOB_NAME = 'wallet-custody-sweep';
+export const CUSTODY_SWEEP_QUEUE = queue('wallet.custody-sweep');
+
+// `runId` is only present when an admin's on-demand POST /wallet/custody/sweep/run
+// enqueued this tick - the router mints it up front so it can hand the caller back a
+// runId synchronously, before the job actually runs. A cron tick omits it and the
+// service mints its own.
+export const CustodySweepJobPayloadSchema = z.object({ runId: UuidSchema.optional() });
+export type CustodySweepJobPayload = z.infer<typeof CustodySweepJobPayloadSchema>;
+
+const IN_FLIGHT_STATUSES = ['pending', 'processing', 'unknown'] as const;
+
+type SweepConfig = NonNullable<NonNullable<PlatformConfig['wallet']>['sweep']>;
+
+export type SweepCycleSummary = {
+  considered: number;
+  swept: number;
+  skippedDust: number;
+  skippedFee: number;
+  skippedCeiling: number;
+  inFlight: number;
+  // Always 0 today: `PaymentAdapter.sweepToPool` can only resolve or throw, so a caught
+  // error has no way to prove "the vendor definitively rejected this and created no
+  // transaction" - see the catch block in processBalance. The field stays in the
+  // summary shape for the day an adapter can prove a definitive rejection.
+  failed: number;
+  unknown: number;
+};
+
+export type SweepGateDecision = 'sweep' | 'dust' | 'fee' | 'ceiling';
+
+/**
+ * Pure gate over a single custody balance, in the exact order
+ * docs/adapters/payment.md diagram D3 specifies: dust floor, then the fee-multiple
+ * floor, then the fee ceiling (overridden only when the pool is running dry).
+ *
+ * `poolBalance` is null whenever the caller hasn't fetched it (no ceiling breach yet,
+ * the adapter has no `getPoolBalance`, or the asset has no `poolLiquidityFloor`) - null
+ * always means "the ceiling stays absolute", never an accidental override.
+ */
+export function gateSweepBalance(args: {
+  amount: string;
+  estimatedFee: string;
+  minDeposit: string;
+  feeMultiple: string;
+  sweepFeeCeiling: string | null;
+  poolLiquidityFloor: string | null;
+  poolBalance: string | null;
+}): SweepGateDecision {
+  if (moneyCompare(args.amount, args.minDeposit) < 0) {
+    return 'dust';
+  }
+  const feeFloor = moneyScaleBy(args.estimatedFee, args.feeMultiple);
+  if (moneyCompare(args.amount, feeFloor) < 0) {
+    return 'fee';
+  }
+  if (args.sweepFeeCeiling !== null && moneyCompare(args.estimatedFee, args.sweepFeeCeiling) > 0) {
+    const poolBelowFloor =
+      args.poolLiquidityFloor !== null &&
+      args.poolBalance !== null &&
+      moneyCompare(args.poolBalance, args.poolLiquidityFloor) < 0;
+    if (!poolBelowFloor) {
+      return 'ceiling';
+    }
+  }
+  return 'sweep';
+}
+
+export type CustodySweepServiceDeps = {
+  drizzle: DrizzleService;
+  paymentProviders: PaymentProviderRegistry;
+  audit: AuditWritePort;
+  platformConfig?: PlatformConfig;
+};
+
+/**
+ * Owns the custody sweep cron: moves vendor-side per-player custody balances into the
+ * pooled account they're paid out of. Deliberately separate from WalletService (which
+ * is already ~2000 lines) and deliberately never touches wallet_balance/wallet_transaction
+ * - the player was credited at deposit time, this only moves the vendor's own money
+ * between its own containers. See docs/adapters/payment.md, "Custody: pooling, sweeping
+ * and reconciliation".
+ */
+export class CustodySweepService {
+  private readonly drizzle: DrizzleService;
+  private readonly paymentProviders: PaymentProviderRegistry;
+  private readonly audit: AuditWritePort;
+  private readonly platformConfig?: PlatformConfig;
+
+  constructor({ drizzle, paymentProviders, audit, platformConfig }: CustodySweepServiceDeps) {
+    this.drizzle = drizzle;
+    this.paymentProviders = paymentProviders;
+    this.audit = audit;
+    this.platformConfig = platformConfig;
+  }
+
+  /**
+   * Runs one sweep cycle. No-ops (returns null, touches nothing) when
+   * `platformConfig.wallet.sweep` is absent - an operator who hasn't configured sweep
+   * policy still gets a scheduled job, it just never claims a run. Also returns null
+   * when another cycle already owns the claim.
+   */
+  async runCycle(
+    requestedRunId?: Uuid,
+  ): Promise<{ runId: Uuid; summary: SweepCycleSummary } | null> {
+    const sweepConfig = this.platformConfig?.wallet?.sweep;
+    if (!sweepConfig) {
+      return null;
+    }
+
+    const claim = await this.claimRun(requestedRunId, sweepConfig.staleRunAfterMinutes);
+    if (!claim) {
+      return null;
+    }
+    const { runId, jobRunId } = claim;
+
+    const summary: SweepCycleSummary = {
+      considered: 0,
+      swept: 0,
+      skippedDust: 0,
+      skippedFee: 0,
+      skippedCeiling: 0,
+      inFlight: 0,
+      failed: 0,
+      unknown: 0,
+    };
+
+    // Everything past the claim runs guarded: an adapter that throws (a vendor 5xx is
+    // the ordinary case) would otherwise leave this run's row with finishedAt NULL, and
+    // the partial unique index would then block every subsequent cycle until the
+    // staleness takeover window elapsed. A failed cycle must free its own slot.
+    try {
+      await this.resolveInFlightSweeps(sweepConfig.concurrency);
+
+      const poolBalanceCache = new Map<string, string>();
+      for (const providerName of this.paymentProviders.names()) {
+        const adapter = this.paymentProviders.get(providerName)?.adapter;
+        // sweepToPool is checked here, not after a claim row exists: an adapter that
+        // advertises listSweepableBalances without it would otherwise claim a container,
+        // throw, and park an `unknown` row that holds the in-flight guard forever.
+        if (!adapter?.listSweepableBalances) {
+          continue;
+        }
+        if (!adapter.sweepToPool) {
+          logger.error(
+            { providerName },
+            'provider lists sweepable balances but cannot sweep them; skipping',
+          );
+          continue;
+        }
+        // listSweepableBalances is unbounded; sweeping is throughput work, so the cron
+        // cadence drains the backlog rather than one cycle chasing it all.
+        const balances = (await adapter.listSweepableBalances()).slice(0, sweepConfig.batchSize);
+        summary.considered += balances.length;
+        await mapConcurrent(balances, sweepConfig.concurrency, (balance) =>
+          this.processBalance({
+            providerName,
+            adapter,
+            balance,
+            runId,
+            sweepConfig,
+            summary,
+            poolBalanceCache,
+          }),
+        );
+      }
+    } catch (err) {
+      await this.finishRun(jobRunId, runId, summary, 'failed', err);
+      throw err;
+    }
+
+    await this.finishRun(jobRunId, runId, summary);
+    return { runId, summary };
+  }
+
+  // The insert IS the claim: the partial unique index on wallet_job_run(job_name) WHERE
+  // finished_at IS NULL means a conflicting insert can only mean another cycle already
+  // owns it. A run whose startedAt outlives `staleAfterMinutes` is presumed crashed and
+  // taken over: marked abandoned, then re-claimed inside the same transaction.
+  private async claimRun(
+    requestedRunId: Uuid | undefined,
+    staleAfterMinutes: number,
+  ): Promise<{ runId: Uuid; jobRunId: WalletJobRun['id'] } | null> {
+    const runId = requestedRunId ?? randomUUID();
+    return this.drizzle.db.transaction(async (txn) => {
+      const [inserted] = await txn
+        .insert(walletJobRun)
+        .values({ jobName: CUSTODY_SWEEP_JOB_NAME, runId })
+        .onConflictDoNothing()
+        .returning();
+      if (inserted) {
+        return { runId, jobRunId: inserted.id };
+      }
+
+      const [existing] = await txn
+        .select()
+        .from(walletJobRun)
+        .where(
+          and(eq(walletJobRun.jobName, CUSTODY_SWEEP_JOB_NAME), isNull(walletJobRun.finishedAt)),
+        )
+        .for('update');
+      if (!existing) {
+        // The owning run finished between our failed insert and this read - let the
+        // next tick claim cleanly instead of racing a second insert here.
+        return null;
+      }
+      const staleMs = staleAfterMinutes * 60_000;
+      if (Date.now() - existing.startedAt.getTime() < staleMs) {
+        return null;
+      }
+      await txn
+        .update(walletJobRun)
+        .set({ status: 'abandoned', finishedAt: new Date() })
+        .where(eq(walletJobRun.id, existing.id));
+      const [takenOver] = await txn
+        .insert(walletJobRun)
+        .values({ jobName: CUSTODY_SWEEP_JOB_NAME, runId })
+        .returning();
+      if (!takenOver) {
+        throw new Error('wallet-custody-sweep: failed to claim run after abandoning a stale one');
+      }
+      return { runId, jobRunId: takenOver.id };
+    });
+  }
+
+  // Reusing getWithdrawalStatus for a sweep is deliberate: its {status, txHash} shape
+  // is generic to any vendor transaction (a withdrawal or a sweep), and a sweep's
+  // externalId is exactly the reference such a lookup needs.
+  private async resolveInFlightSweeps(concurrency: number): Promise<void> {
+    const rows = await this.drizzle.db
+      .select()
+      .from(walletCustodySweep)
+      .where(
+        and(
+          inArray(walletCustodySweep.status, IN_FLIGHT_STATUSES),
+          isNotNull(walletCustodySweep.externalId),
+        ),
+      );
+    await mapConcurrent(rows, concurrency, async (row) => {
+      const adapter = this.paymentProviders.get(row.providerName)?.adapter;
+      if (!adapter?.getWithdrawalStatus || !row.externalId) {
+        return;
+      }
+      const result = await adapter.getWithdrawalStatus(row.externalId);
+      if (!result || result.status === 'processing') {
+        return;
+      }
+      await this.drizzle.db
+        .update(walletCustodySweep)
+        .set({ status: result.status, txHash: result.txHash ?? row.txHash })
+        .where(eq(walletCustodySweep.id, row.id));
+    });
+  }
+
+  private async assetFor(currency: string, network: string): Promise<WalletAssetRow | null> {
+    const [row] = await this.drizzle.db
+      .select()
+      .from(walletAsset)
+      .where(and(eq(walletAsset.currency, currency), eq(walletAsset.network, network)));
+    return row ?? null;
+  }
+
+  private async hasInFlightSweep(
+    userId: User['id'],
+    currency: string,
+    network: string,
+  ): Promise<boolean> {
+    const [row] = await this.drizzle.db
+      .select({ id: walletCustodySweep.id })
+      .from(walletCustodySweep)
+      .where(
+        and(
+          eq(walletCustodySweep.userId, userId),
+          eq(walletCustodySweep.currency, currency),
+          eq(walletCustodySweep.network, network),
+          inArray(walletCustodySweep.status, IN_FLIGHT_STATUSES),
+        ),
+      );
+    return row !== undefined;
+  }
+
+  private async getPoolBalanceCached(
+    cache: Map<string, string>,
+    adapter: PaymentAdapter,
+    currency: string,
+    network: string,
+  ): Promise<string | null> {
+    if (!adapter.getPoolBalance) {
+      return null;
+    }
+    const key = `${currency}:${network}`;
+    const cached = cache.get(key);
+    if (cached !== undefined) {
+      return cached;
+    }
+    const balance = await adapter.getPoolBalance(currency, network);
+    cache.set(key, balance);
+    return balance;
+  }
+
+  private async processBalance({
+    providerName,
+    adapter,
+    balance,
+    runId,
+    sweepConfig,
+    summary,
+    poolBalanceCache,
+  }: {
+    providerName: string;
+    adapter: PaymentAdapter;
+    balance: CustodyBalance;
+    runId: Uuid;
+    sweepConfig: SweepConfig;
+    summary: SweepCycleSummary;
+    poolBalanceCache: Map<string, string>;
+  }): Promise<void> {
+    const asset = await this.assetFor(balance.currency, balance.network);
+    if (!asset) {
+      // Funds sitting in a container the operator has no policy for must be visible,
+      // not a log line.
+      await this.drizzle.db
+        .insert(walletReconciliationFinding)
+        .values({
+          runId,
+          providerName,
+          kind: 'unconfigured_asset',
+          currency: balance.currency,
+          network: balance.network,
+          amount: balance.amount,
+          // The dedup key for the partial unique index on (kind, externalId). This
+          // finding has no vendor reference, and the condition recurs every single
+          // cycle until an operator configures the asset - without a stable stand-in
+          // one missing catalog row files a finding every cron tick, drowning the real
+          // findings and pinning the alert threshold permanently over the line. The
+          // asset, not the player, is the subject: one row to act on, not one per
+          // affected container.
+          externalId: `unconfigured:${providerName}:${balance.currency}:${balance.network}`,
+          detail: `no wallet_asset configured for ${balance.currency}/${balance.network}`,
+        })
+        .onConflictDoNothing();
+      return;
+    }
+
+    const gateArgs = {
+      amount: balance.amount,
+      estimatedFee: balance.estimatedFee,
+      minDeposit: asset.minDeposit,
+      feeMultiple: sweepConfig.feeMultiple,
+      sweepFeeCeiling: asset.sweepFeeCeiling,
+      poolLiquidityFloor: asset.poolLiquidityFloor,
+    };
+    let decision = gateSweepBalance({ ...gateArgs, poolBalance: null });
+    // Only fetch the pool balance when the ceiling actually blocks - it's an extra
+    // vendor call, not a free one.
+    if (decision === 'ceiling' && asset.poolLiquidityFloor) {
+      const poolBalance = await this.getPoolBalanceCached(
+        poolBalanceCache,
+        adapter,
+        balance.currency,
+        balance.network,
+      );
+      decision = gateSweepBalance({ ...gateArgs, poolBalance });
+    }
+
+    if (decision === 'dust') {
+      summary.skippedDust += 1;
+      return;
+    }
+    if (decision === 'fee') {
+      summary.skippedFee += 1;
+      return;
+    }
+    if (decision === 'ceiling') {
+      summary.skippedCeiling += 1;
+      return;
+    }
+
+    if (await this.hasInFlightSweep(balance.userId, balance.currency, balance.network)) {
+      summary.inFlight += 1;
+      return;
+    }
+
+    const [claimed] = await this.drizzle.db
+      .insert(walletCustodySweep)
+      .values({
+        userId: balance.userId,
+        providerName,
+        currency: balance.currency,
+        network: balance.network,
+        amount: balance.amount,
+        estimatedFee: balance.estimatedFee,
+        status: 'pending',
+        runId,
+      })
+      .onConflictDoNothing()
+      .returning();
+    if (!claimed) {
+      // Another cycle claimed this (userId, currency, network) between the check above
+      // and this insert - the partial unique index is the real guard, this is a race.
+      summary.inFlight += 1;
+      return;
+    }
+
+    // Deliberately outside any transaction - sweepToPool is a real vendor call
+    // (docs/standards/money.md: treat external payment calls as non-transactional).
+    // The claim row above already holds the durable guard.
+    try {
+      const result = await adapter.sweepToPool?.(balance, { idempotencyKey: claimed.id });
+      if (!result) {
+        throw new Error(
+          `provider "${providerName}" advertised listSweepableBalances but has no sweepToPool`,
+        );
+      }
+      await this.drizzle.db
+        .update(walletCustodySweep)
+        // poolRef records WHICH pool received the funds, not just that a transfer
+        // happened. Player funds must stay separate from operator funds, and that is
+        // the question a regulator asks; without it the ledger cannot answer it.
+        .set({
+          status: 'processing',
+          externalId: result.externalId,
+          poolRef: result.poolRef ?? null,
+        })
+        .where(eq(walletCustodySweep.id, claimed.id));
+      summary.swept += 1;
+    } catch (err) {
+      // A thrown sweepToPool cannot distinguish "the vendor never saw it" from "the
+      // vendor accepted it and the response was lost" - moving to `failed` on the
+      // second case would let the next cycle sweep the same container again and
+      // double-transfer real custody funds. `unknown` keeps the in-flight guard held
+      // (see IN_FLIGHT_STATUSES) until reconciliation or an operator resolves it.
+      logger.warn({ err, sweepId: claimed.id }, 'sweepToPool threw; parking sweep as unknown');
+      await this.drizzle.db
+        .update(walletCustodySweep)
+        .set({ status: 'unknown' })
+        .where(eq(walletCustodySweep.id, claimed.id));
+      summary.unknown += 1;
+    }
+  }
+
+  private async finishRun(
+    jobRunId: WalletJobRun['id'],
+    runId: Uuid,
+    summary: SweepCycleSummary,
+    status: 'completed' | 'failed' = 'completed',
+    err?: unknown,
+  ): Promise<void> {
+    const error =
+      err === undefined ? {} : { error: err instanceof Error ? err.message : String(err) };
+    await this.drizzle.db.transaction(async (txn) => {
+      await txn
+        .update(walletJobRun)
+        .set({ finishedAt: new Date(), status, summary: { ...summary, ...error } })
+        .where(eq(walletJobRun.id, jobRunId));
+      // One entry for the whole cycle, never one per swept balance -
+      // AuditService.recordInTransaction takes a single global advisory lock (the audit
+      // log is a hash chain), so one row per container at thousands of containers would
+      // serialize the entire platform's audit stream behind this job. wallet_custody_sweep
+      // rows are the per-move record; this is the cycle record, joined on runId.
+      await this.audit.recordInTransaction(txn, {
+        actorType: 'system',
+        action: 'wallet.custody.sweep_cycle',
+        resourceType: 'wallet_job_run',
+        resourceId: runId,
+        after: { runId, status, ...summary, ...error },
+      });
+    });
+  }
+}