diff --git a/.changeset/email-outbox-boot-sweep.md b/.changeset/email-outbox-boot-sweep.md
new file mode 100644
index 0000000000..d27e0af621
--- /dev/null
+++ b/.changeset/email-outbox-boot-sweep.md
@@ -0,0 +1,50 @@
+---
+"@objectstack/plugin-email": minor
+---
+
+fix(plugin-email): `sys_email` rows stranded at `queued` are swept at boot, and a failed drain says so at `error` (#5161)
+
+`status: 'queued'` had exactly one consumer: the `afterInsert` outbox drain that
+fires during the insert itself (plus, since #5160, the `email.send.async` job
+`send()` publishes). Nothing ever looked at such a row again. A process that
+died between the insert and the delivery — or a drain whose delivery threw —
+left the row at `queued` **forever**: a state named after a queue that had no
+reader, while the caller had already been told the message was accepted.
+
+**A once-per-boot sweep is now that reader.** At `kernel:ready`, after the
+registries are settled and the `email.send.async` subscriber is attached,
+`sweepStrandedOutbox` picks up `sys_email` rows still at `queued` and advances
+them:
+
+- **durable queue delivery on** → the row is published as an `{ rowId }` job to
+  `email.send.async` through the same producer, options and
+  `sys_email:<id>` idempotency key `send()` uses, so a row that still has a
+  pending job collapses onto it instead of putting a second worker on it;
+- **inline delivery** → the row is delivered and finalized in place (`sent` /
+  `failed`), which is what the drain hook would have done had the process lived.
+
+Only rows **older than five minutes** are eligible. A row inserted seconds ago
+is not stranded, it is someone's in-flight work — this process's `send()`, its
+deferred drain hook, or the same on another instance — and sweeping it would
+send that message twice. (Age, not "created before this boot": one instance's
+boot time says nothing about a sibling's row inserted a second ago.) Rows this
+process is delivering right now, and rows that already carry a `message_id`, are
+skipped. The batch is bounded at 500 rows per boot, oldest first, and says so
+when it truncates. One `info` line reports the counts; boot does **not** wait on
+the sweep, and a sweep that cannot run reports at `error` rather than relying on
+`kernel:ready` error propagation.
+
+**Drain-hook failures are now `error`, not `warn`.** A drain that throws means
+the mail was not sent while the insert reported success and the row still reads
+`queued` — the durability class the degradation-log-level rule pins at `error`.
+Both lines now name the consequence (this message was NOT sent, the row stays at
+`queued`) and the fix (the boot sweep picks it up on the next restart; turn on
+durable queue delivery to have failures retried and dead-lettered instead).
+`deliverPersistedRow` joins `DURABILITY_CRITICAL_CALLEES`, so a future `catch`
+that quietly downgrades it fails `pnpm check:durability-log-level`.
+
+New exports: `sweepStrandedOutbox`, `OUTBOX_OBJECT`, `OUTBOX_SWEEP_MIN_AGE_MS`,
+`OUTBOX_SWEEP_LIMIT`, `EmailService.enqueuePersistedRow`, and
+`EmailServicePlugin.outboxSweepSettled` (the sweep's promise, for callers that
+need determinism). The normal `send()` → deliver path is byte-for-byte
+unchanged.
diff --git a/packages/plugins/plugin-email/src/email-plugin.outbox-sweep.test.ts b/packages/plugins/plugin-email/src/email-plugin.outbox-sweep.test.ts
new file mode 100644
index 0000000000..df9ea011d5
--- /dev/null
+++ b/packages/plugins/plugin-email/src/email-plugin.outbox-sweep.test.ts
@@ -0,0 +1,366 @@
+// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license.
+//
+// EmailServicePlugin — the boot sweep for stranded `sys_email` rows (#5161).
+//
+// The crash these reproduce is the one nothing could recover from: a row is
+// INSERTED at `status:'queued'` and the process dies before the delivery. The
+// row is not synthetic — it is byte-identical to what `send()` / an app's
+// `api.write` leaves behind, and it is written straight into the table so no
+// afterInsert hook fires, which is exactly what "the process died" looks like
+// from the next boot's point of view.
+//
+// Queue mode runs the REAL DbQueueAdapter, so what is asserted is the whole
+// round trip: sweep → job → worker poll → the same row at `sent`.
+
+import { describe, it, expect, vi, beforeEach } from 'vitest';
+import { assertEngineDeleteDispatch } from '@objectstack/objectql';
+import { DbQueueAdapter } from '@objectstack/service-queue';
+import { EmailServicePlugin } from './email-plugin.js';
+import { EmailService, EMAIL_SEND_QUEUE } from './email-service.js';
+
+// ── harness ────────────────────────────────────────────────────────────────
+
+const NOW = Date.UTC(2026, 7, 4, 12, 0, 0);
+const ago = (ms: number) => new Date(NOW - ms).toISOString();
+const min = (n: number) => n * 60_000;
+
+/** Let the drain hook's `setTimeout(0)` and its async chain run. */
+const flush = () => new Promise((r) => setTimeout(r, 5));
+
+interface EngineOpts {
+  /** Make `find` throw for the drain hook's own `where: { id }` re-read. */
+  failIdLookup?: boolean;
+}
+
+/**
+ * ObjectQL-shaped engine: `where` (including the `$lt` the age gate uses),
+ * `orderBy`, `limit`, and the afterInsert hook registry the outbox drain
+ * installs itself into.
+ */
+function fakeEngine(opts: EngineOpts = {}) {
+  const tables = new Map<string, any[]>();
+  const hooks: Record<string, Array<(ctx: any) => any>> = {};
+  const rowsOf = (t: string) => tables.get(t) ?? [];
+  const matches = (row: any, where: Record<string, any>) =>
+    Object.entries(where).every(([k, v]) => {
+      if (v && typeof v === 'object' && !Array.isArray(v)) {
+        return Object.entries(v).every(([op, target]) => {
+          if (op === '$lt') return row[k] < (target as any);
+          throw new Error(`fakeEngine: unsupported operator ${op}`);
+        });
+      }
+      return row[k] === v;
+    });
+  const engine = {
+    tables,
+    rows: (t: string) => [...rowsOf(t)],
+    /** Seed a row WITHOUT firing hooks — "the process died after the insert". */
+    seed(table: string, row: Record<string, any>) {
+      const t = rowsOf(table);
+      t.push({ ...row });
+      tables.set(table, t);
+    },
+    registerHook(event: string, fn: (ctx: any) => any, _meta?: unknown) {
+      (hooks[event] ??= []).push(fn);
+    },
+    unregisterHooksByPackage(_pkg: string) { /* single-boot harness */ },
+    /** The registered afterInsert handlers, for poisoning them directly. */
+    afterInsertHooks: () => hooks.afterInsert ?? [],
+    async find(table: string, o: any = {}) {
+      if (opts.failIdLookup && table === 'sys_email' && typeof o?.where?.id === 'string') {
+        throw new Error('datasource connection lost');
+      }
+      let out = o.where ? rowsOf(table).filter((r) => matches(r, o.where)) : [...rowsOf(table)];
+      for (const ord of [...(o.orderBy ?? [])].reverse()) {
+        out.sort((a, b) => {
+          const av = a[ord.field], bv = b[ord.field];
+          if (av === bv) return 0;
+          return (av > bv ? 1 : -1) * (ord.order === 'desc' ? -1 : 1);
+        });
+      }
+      if (o.offset) out = out.slice(o.offset);
+      if (o.limit) out = out.slice(0, o.limit);
+      return out;
+    },
+    async insert(table: string, data: any) {
+      const row = { ...data };
+      const t = rowsOf(table);
+      t.push(row);
+      tables.set(table, t);
+      for (const fn of hooks.afterInsert ?? []) await fn({ object: table, result: row });
+      return { id: data.id };
+    },
+    async update(table: string, patch: any) {
+      const r = rowsOf(table).find((x) => x.id === patch.id);
+      if (!r) throw new Error(`row ${patch.id} not found in ${table}`);
+      Object.assign(r, patch);
+      return r;
+    },
+    async delete(table: string, o: any) {
+      // [#4550] Pinned to ObjectQL.delete's OWN dispatch predicate, like the
+      // other doubles in this package. A fake looser than the engine it stands
+      // in for is how #4434 shipped a dead REST route with its suite green, and
+      // the case a hand-written mirror always drops is the scalar test
+      // (`where: { id: { $in: [...] } }` looks like an id and is not one).
+      const dispatch = assertEngineDeleteDispatch(o);
+      if (dispatch.kind === 'multi') {
+        const survivors = rowsOf(table).filter((r) => !matches(r, o?.where ?? {}));
+        const deleted = rowsOf(table).length - survivors.length;
+        tables.set(table, survivors);
+        return { deleted };
+      }
+      tables.set(table, rowsOf(table).filter((r) => r.id !== dispatch.id));
+      return { id: dispatch.id };
+    },
+  };
+  return engine;
+}
+
+function fakeCtx(services: Record<string, unknown>) {
+  const handlers: Record<string, Array<() => Promise<void> | void>> = {};
+  const logger = { info: vi.fn(), warn: vi.fn(), error: vi.fn(), debug: vi.fn() };
+  return {
+    logger,
+    getService: <T,>(name: string): T => {
+      if (!(name in services)) throw new Error(`service '${name}' not registered`);
+      return services[name] as T;
+    },
+    registerService: (name: string, svc: unknown) => { services[name] = svc; },
+    hook: (name: string, fn: () => Promise<void> | void) => { (handlers[name] ??= []).push(fn); },
+    fire: async (name: string) => { for (const fn of handlers[name] ?? []) await fn(); },
+  };
+}
+
+function fakeClock(startMs = NOW) {
+  let t = startMs;
+  return { now: () => new Date(t), advance: (ms: number) => { t += ms; } };
+}
+
+interface BootOpts {
+  queue?: boolean;
+  plugin?: Record<string, unknown>;
+  transport?: { send: (m: any) => Promise<any> };
+  engine?: EngineOpts;
+}
+
+async function boot(opts: BootOpts = {}) {
+  const engine = fakeEngine(opts.engine);
+  const clock = fakeClock();
+  const transport = opts.transport ?? { send: vi.fn(async () => ({ messageId: '<sent@x>' })) };
+  const adapter = new DbQueueAdapter({
+    engine: engine as never,
+    clock,
+    options: { autoStart: false, pollIntervalMs: 60_000, defaultMaxAttempts: 3 },
+  });
+  const services: Record<string, unknown> = { manifest: { register: () => {} }, objectql: engine };
+  if (opts.queue) services.queue = adapter;
+
+  const ctx = fakeCtx(services);
+  const plugin = new EmailServicePlugin({ seedTemplates: false, transport, ...(opts.plugin ?? {}) });
+  await plugin.init(ctx as never);
+  await plugin.start(ctx as never);
+
+  /** A row `send()` (or an app write) committed, then the process died. */
+  const strand = (id: string, createdAt: string) => engine.seed('sys_email', {
+    id,
+    from_address: 'no-reply@example.test',
+    to_addresses: 'user@example.test',
+    subject: `Stranded ${id}`,
+    body_text: 'hello',
+    status: 'queued',
+    attempt_count: 0,
+    created_at: createdAt,
+  });
+
+  return {
+    plugin, ctx, engine, adapter, clock, transport, strand,
+    service: () => services.email as EmailService,
+    sysEmail: () => engine.rows('sys_email'),
+    jobs: () => engine.rows('sys_job_queue'),
+    /** Boot, then wait for the (deliberately un-awaited) sweep to settle. */
+    async ready() {
+      await ctx.fire('kernel:ready');
+      return plugin.outboxSweepSettled;
+    },
+  };
+}
+
+const errorLines = (ctx: { logger: { error: ReturnType<typeof vi.fn> } }) =>
+  ctx.logger.error.mock.calls.map((c) => String(c[0]));
+const infoLines = (ctx: { logger: { info: ReturnType<typeof vi.fn> } }) =>
+  ctx.logger.info.mock.calls.map((c) => String(c[0]));
+
+beforeEach(() => { vi.clearAllMocks(); vi.setSystemTime(new Date(NOW)); });
+
+// ── inline mode ────────────────────────────────────────────────────────────
+
+describe('boot sweep — inline delivery', () => {
+  it('advances a row stranded by a crash to sent, instead of leaving it at queued forever', async () => {
+    const h = await boot();
+    h.strand('row-crashed', ago(min(30)));
+
+    const swept = await h.ready();
+
+    expect(h.transport.send).toHaveBeenCalledTimes(1);
+    expect(h.sysEmail()[0]).toMatchObject({
+      id: 'row-crashed', status: 'sent', message_id: '<sent@x>', attempt_count: 1,
+    });
+    expect(swept).toMatchObject({ scanned: 1, sent: 1, failed: 0 });
+    expect(infoLines(h.ctx).join('\n')).toMatch(/outbox sweep — 1 sys_email row\(s\) stranded/);
+  });
+
+  it('finalizes an unsendable row as failed rather than leaving it stranded', async () => {
+    const h = await boot({ transport: { send: vi.fn(async () => { throw new Error('535 auth failed'); }) } });
+    h.strand('row-bad', ago(min(30)));
+
+    const swept = await h.ready();
+
+    expect(h.sysEmail()[0]).toMatchObject({ id: 'row-bad', status: 'failed', attempt_count: 1 });
+    expect(String(h.sysEmail()[0].error)).toMatch(/535 auth failed/);
+    expect(swept).toMatchObject({ failed: 1 });
+    expect(errorLines(h.ctx).join('\n')).toMatch(/never reached a recipient/);
+  });
+
+  it('does not touch a row that was inserted seconds ago', async () => {
+    // Another instance is delivering it right now; a boot must not race it.
+    const h = await boot();
+    h.strand('row-fresh', ago(3_000));
+
+    const swept = await h.ready();
+
+    expect(h.transport.send).not.toHaveBeenCalled();
+    expect(h.sysEmail()[0]).toMatchObject({ status: 'queued' });
+    expect(swept).toMatchObject({ scanned: 0 });
+  });
+});
+
+// ── queue mode ─────────────────────────────────────────────────────────────
+
+describe('boot sweep — durable queue delivery', () => {
+  it('re-queues the stranded row as an { rowId } job that a worker then delivers', async () => {
+    const h = await boot({ queue: true, plugin: { queueDelivery: true } });
+    h.strand('row-crashed', ago(min(30)));
+
+    const swept = await h.ready();
+
+    // The sweep publishes; it does NOT deliver in the boot path.
+    expect(swept).toMatchObject({ scanned: 1, requeued: 1, sent: 0 });
+    expect(h.transport.send).not.toHaveBeenCalled();
+    expect(h.jobs()).toHaveLength(1);
+    expect(h.jobs()[0]).toMatchObject({ queue: EMAIL_SEND_QUEUE, status: 'pending' });
+    // Same payload shape as send()'s own publish — one producer, not two.
+    expect(JSON.parse(h.jobs()[0].payload_json)).toEqual({ rowId: 'row-crashed' });
+    expect(h.jobs()[0].idempotency_key).toBe('sys_email:row-crashed');
+
+    await h.adapter.pollOnce();
+
+    expect(h.sysEmail()).toHaveLength(1);              // still ONE row
+    expect(h.sysEmail()[0]).toMatchObject({ id: 'row-crashed', status: 'sent', attempt_count: 1 });
+    expect(h.jobs()[0]).toMatchObject({ status: 'completed' });
+  });
+
+  it('collapses onto an existing pending job instead of racing a second worker', async () => {
+    // The other half of a crash: the row AND its job survived (the process died
+    // between publishing and the worker running). Re-publishing a second job
+    // would put two workers on one row.
+    const h = await boot({ queue: true, plugin: { queueDelivery: true } });
+    h.strand('row-published', ago(min(30)));
+    await h.adapter.publish(EMAIL_SEND_QUEUE, { rowId: 'row-published' }, {
+      idempotencyKey: 'sys_email:row-published',
+    });
+
+    await h.ready();
+
+    expect(h.jobs()).toHaveLength(1);
+
+    await h.adapter.pollOnce();
+    expect(h.transport.send).toHaveBeenCalledTimes(1);
+    expect(h.sysEmail()[0]).toMatchObject({ status: 'sent' });
+  });
+});
+
+// ── the drain hook's own failures ──────────────────────────────────────────
+
+describe('drain-hook failures are reported at error, with consequence and fix', () => {
+  it('a delivery that throws is an error, not a warn — the mail was not sent', async () => {
+    const h = await boot({ engine: { failIdLookup: true } });
+    await h.ready();
+
+    // An app-inserted outbox row: the hook fires, its re-read explodes.
+    await h.engine.insert('sys_email', {
+      id: 'row-app', from_address: 'a@b.test', to_addresses: 'c@d.test',
+      subject: 'App', body_text: 'x', status: 'queued', created_at: new Date(NOW).toISOString(),
+    });
+    await flush();
+
+    const line = errorLines(h.ctx).find((l) => l.includes('outbox drain FAILED'));
+    expect(line, 'drain failure must be reported at error level').toBeTruthy();
+    expect(h.ctx.logger.warn.mock.calls.map((c) => String(c[0])).join('\n')).not.toMatch(/outbox drain/);
+    expect(line!).toMatch(/row 'row-app'/);
+    expect(line!).toMatch(/was NOT sent/);                 // consequence …
+    expect(line!).toMatch(/stays at `queued`/);            // … concretely
+    expect(line!).toMatch(/boot outbox sweep/);            // fix: what picks it up
+    expect(line!).toMatch(/Durable queue delivery/);       // fix: how to stop waiting
+    expect(line!).toMatch(/datasource connection lost/);   // cause
+  });
+
+  it('a hook that breaks before it can even schedule delivery is an error too', async () => {
+    const h = await boot();
+    await h.ready();
+
+    // A row object whose property access throws — the synchronous arm of the
+    // hook, which used to swallow into a warn nobody reads.
+    const poison = new Proxy({}, { get() { throw new Error('row proxy exploded'); } });
+    for (const fn of h.engine.afterInsertHooks()) await fn(poison);
+
+    const line = errorLines(h.ctx).find((l) => l.includes('outbox drain hook error'));
+    expect(line, 'hook breakage must be reported at error level').toBeTruthy();
+    expect(line!).toMatch(/never scheduled for delivery/);
+    expect(line!).toMatch(/boot outbox sweep/);
+    expect(line!).toMatch(/row proxy exploded/);
+  });
+
+  it('reports at error when the sweep itself cannot run', async () => {
+    const h = await boot();
+    h.engine.find = (async () => { throw new Error('no such table: sys_email'); }) as never;
+
+    const swept = await h.ready();
+
+    expect(swept).toBeUndefined();
+    const line = errorLines(h.ctx).find((l) => l.includes('boot sweep of stranded sys_email rows FAILED'));
+    expect(line, 'a swallowed sweep failure is the durability loss itself').toBeTruthy();
+    expect(line!).toMatch(/still sitting at `queued`/);
+    expect(line!).toMatch(/no such table/);
+  });
+});
+
+// ── the default path is untouched ──────────────────────────────────────────
+
+describe('normal delivery is unchanged', () => {
+  it('send() still delivers inline and the sweep finds nothing to do', async () => {
+    const h = await boot();
+    const swept = await h.ready();
+
+    const res = await h.service().send({ to: 'a@b.com', from: 'x@y.com', subject: 'Hi', text: 'hello' });
+
+    expect(res).toMatchObject({ status: 'sent', messageId: '<sent@x>' });
+    expect(swept).toMatchObject({ scanned: 0 });
+    expect(h.transport.send).toHaveBeenCalledTimes(1);
+  });
+
+  it('an app-inserted row is still drained by the afterInsert hook, exactly once', async () => {
+    const h = await boot();
+    await h.ready();
+
+    await h.engine.insert('sys_email', {
+      id: 'row-app', from_address: 'a@b.test', to_addresses: 'c@d.test',
+      subject: 'App', body_text: 'x', status: 'queued', attempt_count: 0,
+      created_at: new Date(NOW).toISOString(),
+    });
+    await flush();
+
+    expect(h.transport.send).toHaveBeenCalledTimes(1);
+    expect(h.sysEmail()[0]).toMatchObject({ id: 'row-app', status: 'sent', message_id: '<sent@x>' });
+    expect(h.ctx.logger.error).not.toHaveBeenCalled();
+  });
+});
diff --git a/packages/plugins/plugin-email/src/email-plugin.ts b/packages/plugins/plugin-email/src/email-plugin.ts
index 89ab1c8362..a819d06a24 100644
--- a/packages/plugins/plugin-email/src/email-plugin.ts
+++ b/packages/plugins/plugin-email/src/email-plugin.ts
@@ -39,6 +39,7 @@ import {
   bindEmailTemplateProvenanceStamp,
   unbindEmailTemplateProvenanceStamp,
 } from './email-template-provenance.js';
+import { sweepStrandedOutbox, type OutboxSweepResult } from './outbox-sweep.js';
 
 const SYSTEM_CTX = { isSystem: true, positions: [], permissions: [] } as const;
 
@@ -170,6 +171,14 @@ export class EmailServicePlugin implements Plugin {
    * `applyMailSettings` already applies to `provider`.
    */
   private queueDeliveryFromSettings?: boolean;
+  /**
+   * Settles when the boot outbox sweep (#5161) has finished — with its counts,
+   * or `undefined` when the sweep itself failed (already reported at `error`).
+   *
+   * Boot does NOT await it, so this is how anything that needs determinism
+   * (tests, an operator script) observes the sweep instead of guessing.
+   */
+  outboxSweepSettled?: Promise<OutboxSweepResult | undefined>;
 
   constructor(options: EmailServicePluginOptions = {}) {
     this.options = options;
@@ -514,12 +523,29 @@ export class EmailServicePlugin implements Plugin {
                     if (target.status !== 'queued' || target.message_id) return;
                     await svc.deliverPersistedRow(target);
                   } catch (err: any) {
-                    ctx.logger.warn(`EmailServicePlugin: outbox drain failed for ${rowId}: ${err?.message ?? err}`);
+                    // `error`, not `warn` (#5161): the insert returned, the row
+                    // is there, everything looks normal — and the mail was NOT
+                    // sent. Nothing else in this process will look at that row
+                    // again, so a line nobody reads is the whole loss.
+                    ctx.logger.error(
+                      `EmailServicePlugin: outbox drain FAILED for sys_email row '${rowId}' — that message was `
+                      + 'NOT sent and the row stays at `queued`; nothing retries it in this process. Fix: it is '
+                      + 'picked up by the boot outbox sweep on the next restart; to have failures retried and '
+                      + 'dead-lettered instead of waiting for one, turn on Settings → Mail → "Durable queue '
+                      + 'delivery" (@objectstack/service-queue over an ObjectQL engine). '
+                      + `Cause: ${err?.message ?? err}`,
+                    );
                   }
                 })();
               }, 0);
             } catch (err: any) {
-              ctx.logger.warn(`EmailServicePlugin: outbox drain hook error: ${err?.message ?? err}`);
+              // Same class one level up: the row was inserted and never even
+              // scheduled for delivery.
+              ctx.logger.error(
+                'EmailServicePlugin: outbox drain hook error — an inserted sys_email row was never scheduled '
+                + 'for delivery and stays at `queued`, undelivered, while the insert reported success. Fix: the '
+                + `boot outbox sweep picks it up on the next restart. Cause: ${err?.message ?? err}`,
+              );
             }
           },
           { packageId: DRAIN_PKG },
@@ -624,6 +650,37 @@ export class EmailServicePlugin implements Plugin {
         }
       }
 
+      // ── STRANDED OUTBOX SWEEP (#5161) ────────────────────────────────
+      // The `queued` rows nobody was consuming: a process that died between
+      // the insert and the delivery left a row named after a queue that had no
+      // reader. Swept HERE — one anchor with the boot gate above, after the
+      // registries are settled and the subscriber is attached, so a re-queued
+      // row has somewhere to land.
+      //
+      // NOT awaited: a backlog of stranded mail must not hold the server off
+      // its port, and in inline mode every row is a transport round trip. And
+      // self-catching rather than trusting the hook: a `kernel:ready` handler
+      // that throws is silently swallowed on LiteKernel (#5170), which for a
+      // durability sweep would mean losing the report of the very failure it
+      // exists to prevent.
+      if (persistence) {
+        const svc = this.service;
+        this.outboxSweepSettled = (async () => {
+          try {
+            return await sweepStrandedOutbox({ engine: engine as any, service: svc, logger: ctx.logger });
+          } catch (err: any) {
+            ctx.logger.error(
+              'EmailServicePlugin: the boot sweep of stranded sys_email rows FAILED to run — messages accepted '
+              + 'before the last restart are still sitting at `queued`, nothing else looks at them, and the '
+              + 'server will keep reporting healthy with that mail undelivered. Fix: make sure sys_email is '
+              + 'readable by the system context (schema sync ran, the datasource is up), then restart to '
+              + `re-sweep. Cause: ${err?.message ?? err}`,
+            );
+            return undefined;
+          }
+        })();
+      }
+
       // Seed built-in + user-provided templates (upsert by name+locale).
       if (this.options.seedTemplates !== false) {
         const all = [
diff --git a/packages/plugins/plugin-email/src/email-service.ts b/packages/plugins/plugin-email/src/email-service.ts
index c026f00872..24624a91a3 100644
--- a/packages/plugins/plugin-email/src/email-service.ts
+++ b/packages/plugins/plugin-email/src/email-service.ts
@@ -544,6 +544,26 @@ export class EmailService implements IEmailService {
     }
   }
 
+  /**
+   * Hand an ALREADY-PERSISTED `sys_email` row to the durable queue — the boot
+   * sweep's producer entry point (#5161). Returns `false` when queue delivery
+   * is not in force (or the publish failed), which is the caller's signal to
+   * deliver the row inline instead.
+   *
+   * Deliberately routed through the same {@link publishRow} as `send()`: ONE
+   * producer of {@link EmailSendQueuePayload}, one set of publish options, one
+   * `sys_email:<id>` idempotency key. A second publisher spelling the payload
+   * its own way is exactly how the two halves of a queue drift apart — and the
+   * shared key is what makes sweeping a row that still has a pending job
+   * collapse onto that job instead of racing a second worker onto the row.
+   */
+  async enqueuePersistedRow(rowId: string): Promise<boolean> {
+    if (!rowId) return false;
+    const queue = this.options.queueDelivery?.resolve();
+    if (!queue) return false;
+    return this.publishRow(queue, rowId);
+  }
+
   /**
    * State a queue-delivery degradation once, at `error`.
    *
diff --git a/packages/plugins/plugin-email/src/index.ts b/packages/plugins/plugin-email/src/index.ts
index 90825c0037..3c0ebf6f38 100644
--- a/packages/plugins/plugin-email/src/index.ts
+++ b/packages/plugins/plugin-email/src/index.ts
@@ -55,6 +55,15 @@ export {
   EMAIL_TEMPLATE_OBJECT,
   type BootstrapDeclaredEmailTemplatesResult,
 } from './bootstrap-declared-email-templates.js';
+export {
+  sweepStrandedOutbox,
+  OUTBOX_OBJECT,
+  OUTBOX_SWEEP_MIN_AGE_MS,
+  OUTBOX_SWEEP_LIMIT,
+  type OutboxSweepResult,
+  type OutboxSweepService,
+  type SweepStrandedOutboxOptions,
+} from './outbox-sweep.js';
 export {
   bindEmailTemplateProvenanceStamp,
   unbindEmailTemplateProvenanceStamp,
diff --git a/packages/plugins/plugin-email/src/outbox-sweep.test.ts b/packages/plugins/plugin-email/src/outbox-sweep.test.ts
new file mode 100644
index 0000000000..a5c3138732
--- /dev/null
+++ b/packages/plugins/plugin-email/src/outbox-sweep.test.ts
@@ -0,0 +1,318 @@
+// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license.
+//
+// sweepStrandedOutbox — the consumer `status:'queued'` never had (#5161).
+//
+// What these pin, in order of what actually costs money when it is wrong:
+//   1. a stranded row is advanced, so `queued` stops being a terminal state;
+//   2. a row that is merely YOUNG is not touched — sweeping live work is how a
+//      restart would send somebody's mail twice;
+//   3. the route follows the configured mode (queue publish vs inline
+//      delivery), and the queue route reuses `send()`'s own producer;
+//   4. the failure paths are loud and bounded.
+
+import { describe, it, expect, vi } from 'vitest';
+import {
+  sweepStrandedOutbox,
+  OUTBOX_SWEEP_MIN_AGE_MS,
+  OUTBOX_OBJECT,
+  type OutboxSweepService,
+} from './outbox-sweep.js';
+
+const NOW = Date.UTC(2026, 7, 4, 12, 0, 0);
+const ago = (ms: number) => new Date(NOW - ms).toISOString();
+
+/** Minutes, in ms — the age gate's unit. */
+const min = (n: number) => n * 60_000;
+
+interface RowSeed {
+  id: string;
+  created_at: string;
+  status?: string;
+  message_id?: string;
+}
+
+/**
+ * Engine mirroring the `where` / `orderBy` / `limit` subset ObjectQL answers,
+ * INCLUDING the `{ $lt: … }` operator the age gate is written in — a double
+ * that ignored the operator would report the gate working while it filtered
+ * nothing.
+ */
+function fakeEngine(seed: RowSeed[]) {
+  const rows = seed.map((r) => ({ status: 'queued', ...r }) as Record<string, any>);
+  const matches = (row: any, where: Record<string, any>) =>
+    Object.entries(where).every(([k, v]) => {
+      if (v && typeof v === 'object' && !Array.isArray(v)) {
+        return Object.entries(v).every(([op, target]) => {
+          if (op === '$lt') return row[k] < (target as any);
+          throw new Error(`fakeEngine: unsupported operator ${op}`);
+        });
+      }
+      return row[k] === v;
+    });
+  const find = vi.fn(async (object: string, opts: any = {}) => {
+    if (object !== OUTBOX_OBJECT) return [];
+    let out = opts.where ? rows.filter((r) => matches(r, opts.where)) : [...rows];
+    for (const ord of [...(opts.orderBy ?? [])].reverse()) {
+      out.sort((a, b) => {
+        const av = a[ord.field], bv = b[ord.field];
+        if (av === bv) return 0;
+        return (av > bv ? 1 : -1) * (ord.order === 'desc' ? -1 : 1);
+      });
+    }
+    if (opts.limit) out = out.slice(0, opts.limit);
+    return out;
+  });
+  return { rows, find };
+}
+
+interface ServiceOpts {
+  /** true ⇒ queue mode (publish succeeds), false ⇒ inline. */
+  enqueue?: boolean | ((rowId: string) => boolean);
+  deliver?: (row: Record<string, any>) => any;
+  managed?: string[];
+}
+
+function fakeService(opts: ServiceOpts = {}) {
+  const managed = new Set(opts.managed ?? []);
+  const enqueuePersistedRow = vi.fn(async (rowId: string) => (
+    typeof opts.enqueue === 'function' ? opts.enqueue(rowId) : opts.enqueue === true
+  ));
+  const deliverPersistedRow = vi.fn(async (row: Record<string, any>) => (
+    opts.deliver ? opts.deliver(row) : { id: String(row.id), status: 'sent', messageId: '<x@y>' }
+  ));
+  return {
+    isServiceManaged: (id: string) => managed.has(id),
+    enqueuePersistedRow,
+    deliverPersistedRow,
+  } as unknown as OutboxSweepService & {
+    enqueuePersistedRow: ReturnType<typeof vi.fn>;
+    deliverPersistedRow: ReturnType<typeof vi.fn>;
+  };
+}
+
+function fakeLogger() {
+  return { info: vi.fn(), warn: vi.fn(), error: vi.fn() };
+}
+
+const lines = (fn: ReturnType<typeof vi.fn>) => fn.mock.calls.map((c) => String(c[0]));
+
+// ── 1. the stranded row is advanced ────────────────────────────────────────
+
+describe('sweepStrandedOutbox — the row nobody was consuming', () => {
+  it('delivers a row left behind by a process that died after the insert', async () => {
+    // The exact scenario: a `queued` row committed, the drain hook never
+    // completed, no queue job exists. Before #5161 nothing ever read it again.
+    const engine = fakeEngine([{ id: 'row-crashed', created_at: ago(min(30)) }]);
+    const service = fakeService();
+    const logger = fakeLogger();
+
+    const res = await sweepStrandedOutbox({ engine, service, logger, now: () => NOW });
+
+    expect(service.deliverPersistedRow).toHaveBeenCalledTimes(1);
+    expect(service.deliverPersistedRow.mock.calls[0][0]).toMatchObject({ id: 'row-crashed' });
+    expect(res).toMatchObject({ scanned: 1, sent: 1, requeued: 0, failed: 0, skipped: 0 });
+  });
+
+  it('queries only queued rows past the age gate, oldest first, bounded', async () => {
+    const engine = fakeEngine([]);
+    await sweepStrandedOutbox({ engine, service: fakeService(), now: () => NOW });
+
+    const [object, opts] = engine.find.mock.calls[0] as [string, any];
+    expect(object).toBe('sys_email');
+    expect(opts.where).toEqual({
+      status: 'queued',
+      created_at: { $lt: new Date(NOW - OUTBOX_SWEEP_MIN_AGE_MS).toISOString() },
+    });
+    expect(opts.orderBy).toEqual([{ field: 'created_at', order: 'asc' }]);
+    expect(opts.limit).toBeGreaterThan(0);
+    expect(opts.context).toMatchObject({ isSystem: true });
+  });
+
+  it('says nothing at all when there is nothing stranded (the healthy boot)', async () => {
+    const logger = fakeLogger();
+    const res = await sweepStrandedOutbox({ engine: fakeEngine([]), service: fakeService(), logger, now: () => NOW });
+    expect(res).toMatchObject({ scanned: 0, sent: 0, requeued: 0 });
+    expect(logger.info).not.toHaveBeenCalled();
+    expect(logger.error).not.toHaveBeenCalled();
+  });
+});
+
+// ── 2. the age gate ────────────────────────────────────────────────────────
+
+describe('the age gate — a young row is somebody else\'s in-flight work', () => {
+  it('leaves a row inserted seconds ago alone, and takes the old one', async () => {
+    // A sibling instance inserted `row-fresh` two seconds ago and its drain
+    // hook is mid-flight. Sweeping it would send that message twice, and this
+    // instance's own boot time cannot tell the two rows apart — only age can.
+    const engine = fakeEngine([
+      { id: 'row-fresh', created_at: ago(2_000) },
+      { id: 'row-old', created_at: ago(min(45)) },
+    ]);
+    const service = fakeService();
+
+    const res = await sweepStrandedOutbox({ engine, service, now: () => NOW });
+
+    expect(service.deliverPersistedRow).toHaveBeenCalledTimes(1);
+    expect(service.deliverPersistedRow.mock.calls[0][0]).toMatchObject({ id: 'row-old' });
+    expect(res.scanned).toBe(1);
+  });
+
+  it('holds a row until it crosses the gate, then sweeps it', async () => {
+    const engine = fakeEngine([{ id: 'row-1', created_at: ago(min(4)) }]);
+    const service = fakeService();
+
+    expect((await sweepStrandedOutbox({ engine, service, now: () => NOW })).scanned).toBe(0);
+    // …the next boot, ten minutes later.
+    const later = await sweepStrandedOutbox({ engine, service, now: () => NOW + min(10) });
+    expect(later).toMatchObject({ scanned: 1, sent: 1 });
+  });
+});
+
+// ── 3. routing + idempotency backstops ─────────────────────────────────────
+
+describe('routing follows the configured delivery mode', () => {
+  it('queue mode: publishes the row through send()\'s own producer, never the transport', async () => {
+    const engine = fakeEngine([
+      { id: 'row-a', created_at: ago(min(30)) },
+      { id: 'row-b', created_at: ago(min(20)) },
+    ]);
+    const service = fakeService({ enqueue: true });
+    const logger = fakeLogger();
+
+    const res = await sweepStrandedOutbox({ engine, service, logger, now: () => NOW });
+
+    expect(service.enqueuePersistedRow.mock.calls.map((c) => c[0])).toEqual(['row-a', 'row-b']);
+    expect(service.deliverPersistedRow).not.toHaveBeenCalled();
+    expect(res).toMatchObject({ scanned: 2, requeued: 2, sent: 0, failed: 0 });
+    expect(lines(logger.info)[0]).toMatch(/2 re-queued for durable delivery/);
+  });
+
+  it('falls back to inline delivery when the publish fails — the row is already committed', async () => {
+    const engine = fakeEngine([{ id: 'row-a', created_at: ago(min(30)) }]);
+    const service = fakeService({ enqueue: false });
+
+    const res = await sweepStrandedOutbox({ engine, service, now: () => NOW });
+
+    expect(service.enqueuePersistedRow).toHaveBeenCalledWith('row-a');
+    expect(service.deliverPersistedRow).toHaveBeenCalledTimes(1);
+    expect(res).toMatchObject({ requeued: 0, sent: 1 });
+  });
+
+  it('never touches a row this process is delivering right now', async () => {
+    const engine = fakeEngine([
+      { id: 'row-managed', created_at: ago(min(30)) },
+      { id: 'row-free', created_at: ago(min(30)) },
+    ]);
+    const service = fakeService({ managed: ['row-managed'] });
+
+    const res = await sweepStrandedOutbox({ engine, service, now: () => NOW });
+
+    expect(service.deliverPersistedRow).toHaveBeenCalledTimes(1);
+    expect(service.deliverPersistedRow.mock.calls[0][0]).toMatchObject({ id: 'row-free' });
+    expect(res).toMatchObject({ scanned: 2, sent: 1, skipped: 1 });
+  });
+
+  it('never re-sends a row the transport already accepted', async () => {
+    // `message_id` set with the status update lost (a crash between the two
+    // writes) — the same guard the email.send.async subscriber applies.
+    const engine = fakeEngine([{ id: 'row-sent', created_at: ago(min(30)), message_id: '<already@x>' }]);
+    const service = fakeService();
+
+    const res = await sweepStrandedOutbox({ engine, service, now: () => NOW });
+
+    expect(service.deliverPersistedRow).not.toHaveBeenCalled();
+    expect(service.enqueuePersistedRow).not.toHaveBeenCalled();
+    expect(res).toMatchObject({ scanned: 1, skipped: 1 });
+  });
+});
+
+// ── 4. failure paths ───────────────────────────────────────────────────────
+
+describe('failures are loud, counted, and never stop the batch', () => {
+  it('records an inline delivery that failed and reports the consequence at error', async () => {
+    const engine = fakeEngine([{ id: 'row-a', created_at: ago(min(30)) }]);
+    const service = fakeService({
+      deliver: () => ({ id: 'row-a', status: 'failed', error: '535 authentication failed' }),
+    });
+    const logger = fakeLogger();
+
+    const res = await sweepStrandedOutbox({ engine, service, logger, now: () => NOW });
+
+    expect(res).toMatchObject({ scanned: 1, failed: 1, sent: 0 });
+    const err = lines(logger.error).join('\n');
+    expect(err).toMatch(/could NOT be delivered/);
+    expect(err).toMatch(/never reached a recipient/);   // consequence
+    expect(err).toMatch(/Durable queue delivery/);      // fix
+  });
+
+  it('keeps going after a row that throws, and says so once', async () => {
+    const engine = fakeEngine([
+      { id: 'row-bad', created_at: ago(min(30)) },
+      { id: 'row-bad-2', created_at: ago(min(29)) },
+      { id: 'row-good', created_at: ago(min(28)) },
+    ]);
+    const service = fakeService({
+      deliver: (row) => {
+        if (String(row.id).startsWith('row-bad')) throw new Error('engine exploded');
+        return { id: row.id, status: 'sent' };
+      },
+    });
+    const logger = fakeLogger();
+
+    const res = await sweepStrandedOutbox({ engine, service, logger, now: () => NOW });
+
+    expect(res).toMatchObject({ scanned: 3, failed: 2, sent: 1 });
+    // One per-row line (say it once), plus the batch summary error.
+    const perRow = lines(logger.error).filter((l) => l.includes("could not advance sys_email row"));
+    expect(perRow).toHaveLength(1);
+    expect(perRow[0]).toMatch(/row-bad/);
+    expect(perRow[0]).toMatch(/engine exploded/);
+  });
+
+  it('propagates a failure of the query itself — the sweep did not happen', async () => {
+    const engine = { find: vi.fn(async () => { throw new Error('no such table: sys_email'); }) };
+    await expect(sweepStrandedOutbox({ engine, service: fakeService(), now: () => NOW }))
+      .rejects.toThrow(/no such table/);
+  });
+
+  it('bounds the batch and says more remain', async () => {
+    const engine = fakeEngine(
+      Array.from({ length: 5 }, (_, i) => ({ id: `row-${i}`, created_at: ago(min(30 + i)) })),
+    );
+    const logger = fakeLogger();
+
+    const res = await sweepStrandedOutbox({ engine, service: fakeService(), logger, now: () => NOW, limit: 3 });
+
+    expect(res).toMatchObject({ scanned: 3, sent: 3, truncated: true });
+    expect(lines(logger.info)[0]).toMatch(/Batch limit 3 reached — more rows remain/);
+  });
+
+  it('an exactly-full batch is not reported as truncated', async () => {
+    // Read one row past the limit, so "there are more" is a fact rather than an
+    // inference from a full page — otherwise every boot with exactly `limit`
+    // stranded rows tells the operator to expect a second sweep that has
+    // nothing to do.
+    const engine = fakeEngine(
+      Array.from({ length: 3 }, (_, i) => ({ id: `row-${i}`, created_at: ago(min(30 + i)) })),
+    );
+    const logger = fakeLogger();
+
+    const res = await sweepStrandedOutbox({ engine, service: fakeService(), logger, now: () => NOW, limit: 3 });
+
+    expect(res).toMatchObject({ scanned: 3, sent: 3, truncated: false });
+    expect(lines(logger.info)[0]).not.toMatch(/Batch limit/);
+  });
+
+  it('summarises a normal sweep in one info line', async () => {
+    const engine = fakeEngine([{ id: 'row-a', created_at: ago(min(30)) }]);
+    const logger = fakeLogger();
+
+    await sweepStrandedOutbox({ engine, service: fakeService(), logger, now: () => NOW });
+
+    expect(logger.info).toHaveBeenCalledTimes(1);
+    const line = lines(logger.info)[0];
+    expect(line).toMatch(/outbox sweep — 1 sys_email row\(s\) stranded at 'queued' for over 5m/);
+    expect(line).toMatch(/1 sent inline, 0 failed, 0 skipped\./);
+    expect(logger.error).not.toHaveBeenCalled();
+  });
+});
diff --git a/packages/plugins/plugin-email/src/outbox-sweep.ts b/packages/plugins/plugin-email/src/outbox-sweep.ts
new file mode 100644
index 0000000000..56998acb12
--- /dev/null
+++ b/packages/plugins/plugin-email/src/outbox-sweep.ts
@@ -0,0 +1,234 @@
+// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license.
+
+/**
+ * sweepStrandedOutbox — the missing consumer for `sys_email` rows left at
+ * `status:'queued'` (#5161).
+ *
+ * ## What was stranded
+ * A `queued` row had exactly ONE consumption moment: the `afterInsert` drain
+ * hook that fires during the insert itself (and, since #5160, the
+ * `email.send.async` job `send()` publishes). Nothing ever looked at such a row
+ * again. A process that died between the insert and the delivery — or a drain
+ * hook whose delivery threw — left the row at `queued` forever: a state named
+ * after a queue that had no consumer, which is `declared ≠ delivered` one layer
+ * up from the transports #5087 fixed.
+ *
+ * This is that consumer, run once per boot at `kernel:ready`.
+ *
+ * ## Which rows it touches, and why the age gate is not optional
+ * Only rows that are **older than {@link OUTBOX_SWEEP_MIN_AGE_MS}**. A row
+ * inserted seconds ago is not stranded — it is somebody's in-flight work: this
+ * process's `send()` (between `insert` and `transport.send`), this process's
+ * drain hook (deferred by `setTimeout(0)`), or, in a multi-instance
+ * deployment, the same on a sibling instance that this process knows nothing
+ * about. The age gate is what keeps a boot from stealing live work and sending
+ * a message twice.
+ *
+ * Note what the gate deliberately is NOT: "created before this process
+ * booted". This instance's boot time says nothing about a sibling that
+ * inserted a row one second ago — that row is younger than our boot and very
+ * much not ours. Age is the only property that means the same thing on every
+ * instance.
+ *
+ * Two further guards sit under it, and they are backstops, not permission:
+ *   - rows currently owned by this process's `send()` are skipped
+ *     ({@link EmailService.isServiceManaged});
+ *   - a row that already carries a `message_id`, or is no longer `queued`, is
+ *     left alone — the same idempotency the `email.send.async` subscriber
+ *     applies before it delivers.
+ *
+ * ## How it delivers — the mode decides, not this function
+ * Queue mode (#5160): the row is published as an `{ rowId }` job to
+ * {@link EMAIL_SEND_QUEUE} through {@link EmailService.enqueuePersistedRow},
+ * i.e. the SAME payload, options and `sys_email:<id>` idempotency key `send()`
+ * publishes — so a swept row that still has a pending job collapses onto it
+ * instead of racing a second worker onto the same row, and the retry/DLQ
+ * budget is the queue's, exactly as for a fresh send.
+ *
+ * Inline mode: {@link EmailService.deliverPersistedRow} finalizes the row in
+ * place, which is what the drain hook would have done had the process lived.
+ * Inline mode has no cross-process coordination — it never had any — so the
+ * age gate is the whole of the protection there. Durable multi-instance
+ * delivery is what queue mode exists for.
+ */
+
+import type { SendEmailResult } from '@objectstack/spec/contracts';
+
+/** System read context — a boot sweep is not an end-user query. */
+const SYSTEM_CTX = { isSystem: true, positions: [], permissions: [] } as const;
+
+/** Backing object; a constant so the query and the log line agree. */
+export const OUTBOX_OBJECT = 'sys_email';
+
+/**
+ * How old a `queued` row must be before the sweep treats it as stranded
+ * rather than in flight. Five minutes: long enough that any live `send()` /
+ * drain-hook delivery has either finalized the row or thrown (the inline retry
+ * loop caps its own backoff at 2s), short enough that a restart after a crash
+ * still gets the mail out in the same maintenance window.
+ */
+export const OUTBOX_SWEEP_MIN_AGE_MS = 5 * 60_000;
+
+/**
+ * Rows handled per boot. A bound, not a policy: the sweep must not turn a
+ * pathological backlog into an unbounded burst of transport calls during
+ * startup. When it truncates it says so, and the next boot continues.
+ */
+export const OUTBOX_SWEEP_LIMIT = 500;
+
+/** Human form of the age gate, for the log lines. */
+function humanMs(ms: number): string {
+  if (ms % 60_000 === 0) return `${ms / 60_000}m`;
+  if (ms % 1000 === 0) return `${ms / 1000}s`;
+  return `${ms}ms`;
+}
+
+/**
+ * Structural logger — `meta` is `any` so the kernel's own `Logger`
+ * (`meta?: Record<string, any>`) satisfies it without a cast at the call site.
+ * Every line this module writes is self-contained text; nothing is passed as
+ * meta, because a consequence buried in a metadata bag is a consequence nobody
+ * greps.
+ */
+interface SweepLogger {
+  info?: (msg: string, meta?: any) => void;
+  warn?: (msg: string, meta?: any) => void;
+  error?: (msg: string, meta?: any) => void;
+}
+
+/** The two `EmailService` entry points the sweep needs, and nothing else. */
+export interface OutboxSweepService {
+  isServiceManaged(id: string): boolean;
+  /** Publish the row as a durable job; false ⇒ queue delivery is not in force. */
+  enqueuePersistedRow(rowId: string): Promise<boolean>;
+  deliverPersistedRow(row: Record<string, unknown>): Promise<SendEmailResult>;
+}
+
+export interface SweepStrandedOutboxOptions {
+  /** ObjectQL-shaped engine (`find(object, { where, orderBy, limit })`). */
+  engine: { find(object: string, opts: Record<string, unknown>): Promise<unknown> };
+  service: OutboxSweepService;
+  logger?: SweepLogger;
+  /** Clock seam for tests. */
+  now?: () => number;
+  /** Override {@link OUTBOX_SWEEP_MIN_AGE_MS} (tests). */
+  minAgeMs?: number;
+  /** Override {@link OUTBOX_SWEEP_LIMIT} (tests). */
+  limit?: number;
+}
+
+export interface OutboxSweepResult {
+  /** Rows the query returned — already past the age gate. */
+  scanned: number;
+  /** Rows handed to the durable queue as `{ rowId }` jobs. */
+  requeued: number;
+  /** Rows delivered inline in this sweep and finalized `sent`. */
+  sent: number;
+  /** Rows delivered inline in this sweep and finalized `failed`. */
+  failed: number;
+  /** Rows deliberately left alone (service-managed, or already delivered). */
+  skipped: number;
+  /** True when the batch limit was reached — more remain for the next boot. */
+  truncated: boolean;
+}
+
+/** Normalize the two find() return shapes ObjectQL hands back. */
+function toRows(raw: unknown): Array<Record<string, any>> {
+  if (Array.isArray(raw)) return raw as Array<Record<string, any>>;
+  const data = (raw as any)?.data;
+  return Array.isArray(data) ? data : [];
+}
+
+/**
+ * Advance every stranded `sys_email` row once. Never throws for a single
+ * row — one undeliverable message must not stop the rest of the backlog —
+ * but DOES propagate a failure of the query itself, so the caller can report
+ * that the sweep as a whole did not happen.
+ */
+export async function sweepStrandedOutbox(
+  opts: SweepStrandedOutboxOptions,
+): Promise<OutboxSweepResult> {
+  const { engine, service, logger } = opts;
+  const now = opts.now?.() ?? Date.now();
+  const minAgeMs = opts.minAgeMs ?? OUTBOX_SWEEP_MIN_AGE_MS;
+  const limit = Math.max(1, opts.limit ?? OUTBOX_SWEEP_LIMIT);
+  const cutoff = new Date(now - minAgeMs).toISOString();
+
+  // One row past the batch so "there are more" is a fact, not an inference
+  // from a full page (a page that is exactly full is the common false positive).
+  const page = toRows(await engine.find(OUTBOX_OBJECT, {
+    where: { status: 'queued', created_at: { $lt: cutoff } },
+    orderBy: [{ field: 'created_at', order: 'asc' }],
+    limit: limit + 1,
+    context: SYSTEM_CTX,
+  }));
+  const rows = page.slice(0, limit);
+
+  const result: OutboxSweepResult = {
+    scanned: rows.length,
+    requeued: 0,
+    sent: 0,
+    failed: 0,
+    skipped: 0,
+    truncated: page.length > limit,
+  };
+  if (rows.length === 0) return result;
+
+  /** Say a per-row breakage once — the rest is in the summary counts. */
+  let rowErrorReported = false;
+
+  for (const row of rows) {
+    const rowId = row?.id != null ? String(row.id) : '';
+    // Backstops under the age gate: never touch a row `send()` owns right
+    // now, and never re-send one the transport already accepted.
+    if (!rowId || row.status !== 'queued' || row.message_id) { result.skipped++; continue; }
+    if (service.isServiceManaged(rowId)) { result.skipped++; continue; }
+
+    try {
+      if (await service.enqueuePersistedRow(rowId)) {
+        result.requeued++;
+        continue;
+      }
+      // Inline mode — or a publish that failed, in which case delivering
+      // inline is what `send()` itself falls back to: the row is already
+      // committed, and leaving it for nobody is the bug being fixed.
+      const delivered = await service.deliverPersistedRow(row);
+      if (delivered.status === 'failed') result.failed++;
+      else result.sent++;
+    } catch (err: any) {
+      result.failed++;
+      if (!rowErrorReported) {
+        rowErrorReported = true;
+        logger?.error?.(
+          `EmailServicePlugin: outbox sweep could not advance sys_email row '${rowId}' — that message stays `
+          + 'at `queued`, undelivered, and nothing will look at it again until the next restart, while the '
+          + 'server keeps reporting healthy. Fix: the cause below comes from the datasource or the queue, '
+          + 'not from the message itself (a message that cannot be sent is recorded as `failed` on its own '
+          + `row); restore that dependency and restart to re-sweep. Cause: ${err?.message ?? err}`,
+        );
+      }
+    }
+  }
+
+  logger?.info?.(
+    `EmailServicePlugin: outbox sweep — ${result.scanned} sys_email row(s) stranded at 'queued' for over `
+    + `${humanMs(minAgeMs)} (accepted, never delivered): ${result.requeued} re-queued for durable delivery, `
+    + `${result.sent} sent inline, ${result.failed} failed, ${result.skipped} skipped`
+    + (result.truncated
+      ? `. Batch limit ${limit} reached — more rows remain and will be swept on the next boot.`
+      : '.'),
+  );
+
+  if (result.failed > 0) {
+    logger?.error?.(
+      `EmailServicePlugin: ${result.failed} stranded sys_email row(s) could NOT be delivered by the boot `
+      + 'sweep. Those messages were accepted by the platform and have still never reached a recipient; '
+      + 'nothing retries them in this process. Fix: read the failures with '
+      + "`SELECT id, error FROM sys_email WHERE status = 'failed'`, fix the transport (Settings → Mail), and "
+      + 'turn on Settings → Mail → "Durable queue delivery" so future failures are retried and dead-lettered '
+      + 'instead of depending on the next restart.',
+    );
+  }
+
+  return result;
+}
diff --git a/scripts/check-durability-degradation-log-level.mjs b/scripts/check-durability-degradation-log-level.mjs
index 043a60fb40..3c4bcb563f 100644
--- a/scripts/check-durability-degradation-log-level.mjs
+++ b/scripts/check-durability-degradation-log-level.mjs
@@ -132,6 +132,10 @@ const DURABILITY_CRITICAL_CALLEES = new Map([
         'performSeedWrite',
         "A seed write's post-write roll-up summary recompute was swallowed — the rows landed, but a persisted summary column now disagrees with the detail rows it summarizes and nothing recomputes it, while every row counter and `success` still read clean (#4998, framework#3147).",
     ],
+    [
+        'deliverPersistedRow',
+        "An accepted email was never transmitted — the sys_email row stays at `status:'queued'` with the caller already told the message was accepted, and the only other reader of that row is the once-per-boot outbox sweep (#5161).",
+    ],
     [
         'dropPromotedDraftRow',
         "A published draft was never drained — the active row is correct, but the `state='draft'` row is still in `sys_metadata`, so Studio/Setup keeps showing unpublished changes that do not exist and the next publish promotes the same stale body again (#4981).",