diff --git a/cron/bot_chat_delivery.py b/cron/bot_chat_delivery.py
new file mode 100644
index 0000000000000..5d38f96f24dca
--- /dev/null
+++ b/cron/bot_chat_delivery.py
@@ -0,0 +1,128 @@
+"""Defer never-started cron outputs behind unsupported Bot Chat owners.
+
+Inspired by 686f6c61's queue proposal (#100319). Unlike retrying failed CLI
+turns, only pending requests are eligible: a persisted claim never expires.
+"""
+from __future__ import annotations
+
+import contextvars
+import json
+import logging
+import threading
+from pathlib import Path
+
+from hermes_cli.active_sessions import _FileLock
+from hermes_constants import get_hermes_home
+from utils import atomic_json_write
+
+logger = logging.getLogger(__name__)
+_running: set[Path] = set()
+_running_lock = threading.Lock()
+
+
+def _root() -> Path:
+    return get_hermes_home().resolve() / "cron" / "bot_chat_pending"
+
+
+def read_pending(key: str) -> dict | None:
+    try:
+        return json.loads((_root() / f"{key}.json").read_text(encoding="utf-8"))
+    except FileNotFoundError:
+        return None
+
+
+def _records(root: Path) -> list[tuple[Path, dict]]:
+    records = []
+    for path in root.glob("*.json"):
+        try:
+            record = json.loads(path.read_text(encoding="utf-8"))
+        except (json.JSONDecodeError, UnicodeDecodeError) as exc:
+            # Keep damaged receipts as evidence; never replay them or block peers.
+            logger.error("Unreadable deferred Bot Chat receipt %s: %s", path, exc)
+            continue
+        records.append((path, record))
+    return records
+
+
+def defer(key: str, job: dict, content: str, profile: str, home: Path) -> dict:
+    root = _root()
+    root.mkdir(parents=True, exist_ok=True, mode=0o700)
+    with _FileLock(root / ".lock"):
+        record = read_pending(key)
+        if record is not None:
+            if record["content"] != content or record["home"] != str(home):
+                raise ValueError("delivery id already belongs to a different payload")
+            return record
+        sequence = max((record["sequence"] for _, record in _records(root)), default=0) + 1
+        record = dict(id=key, status="queued", job=job, content=content,
+                      profile=profile, home=str(home), sequence=sequence)
+        atomic_json_write(root / f"{key}.json", record, fsync_dir=True, mode=0o600)
+        return record
+
+
+def drain(root: Path | None = None) -> None:
+    """Serialize drains across processes without holding the producer lock."""
+    root = root if root is not None else _root()
+    if root.is_dir():
+        with _FileLock(root / ".drain.lock"):
+            _drain(root)
+
+
+def _drain(root: Path) -> None:
+    """Claim before execution. Errors/interruptions never authorize another turn."""
+    from cron.scheduler_delivery import _deliver_to_bot_chat
+    from tools.bot_live_delivery import find_canonical_live_owner, find_canonical_owner
+
+    with _FileLock(root / ".lock"):
+        records = sorted(_records(root), key=lambda item: item[1]["sequence"])
+    for path, _ in records:
+        with _FileLock(root / ".lock"):
+            record = json.loads(path.read_text(encoding="utf-8"))
+            if record["status"] != "queued":
+                continue
+            home = Path(record["home"])
+            try:
+                owner = find_canonical_owner(home)
+                if owner is not None and find_canonical_live_owner(home) is None:
+                    continue
+            except Exception:
+                # Discovery uncertainty is not permission to launch.
+                continue
+            record["status"] = "claimed"
+            atomic_json_write(path, record, fsync_dir=True, mode=0o600)
+        job = record["job"]
+        job.pop("_bot_chat_delivery_receipts", None)
+        try:
+            error = _deliver_to_bot_chat(job, record["content"], record["profile"], deferred=record)
+        except Exception as exc:
+            # The claim survives uncertainty; one failed attempt must not stop peers.
+            error = f"{type(exc).__name__}: {exc}"
+            logger.exception("Deferred Bot Chat delivery %s failed", record["id"])
+        receipt = job.get("_bot_chat_delivery_receipts", {}).get(
+            f"bot-chat:{record['profile'] or '(own)'}")
+        status = "transferred" if receipt else "ambiguous" if error else "settled"
+        record.update(status=status, error=error)
+        # A transferred live-owner receipt remains authoritative, including queued.
+        atomic_json_write(path, record, fsync_dir=True, mode=0o600)
+
+
+def drain_in_background() -> None:
+    """Do not hold up unrelated cron ticks while the eventual Bot Chat turn runs."""
+    home = get_hermes_home().resolve()
+    root = home / "cron" / "bot_chat_pending"
+    if not root.is_dir():
+        return
+    with _running_lock:
+        if home in _running:
+            return
+        _running.add(home)
+
+    def run():
+        try:
+            drain(root)
+        finally:
+            with _running_lock:
+                _running.discard(home)
+
+    threading.Thread(target=contextvars.copy_context().run, args=(run,), daemon=True,
+                     name="cron-bot-chat-drain").start()
diff --git a/cron/scheduler.py b/cron/scheduler.py
index d91f9ef374ebb..99d8b0485f1c4 100644
--- a/cron/scheduler.py
+++ b/cron/scheduler.py
@@ -3789,6 +3789,11 @@ def tick(
             logger.debug("Cron dispatch paused while gateway drains existing work")
             return 0
 
+        from cron.bot_chat_delivery import drain, drain_in_background
+        if sync:
+            drain()
+        else:
+            drain_in_background()
         _maybe_reap_dead_owners()
         # Periodic worktree GC (6h, threaded) — the only sweep gateway-only boxes get.
         try:
diff --git a/cron/scheduler_delivery.py b/cron/scheduler_delivery.py
index 90c02b331c0ef..459772d88d705 100644
--- a/cron/scheduler_delivery.py
+++ b/cron/scheduler_delivery.py
@@ -653,7 +653,7 @@ def _get_bot_chat_delivery_timeout() -> int:
         return 600
 
 
-def _deliver_to_bot_chat(job: dict, content: str, profile: str) -> Optional[str]:
+def _deliver_to_bot_chat(job: dict, content: str, profile: str, *, deferred: Optional[dict] = None) -> Optional[str]:
     """Hand output to the live Bot Chat owner, or use the legacy unowned CLI lane.
 
     None means completed; a queued/claimed receipt returns an explicit unverified status
@@ -679,7 +679,11 @@ def _deliver_to_bot_chat(job: dict, content: str, profile: str) -> Optional[str]
     )
     try:
         source_home = get_hermes_home().resolve()
-        home = (get_profile_dir(profile) if profile else source_home).resolve()
+        from pathlib import Path
+        home = (Path(deferred["home"]) if deferred is not None else
+                get_profile_dir(profile) if profile else source_home).resolve()
+        if deferred is not None and not (home / "state.db").is_file():
+            return f"bot-chat delivery target no longer exists: {home}; do not resend"
         # run_one_job/claim_fire attach the durable execution id before delivery. The
         # transient fallback supports direct helper callers, never deduping recurring
         # runs by their (potentially identical) output or previous last_run timestamp.
@@ -690,9 +694,26 @@ def _deliver_to_bot_chat(job: dict, content: str, profile: str) -> Optional[str]
             [str(source_home), job_id, str(run_id), str(home)],
             ensure_ascii=False, separators=(",", ":"),
         ).encode("utf-8")).hexdigest()
+        if deferred is not None:
+            key = deferred["id"]
         # Read BEFORE discovery: the previous owner may have exited after accepting.
         # No receipt state, including ambiguous/failed, authorizes a CLI replay.
         receipt = read_delivery_result(home, key)
+        if receipt is None and not deferred:
+            from cron.bot_chat_delivery import defer, read_pending
+            from tools.bot_live_delivery import find_canonical_owner
+
+            pending = read_pending(key)
+            if pending is None and find_canonical_live_owner(home) is None and find_canonical_owner(home):
+                pending = defer(key, dict(job), content, profile, home)
+            if pending is not None:
+                if pending["content"] != content or pending["home"] != str(home):
+                    raise ValueError("delivery id already belongs to a different payload")
+                status = pending["status"]
+                target = f"bot-chat:{profile_label}"
+                job.setdefault("_bot_chat_delivery_receipts", {})[target] = {
+                    "status": status, "delivery_id": key}
+                return None if status == "settled" else f"{target} {status} (receipt {key}): completion unverified; do not resend"
         if receipt is None:
             owner = find_canonical_live_owner(home)
             if owner is not None:
@@ -735,13 +756,13 @@ def _fail(msg: str, **log_kwargs) -> str:
     from agent.delegation_context import delegated_child_subprocess_env
     from tools.environments.local import strip_launch_profile_env
     env = strip_launch_profile_env(delegated_child_subprocess_env(os.environ))
-    if profile:
-        argv += ["-p", profile]
-        # -p owns profile resolution; this scheduler's HERMES_HOME must not shadow it.
-        env.pop("HERMES_HOME", None)
-    else:
-        # Multiplex workers carry the profile in a ContextVar, not os.environ.
-        env["HERMES_HOME"] = str(source_home)
+    if not home.is_dir():
+        return _fail(f"bot-chat delivery target no longer exists: {home}; do not resend")
+    # Discovery (or deferred admission) owns the destination, not HOME or a
+    # subsequently changed active_profile. Do not resolve the name a second time.
+    env["HERMES_HOME"] = str(home)
+    if home.parent.name != "profiles":
+        argv += ["-p", "default"]
 
     query_file = None
     try:
@@ -761,7 +782,7 @@ def _fail(msg: str, **log_kwargs) -> str:
         if result.returncode != 0:
             tail = (result.stderr or result.stdout or "").strip()[-500:]
             return _fail(
-                f"bot-chat delivery to profile '{profile_label}' failed (exit {result.returncode})"
+                f"bot-chat delivery to profile '{profile_label}' failed (exit {result.returncode}) at {home}"
                 + (f": {tail}" if tail else ""))
         logger.info("Job '%s': delivered to Bot Chat of profile '%s'", job_id, profile_label)
         return None
diff --git a/evals/botmode-dm-delivery/README.md b/evals/botmode-dm-delivery/README.md
new file mode 100644
index 0000000000000..7d4b9bbdb8a9c
--- /dev/null
+++ b/evals/botmode-dm-delivery/README.md
@@ -0,0 +1,107 @@
+# Native Bot Mode delivery probe
+
+Real Electron, production Python backend and tool execution, disposable HOME/HERMES_HOME,
+loopback scripted inference (no paid model). Linux seat fixture; run from repository root:
+
+```sh
+npm ci --no-audit --no-fund
+cp evals/botmode-dm-delivery/probe-dm-delivery.spec.ts apps/desktop/e2e/
+git apply evals/botmode-dm-delivery/mock-trigger.patch
+(cd apps/desktop && npm run build)
+(cd apps/desktop && DISPLAY=:0 XAUTHORITY=/run/user/1000/xauth_cnpsqU \
+  XDG_RUNTIME_DIR=/run/user/1000 VIRTUAL_ENV="$VIRTUAL_ENV" \
+  HERMES_DESKTOP_CDP_PORT=off npx playwright test e2e/probe-dm-delivery.spec.ts --reporter=list)
+git apply -R evals/botmode-dm-delivery/mock-trigger.patch
+rm apps/desktop/e2e/probe-dm-delivery.spec.ts
+```
+
+Use the current seat's actual Xauthority path and an existing runtime venv.
+Artifacts default to `/tmp/botmode-dm-review/native` (override with `BOT_DM_EVIDENCE`); sandbox path is printed. The
+fixture's generated hermes shim pins every child to this checkout, not an installed launcher.
+
+## Verified results
+
+Base `cf35e7351e770`: nested beta quiet CLI executes real `message_agent` to the named
+alpha Desktop owner. Admission, execution and reply succeed, and the quiet sender
+receives its completion. A reload then shows the attributed incoming message. The
+live pre-reload view showed the reply but omitted the incoming row in this fixture;
+that renderer-refresh behavior is not fixed by this cron change.
+
+Base CLI-owner case: separate cron producer returns exact `SESSION_NOT_OWNED` refusal.
+Release owner, run scheduler tick, open Beta in Desktop: no cron output, assertion red.
+Fixed case: queued receipt; release owner; real synchronous scheduler tick drains;
+Beta Desktop renders `CLI_OWNER_CRON_SENTINEL` and its reply once. Final two-case
+native run: **2 passed (1.2m)**. Supported-owner nested case also passed independently
+on base (**1 passed (40.0s)**).
+
+This does not establish native Windows parity, general retry after unowned CLI
+failure, or correction of the issue #105460 `--in ~` premise. CLI title resolution
+is profile-DB-based, not workspace selection; the named live-owner route is positive.
+
+The queue is deliberately at-most-once after claim. A crash before spawning but
+after claiming remains inspectable as claimed; it is not retried automatically.
+
+## Independent review follow-up
+
+The probe now admits the named Beta destination from the default scheduler, then
+changes the ticker's `HOME` to a different existing directory before drain.
+Published head `c91dfcbfe810c` fails with `Profile 'beta' does not exist`, leaving
+zero sentinel inputs in Beta. The follow-up pins the admitted home and ID; Beta
+renders one input and reply. Set `BOT_DM_CORRUPT=1` to add one malformed JSON
+record alongside the valid admission: before the follow-up the real tick raises
+`JSONDecodeError`; afterward the damaged record stays on disk while Beta delivers.
+
+Fresh built native run with both root change and corruption: **2 passed (1.4m)**,
+including the existing nested `message_agent` control. Receipts/screenshots:
+`/tmp/botmode-dm-review/{native-red2,corrupt-red,final-native}` and matching `.log`
+files. The first follow-up run also exposed a fixture mistake (the changed HOME
+was not created, so `--in ~` correctly refused); that failed receipt is retained
+in `native-green.log`, and both source legs were rerun with an existing HOME.
+Prior `/tmp/botmode-dm-recovery*` evidence remains untouched.
+
+## Per-record delivery exception isolation
+
+Set `BOT_DM_EXCEPTION=1` and select `-g "cron output"` for the controlled native
+exception probe. `exception-tick.py` raises `PermissionError` at the actual
+post-discovery target `Path.is_dir()` boundary, not at the delivery helper.
+The same exception was first reproduced with real directory traversal permission
+loss on Python 3.11/Linux; the retained test uses a portable controlled fault.
+Before the guard, native tick exits 1, leaving the head claimed and sibling queued.
+Afterward, the head is ambiguous, the sibling settles and renders once, and a
+second real tick replays neither. Logs: `/tmp/botmode-dm-exception-{red,green}.log`;
+receipts and screenshot: `/tmp/botmode-dm-exception/{red,green}/`.
+
+The review's repeated-head starvation claim is not reachable: the claim commits
+before delivery and later scans skip every non-queued record. One failed tick is
+real; recurring replay of that same head is not. Indefinite queued/payload retention
+is intentional, with no TTL or automatic ambiguous retry introduced here.
+
+## Ordinary custom-root fallback (#104066 / #104055)
+
+`probe-cron-root.spec.ts` adds the never-deferred sibling: copy it to
+`apps/desktop/e2e/` and run with the same native fixture (no mock-trigger patch
+needed for this case). It keeps default's real Desktop Bot Chat lease, submits
+ordinary cron output to unowned Alpha from a separate Python producer under a
+custom Hermes root, and holds the real quiet CLI child at loopback inference.
+The child shim PID must match Alpha's real CLI lease; default's lease is unchanged.
+After release, Alpha has exactly one input and Desktop renders the output.
+The same case removes unused Beta and verifies delivery neither recreates Beta nor
+creates a second `.hermes` root under HOME.
+
+Both `origin/main`'s scheduler and pre-follow-up `c827ae179d67c` fail with
+`Profile 'alpha' does not exist` before any recipient turn. Fixed native run:
+**1 passed (46.3s)**. Two invariant tests exercise the actual CLI startup resolver
+across named/default/own destinations with a changed active profile, and refusal
+when the destination is missing initially or disappears during discovery:
+**5 failed before, 5 passed after**. The old env-clearing test is replaced by these
+behavior checks rather than retaining the broken expectation.
+
+Evidence: `/tmp/botmode-cron-root/{before2,origin-main,after}.log`,
+`after/{owners.json,children.log,rows.json,result.json,missing.json,ordinary-recipient.png}`.
+The first fixture attempt (`before.log`) used the wrong default row label; the
+actual Desktop label is Hermes. No production failure is claimed for that attempt.
+Full cron directory: **1346 passed, 1 skipped across 116 files**; sibling mailbox,
+DM, gateway consumer and profile tests: **124 passed, 3 skipped across 4 files**.
+Credit @fangliquanflq's #104066 for the root-boundary diagnosis and anchoring fix;
+this combined branch reuses its already-resolved destination instead of repeating
+name resolution. No retry or receipt semantics change.
diff --git a/evals/botmode-dm-delivery/exception-tick.py b/evals/botmode-dm-delivery/exception-tick.py
new file mode 100644
index 0000000000000..c0482dac9d0bf
--- /dev/null
+++ b/evals/botmode-dm-delivery/exception-tick.py
@@ -0,0 +1,41 @@
+"""Controlled filesystem fault; run only against the native disposable sandbox."""
+import json
+import os
+from pathlib import Path
+
+from cron import scheduler_delivery as delivery
+from cron.bot_chat_delivery import _root
+from cron.scheduler import tick
+
+root = _root()
+original_which = delivery.shutil.which
+original_is_dir = Path.is_dir
+armed = False
+raised = 0
+
+
+def resolve_cli(*args, **kwargs):
+    global armed
+    armed = raised == 0
+    return original_which(*args, **kwargs)
+
+
+def is_dir(path):
+    global armed, raised
+    if armed and path == Path(os.environ["HERMES_HOME"]) / "profiles" / "beta":
+        armed = False
+        raised += 1
+        raise PermissionError("controlled target traversal denied after discovery")
+    return original_is_dir(path)
+
+
+delivery.shutil.which = resolve_cli
+Path.is_dir = is_dir
+try:
+    tick(verbose=False)
+    tick(verbose=False)
+finally:
+    Path.is_dir = original_is_dir
+    delivery.shutil.which = original_which
+    records = [json.loads(p.read_text(encoding="utf-8")) for p in root.glob("*.json") if p.name != "broken.json"]
+    Path(os.environ["BOT_DM_EXCEPTION_RECEIPT"]).write_text(json.dumps({"raised": raised, "records": records}, indent=2), encoding="utf-8")
diff --git a/evals/botmode-dm-delivery/mock-trigger.patch b/evals/botmode-dm-delivery/mock-trigger.patch
new file mode 100644
index 0000000000000..bae97952c95db
--- /dev/null
+++ b/evals/botmode-dm-delivery/mock-trigger.patch
@@ -0,0 +1,26 @@
+diff --git a/tests-js/scripts/mock-server.ts b/tests-js/scripts/mock-server.ts
+index 6dffd2cf0f388..09c6de07d81b2 100644
+--- a/tests-js/scripts/mock-server.ts
++++ b/tests-js/scripts/mock-server.ts
+@@ -500,6 +500,21 @@ export function startMockServer(options: MockServerOptions = {}): Promise<MockSe
+             _receivedUserTexts.push(userText)
+           }
+ 
++          const dmMatch = userText.match(/E2E_DM\(([^)]+)\)\[([\s\S]*)\]/)
++          const lastUserIndex = messages.lastIndexOf(lastUserMsg)
++          if (dmMatch && !messages.slice(lastUserIndex + 1).some(m => m?.role === 'tool')) {
++            const turn: ScriptedTurn = {
++              text: '',
++              toolCalls: [{ name: 'message_agent', args: { target: dmMatch[1], message: dmMatch[2] } }],
++            }
++            if (stream) {
++              streamScriptedTurn(res, model, turn)
++            } else {
++              nonStreamingScriptedTurn(res, model, turn)
++            }
++            return
++          }
++
+           const isInterimTrigger = userText.includes('E2E_INTERIM_TRIGGER')
+           const isSidebarTrigger = userText.includes('E2E_SIDEBAR_TRIGGER')
+           const isSidebarCrossTrigger = userText.includes('E2E_SIDEBAR_CROSS')
diff --git a/evals/botmode-dm-delivery/probe-cron-root.spec.ts b/evals/botmode-dm-delivery/probe-cron-root.spec.ts
new file mode 100644
index 0000000000000..d9a9757e32951
--- /dev/null
+++ b/evals/botmode-dm-delivery/probe-cron-root.spec.ts
@@ -0,0 +1,105 @@
+import { execFileSync, spawn } from 'node:child_process'
+import fs from 'node:fs'
+import path from 'node:path'
+import { buildAppEnv, createSandbox, launchDesktop, waitForAppReady, writeEnvFile, writeMockProviderConfig, type MockBackendFixture } from './fixtures'
+import { MOCK_REPLY, startMockServer } from '../../../tests-js/scripts/mock-server'
+import { expect, test } from './test'
+
+const repo = path.resolve(import.meta.dirname, '../../..')
+const python = path.join(process.env.VIRTUAL_ENV || path.join(repo, '.venv'), 'bin', 'python')
+const evidence = process.env.BOT_DM_EVIDENCE || '/tmp/botmode-cron-root/native'
+let fixture: MockBackendFixture
+let env: Record<string, string>
+
+test.beforeAll(async () => {
+  fs.mkdirSync(evidence, { recursive: true })
+  const sandbox = createSandbox('cron-root')
+  const mock = await startMockServer({ holdFirstCompletionContaining: 'ORDINARY_CRON_SENTINEL' })
+  for (const name of ['default', 'alpha', 'beta']) {
+    const home = name === 'default' ? sandbox.hermesHome : path.join(sandbox.hermesHome, 'profiles', name)
+    fs.mkdirSync(home, { recursive: true })
+    writeMockProviderConfig(home, mock.url)
+    writeEnvFile(home)
+    fs.writeFileSync(path.join(home, 'SOUL.md'), `# ${name}\nA Bot Mode teammate.\n`)
+    fs.writeFileSync(path.join(home, 'profile.yaml'), `name: ${name}\nui_meta:\n  hermes-bots: {}\n`)
+  }
+  const bin = path.join(sandbox.root, 'bin')
+  fs.mkdirSync(bin)
+  fs.writeFileSync(path.join(bin, 'hermes'), `#!/bin/sh\ncd ${repo}\nprintf '%s\\n' "$$ $HERMES_HOME $*" >> ${evidence}/children.log\nexec ${python} -m hermes_cli.main "$@"\n`, { mode: 0o755 })
+  env = buildAppEnv(sandbox, { HOME: sandbox.root, HERMES_DESKTOP_PYTHON: python,
+    HERMES_DESKTOP_HERMES: path.join(bin, 'hermes'), PATH: `${bin}:${process.env.PATH}`,
+    PYTHONPATH: repo, HERMES_SINGLE_QUERY_LINGER_SECONDS: '1' })
+  const { app, page } = await launchDesktop(env)
+  fixture = { app, page, sandbox, mock, mockUrl: mock.url, cleanup: async () => {
+    await app.close().catch(() => undefined)
+    await mock.close()
+  } }
+  fs.writeFileSync(path.join(evidence, 'sandbox.txt'), sandbox.root)
+  await waitForAppReady(fixture, 120_000)
+})
+
+test.afterAll(async () => { await fixture?.cleanup() })
+
+function probe(script: string, extraEnv = {}) {
+  return JSON.parse(execFileSync(python, ['-c', script], { env: { ...env, ...extraEnv }, cwd: repo, encoding: 'utf8', timeout: 30_000 }))
+}
+
+async function openBot(name: string) {
+  const page = fixture.page
+  await page.getByRole('button', { name: 'Bots', exact: true }).or(page.getByRole('tab', { name: 'Bots', exact: true })).first().click()
+  const row = page.getByRole('button', { name: new RegExp(`^${name}\\b`, 'i') }).filter({ visible: true }).first()
+  await expect(row).toBeVisible({ timeout: 30_000 })
+  await row.click()
+  const composer = page.locator('[data-slot="composer-root"] [contenteditable="true"]').filter({ visible: true }).first()
+  await expect(composer).toBeVisible({ timeout: 120_000 })
+  return composer
+}
+
+test('ordinary cron pins an unowned named target under a custom root', async () => {
+  test.setTimeout(240_000)
+  const page = fixture.page
+  const composer = await openBot('Hermes')
+  await expect(page.getByText('Say something to get started.').filter({ visible: true })).toBeVisible({ timeout: 120_000 })
+  await composer.fill('initialize default Desktop owner')
+  await page.keyboard.press('Enter')
+  await expect(page.getByText(MOCK_REPLY).filter({ visible: true }).first()).toBeVisible({ timeout: 60_000 })
+  const discovery = 'from pathlib import Path; import json,os; from tools.bot_live_delivery import find_canonical_owner; h=Path(os.environ["HERMES_HOME"]); print(json.dumps({"default":find_canonical_owner(h),"alpha":find_canonical_owner(h/"profiles"/"alpha")}))'
+  const before = probe(discovery)
+  expect(before.default.surface).toBe('desktop')
+  expect(before.alpha).toBeNull()
+  const output = fs.openSync(path.join(evidence, 'producer.log'), 'w')
+  const resultPath = path.join(evidence, 'result.json')
+  const script = `import json; from pathlib import Path; from cron.scheduler_delivery import _deliver_to_bot_chat; j={"id":"ordinary-cron","name":"Ordinary cron","execution_id":"never-deferred"}; result=_deliver_to_bot_chat(j,"ORDINARY_CRON_SENTINEL","alpha"); Path(${JSON.stringify(resultPath)}).write_text(json.dumps({"result":result,"job":j}))`
+  const child = spawn(python, ['-c', script], { env, cwd: repo, stdio: ['ignore', output, output] })
+  try {
+    await expect.poll(() => fs.existsSync(resultPath) ? 'exited' : fixture.mock.receivedPrompts.some(p => p.includes('ORDINARY_CRON_SENTINEL')) ? 'held' : 'waiting', { timeout: 90_000 }).not.toBe('waiting')
+    if (fs.existsSync(resultPath)) {
+      console.log('EARLY_RESULT', fs.readFileSync(resultPath, 'utf8'))
+      expect(JSON.parse(fs.readFileSync(resultPath, 'utf8')).result).toBeNull()
+    }
+    await fixture.mock.waitForHeldCompletion()
+    const held = probe(discovery)
+    expect(held.alpha.surface).toBe('cli')
+    const launched = fs.readFileSync(path.join(evidence, 'children.log'), 'utf8').trim().split('\n').at(-1)!
+    expect(Number(launched.split(' ')[0])).toBe(held.alpha.pid)
+    expect(launched).toContain(path.join(fixture.sandbox.hermesHome, 'profiles', 'alpha'))
+    expect(held.default.lease_id).toBe(before.default.lease_id)
+    expect(fs.existsSync(path.join(fixture.sandbox.hermesHome, 'cron', 'bot_chat_pending'))).toBe(false)
+    fs.writeFileSync(path.join(evidence, 'owners.json'), JSON.stringify({ before, held }, null, 2))
+    fixture.mock.releaseHeldStream()
+    await expect.poll(() => child.exitCode, { timeout: 60_000 }).toBe(0)
+    expect(JSON.parse(fs.readFileSync(resultPath, 'utf8')).result).toBeNull()
+    await openBot('alpha')
+    await expect(page.getByText(/ORDINARY_CRON_SENTINEL/).filter({ visible: true }).first()).toBeVisible({ timeout: 60_000 })
+    const rows = probe('import sqlite3,json,os; from pathlib import Path; h=Path(os.environ["HERMES_HOME"]); print(json.dumps({n:sqlite3.connect(h/"state.db" if n=="default" else h/"profiles"/n/"state.db").execute("select session_id,role,content from messages").fetchall() for n in ["default","alpha"]}))')
+    expect(rows.alpha.filter((r: string[]) => r[1] === 'user' && r[2].includes('ORDINARY_CRON_SENTINEL'))).toHaveLength(1)
+    expect(rows.default.filter((r: string[]) => r[2].includes('ORDINARY_CRON_SENTINEL'))).toHaveLength(0)
+    fs.writeFileSync(path.join(evidence, 'rows.json'), JSON.stringify(rows, null, 2))
+    await page.screenshot({ path: path.join(evidence, 'ordinary-recipient.png') })
+    const missing = probe('import json,os; from pathlib import Path; from cron.scheduler_delivery import _deliver_to_bot_chat; h=Path(os.environ["HERMES_HOME"]); p=h/"profiles"/"beta"; p.rename(h/"profiles"/"beta-removed"); result=_deliver_to_bot_chat({"id":"removed","execution_id":"missing"},"MUST_NOT_RUN","beta"); print(json.dumps({"result":result,"recreated":p.exists(),"wrong_root":(Path.home()/".hermes").exists()}))')
+    expect(missing.result).not.toBeNull()
+    expect(missing.recreated).toBe(false)
+    expect(missing.wrong_root).toBe(false)
+    fs.writeFileSync(path.join(evidence, 'missing.json'), JSON.stringify(missing, null, 2))
+  } finally { fixture.mock.releaseHeldStream(); child.kill(); fs.closeSync(output) }
+})
diff --git a/evals/botmode-dm-delivery/probe-dm-delivery.spec.ts b/evals/botmode-dm-delivery/probe-dm-delivery.spec.ts
new file mode 100644
index 0000000000000..6b57e5a0273c6
--- /dev/null
+++ b/evals/botmode-dm-delivery/probe-dm-delivery.spec.ts
@@ -0,0 +1,118 @@
+import { execFileSync, spawn } from 'node:child_process'
+import fs from 'node:fs'
+import path from 'node:path'
+import { buildAppEnv, createSandbox, launchDesktop, waitForAppReady, writeEnvFile, writeMockProviderConfig, type MockBackendFixture } from './fixtures'
+import { MOCK_REPLY, startMockServer } from '../../../tests-js/scripts/mock-server'
+import { expect, test } from './test'
+
+const repo = path.resolve(import.meta.dirname, '../../..')
+const python = path.join(process.env.VIRTUAL_ENV || path.join(repo, '.venv'), 'bin', 'python')
+let fixture: MockBackendFixture
+let env: Record<string, string>
+const evidence = process.env.BOT_DM_EVIDENCE || '/tmp/botmode-dm-review/native'
+
+test.beforeAll(async () => {
+  fs.mkdirSync(evidence, { recursive: true })
+  const sandbox = createSandbox('dm-delivery')
+  const mock = await startMockServer({ holdFirstCompletionContaining: 'CLI_OWNER_HOLD' })
+  for (const name of ['default', 'alpha', 'beta']) {
+    const home = name === 'default' ? sandbox.hermesHome : path.join(sandbox.hermesHome, 'profiles', name)
+    fs.mkdirSync(home, { recursive: true })
+    writeMockProviderConfig(home, mock.url)
+    writeEnvFile(home)
+    fs.writeFileSync(path.join(home, 'SOUL.md'), `# ${name}\nA Bot Mode teammate.\n`)
+    fs.writeFileSync(path.join(home, 'profile.yaml'), 'name: ' + name + '\nui_meta:\n  hermes-bots: {}\n')
+  }
+  const bin = path.join(sandbox.root, 'bin')
+  fs.mkdirSync(bin)
+  fs.writeFileSync(path.join(bin, 'hermes'), `#!/bin/sh\ncd ${repo}\nexec ${python} -m hermes_cli.main "$@"\n`, { mode: 0o755 })
+  env = buildAppEnv(sandbox, { HOME: sandbox.root, HERMES_DESKTOP_PYTHON: python,
+    HERMES_DESKTOP_HERMES: path.join(bin, 'hermes'), PATH: `${bin}:${process.env.PATH}`,
+    PYTHONPATH: repo, HERMES_SINGLE_QUERY_LINGER_SECONDS: '30' })
+  const { app, page } = await launchDesktop(env)
+  fixture = { app, page, sandbox, mock, mockUrl: mock.url, cleanup: async () => {
+    await app.close().catch(() => undefined)
+    await mock.close()
+  } }
+  console.log('SANDBOX', sandbox.root)
+  fs.writeFileSync(path.join(evidence, 'sandbox.txt'), sandbox.root)
+  await waitForAppReady(fixture, 120_000)
+})
+
+test.afterAll(async () => { await fixture?.cleanup() })
+
+test('cron output waits for a CLI-only owner and arrives after owner release', async () => {
+  test.setTimeout(240_000)
+  const output = fs.openSync(path.join(evidence, 'cli-owner.log'), 'w')
+  const child = spawn(python, ['-m', 'hermes_cli.main', '-p', 'beta', 'chat', '--in', '~', '-c', 'Bot Chat', '--create-if-missing', '-Q', '-q', 'CLI_OWNER_HOLD'], { cwd: repo, env, stdio: ['ignore', output, output] })
+  const cronEnv = { ...env, HERMES_HOME: fixture.sandbox.hermesHome }
+  try {
+    await fixture.mock.waitForHeldCompletion()
+    const script = 'import json; from cron.scheduler_delivery import _deliver_to_bot_chat; j={"id":"cli-residual","name":"CLI residual","execution_id":"fixed-execution"}; result=_deliver_to_bot_chat(j,"CLI_OWNER_CRON_SENTINEL","beta"); print(json.dumps({"result":result,"job":j}))'
+    const setup = process.env.BOT_DM_EXCEPTION ? 'from pathlib import Path; from cron.bot_chat_delivery import defer; from hermes_constants import get_hermes_home; defer("e"*64,{"id":"exception-head"},"EXCEPTION_MUST_NOT_RUN","beta",get_hermes_home()/"profiles"/"beta"); ' : ''
+    const result = JSON.parse(execFileSync(python, ['-c', setup + script], { env: cronEnv, cwd: repo, encoding: 'utf8', timeout: 30_000 }))
+    console.log('CLI_OWNER_CRON_ADMISSION', JSON.stringify(result))
+    fs.writeFileSync(path.join(evidence, 'cli-owner-admission.json'), JSON.stringify(result, null, 2))
+    if (process.env.BOT_DM_CORRUPT) fs.writeFileSync(path.join(fixture.sandbox.hermesHome, 'cron', 'bot_chat_pending', 'broken.json'), '{')
+    fixture.mock.releaseHeldStream()
+    await expect.poll(() => child.exitCode, { timeout: 60_000 }).toBe(0)
+    fs.mkdirSync(path.join(fixture.sandbox.root, 'changed-launch-home'), { recursive: true })
+    const tickArgs = process.env.BOT_DM_EXCEPTION ? [path.join(repo, 'evals/botmode-dm-delivery/exception-tick.py')] : ['-c', 'import time; from cron.scheduler import tick; from cron.bot_chat_delivery import _running; tick(verbose=False);\nwhile _running: time.sleep(0.1)']
+    const receiptPath = path.join(evidence, 'exception-receipts.json')
+    const ticker = spawn(python, tickArgs, { env: { ...cronEnv, HOME: path.join(fixture.sandbox.root, 'changed-launch-home'), BOT_DM_EXCEPTION_RECEIPT: receiptPath }, cwd: repo, stdio: ['ignore', output, output] })
+    await expect.poll(() => ticker.exitCode, { timeout: 90_000 }).not.toBeNull()
+    expect(ticker.exitCode).toBe(0)
+    if (process.env.BOT_DM_EXCEPTION) {
+      const receipts = JSON.parse(fs.readFileSync(receiptPath, 'utf8'))
+      expect(receipts.raised).toBe(1)
+      expect(receipts.records.find((r: { id: string }) => r.id === 'e'.repeat(64)).status).toBe('ambiguous')
+      expect(receipts.records.find((r: { id: string }) => r.id !== 'e'.repeat(64)).status).toBe('settled')
+      expect(dbMessages('beta').filter(([, text]) => text.includes('EXCEPTION_MUST_NOT_RUN'))).toHaveLength(0)
+    }
+    await openBot('beta')
+    expect(dbMessages('beta').filter(([role, text]) => role === 'user' && text.includes('CLI_OWNER_CRON_SENTINEL'))).toHaveLength(1)
+    await expect(fixture.page.getByText(/CLI_OWNER_CRON_SENTINEL/).filter({ visible: true }).first()).toBeVisible({ timeout: 45_000 })
+    await fixture.page.screenshot({ path: path.join(evidence, 'cli-owner-after-release.png') })
+  } finally { fixture.mock.releaseHeldStream(); child.kill(); fs.closeSync(output) }
+})
+
+async function openBot(name: string) {
+  const page = fixture.page
+  await page.getByRole('button', { name: 'Bots', exact: true }).or(page.getByRole('tab', { name: 'Bots', exact: true })).first().click()
+  const row = page.getByRole('button', { name: new RegExp(`^${name}\\b`, 'i') }).filter({ visible: true }).first()
+  await expect(row).toBeVisible({ timeout: 30_000 })
+  await row.click()
+  const composer = page.locator('[data-slot="composer-root"] [contenteditable="true"]').filter({ visible: true }).first()
+  await expect(composer).toBeVisible({ timeout: 120_000 })
+  return composer
+}
+
+function dbMessages(name: string) {
+  const home = name === 'default' ? fixture.sandbox.hermesHome : path.join(fixture.sandbox.hermesHome, 'profiles', name)
+  return JSON.parse(execFileSync(python, ['-c', 'import sqlite3,json,sys; c=sqlite3.connect(sys.argv[1]); print(json.dumps(c.execute("select role,content from messages").fetchall()))', path.join(home, 'state.db')], { env, cwd: repo, encoding: 'utf8' })) as string[][]
+}
+
+test('named Bot Chat receives a nested one-shot message_agent delivery once', async () => {
+  test.setTimeout(300_000)
+  const page = fixture.page
+  const composer = await openBot('alpha')
+  await expect(page.getByText('Say something to get started.').filter({ visible: true })).toBeVisible({ timeout: 120_000 })
+  await composer.fill('initialize alpha owner')
+  await page.keyboard.press('Enter')
+  await expect(page.getByText(MOCK_REPLY).filter({ visible: true }).first()).toBeVisible({ timeout: 60_000 })
+  const output = fs.openSync(path.join(evidence, 'oneshot.log'), 'w')
+  const child = spawn(python, ['-m', 'hermes_cli.main', '-p', 'beta', 'chat', '--in', '~', '-c', 'Bot Chat', '--create-if-missing', '-Q', '-q', 'E2E_DM(alpha)[nested-one-shot-sentinel]'], { cwd: repo, env, stdio: ['ignore', output, output] })
+  try {
+    await expect.poll(() => dbMessages('alpha').filter(([role, text]) => role === 'user' && text.includes('nested-one-shot-sentinel')).length, { timeout: 120_000 }).toBe(1)
+    console.log('ALPHA_ROWS', JSON.stringify(dbMessages('alpha')))
+    console.log('BETA_ROWS', JSON.stringify(dbMessages('beta')))
+    await page.reload()
+    await waitForAppReady(fixture, 120_000)
+    await openBot('alpha')
+    await expect(page.getByText('show message', { exact: true }).first()).toBeVisible({ timeout: 30_000 })
+    for (const toggle of await page.getByText('show message', { exact: true }).all()) await toggle.click()
+    await expect(page.getByText(/nested-one-shot-sentinel/).filter({ visible: true }).first()).toBeVisible({ timeout: 30_000 })
+    await page.screenshot({ path: path.join(evidence, 'nested-delivery-desktop.png') })
+    fs.writeFileSync(path.join(evidence, 'rows.json'), JSON.stringify({ alpha: dbMessages('alpha'), beta: dbMessages('beta'), prompts: fixture.mock.receivedPrompts }, null, 2))
+  } finally { child.kill(); fs.closeSync(output) }
+})
diff --git a/tests/cron/test_bot_chat_cli_home.py b/tests/cron/test_bot_chat_cli_home.py
new file mode 100644
index 0000000000000..96c8c6ae7bd9f
--- /dev/null
+++ b/tests/cron/test_bot_chat_cli_home.py
@@ -0,0 +1,78 @@
+"""The unowned CLI lane executes only at the home used for owner discovery."""
+import json
+from pathlib import Path
+import subprocess
+import sys
+from unittest.mock import Mock
+
+import pytest
+
+from cron import scheduler_delivery as delivery
+from hermes_constants import reset_hermes_home_override, set_hermes_home_override
+
+
+@pytest.mark.parametrize("profile", ["beta", "default", ""])
+def test_cli_keeps_discovered_home_when_launch_selection_changes(tmp_path, monkeypatch, profile):
+    root = tmp_path / "custom"
+    home = root / "profiles" / "beta" if profile == "beta" else root
+    home.mkdir(parents=True)
+    other = root / "profiles" / "other"
+    other.mkdir(parents=True)
+    monkeypatch.setenv("HOME", str(tmp_path))
+    monkeypatch.setenv("HERMES_HOME", str(root))
+    token = set_hermes_home_override(str(root))
+    real_run = subprocess.run
+    seen = []
+
+    def discover(target):
+        assert target == home
+        (root / "active_profile").write_text("other", encoding="utf-8")
+        return None
+
+    def run(argv, **kwargs):
+        # Exercise the actual startup resolver with the production child env/flags.
+        code = ('import json,sys; sys.argv=["hermes"]+json.loads(sys.argv[1]); '
+                'import hermes_cli.main; from hermes_constants import get_hermes_home; '
+                'print(json.dumps(str(get_hermes_home())))')
+        result = real_run([sys.executable, "-c", code, json.dumps(argv[1:])],
+                          env=kwargs["env"], capture_output=True, text=True, timeout=30)
+        assert result.returncode == 0, result.stderr
+        seen.append(Path(json.loads(result.stdout.strip().splitlines()[-1])))
+        return subprocess.CompletedProcess(argv, 0, "", "")
+
+    monkeypatch.setattr("tools.bot_live_delivery.find_canonical_live_owner", discover)
+    monkeypatch.setattr(delivery.shutil, "which", lambda _: "/bin/hermes")
+    monkeypatch.setattr(delivery.subprocess, "run", run)
+    try:
+        assert delivery._deliver_to_bot_chat({"id": "job"}, "output", profile) is None
+        assert seen == [home]
+        assert not (tmp_path / ".hermes").exists()
+    finally:
+        reset_hermes_home_override(token)
+
+
+@pytest.mark.parametrize("removed_during_discovery", [False, True])
+def test_missing_destination_never_launches_or_recreates(tmp_path, monkeypatch, removed_during_discovery):
+    root = tmp_path / "custom"
+    root.mkdir()
+    home = root / "profiles" / "beta"
+    if removed_during_discovery:
+        home.mkdir(parents=True)
+    monkeypatch.setenv("HERMES_HOME", str(root))
+    token = set_hermes_home_override(str(root))
+    run = Mock(return_value=subprocess.CompletedProcess([], 0, "", ""))
+
+    def discover(target):
+        if home.exists():
+            home.rmdir()
+        return None
+
+    monkeypatch.setattr("tools.bot_live_delivery.find_canonical_live_owner", discover)
+    monkeypatch.setattr(delivery.subprocess, "run", run)
+    try:
+        error = delivery._deliver_to_bot_chat({"id": "job"}, "output", "beta")
+        assert error is not None and str(home) in error
+        run.assert_not_called()
+        assert not home.exists()
+    finally:
+        reset_hermes_home_override(token)
diff --git a/tests/cron/test_bot_chat_pending.py b/tests/cron/test_bot_chat_pending.py
new file mode 100644
index 0000000000000..4cf3d751a0d06
--- /dev/null
+++ b/tests/cron/test_bot_chat_pending.py
@@ -0,0 +1,104 @@
+"""Only never-started cron delivery may wait for a CLI owner's release."""
+import subprocess
+from pathlib import Path
+from unittest.mock import Mock
+
+import pytest
+
+from cron import bot_chat_delivery as queue
+from cron import scheduler_delivery as delivery
+from hermes_cli.active_sessions import try_acquire_active_session
+from hermes_state import SessionDB
+
+
+@pytest.mark.parametrize("error", [None, subprocess.TimeoutExpired("hermes", 1)])
+def test_cli_owner_deferral_and_attempt_fence(tmp_path, monkeypatch, error):
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    db = SessionDB(db_path=tmp_path / "state.db")
+    db.create_session(session_id="chat", source="cli")
+    db.set_session_title("chat", "Bot Chat")
+    lease, refusal = try_acquire_active_session(session_id="chat", surface="cli", config={}, registry_home=tmp_path)
+    assert refusal is None and lease is not None
+    run = Mock(side_effect=error, return_value=subprocess.CompletedProcess([], 0, "", ""))
+    monkeypatch.setattr(delivery.subprocess, "run", run)
+    monkeypatch.setattr(delivery.shutil, "which", lambda _: "/bin/hermes")
+    job = {"id": "job", "execution_id": "execution"}
+    try:
+        assert "queued" in delivery._deliver_to_bot_chat(job, "output", "")
+        key = job["_bot_chat_delivery_receipts"]["bot-chat:(own)"]["delivery_id"]
+        queue.drain()
+        run.assert_not_called()
+        lease.release()
+        queue.drain()
+        assert run.call_count == 1
+        expected = "ambiguous" if error else "settled"
+        assert queue.read_pending(key)["status"] == expected
+        queue.drain()
+        delivery._deliver_to_bot_chat(job, "output", "")
+        assert run.call_count == 1
+    finally:
+        lease.release()
+        db.close()
+
+
+def test_delivery_exception_retains_attempt_and_continues_siblings(tmp_path, monkeypatch):
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    blocked_parent = tmp_path / "blocked"
+    blocked_home = blocked_parent / "recipient"
+    blocked_home.mkdir(parents=True)
+    for home in (blocked_home, tmp_path):
+        db = SessionDB(db_path=home / "state.db")
+        db.create_session(session_id="chat", source="cli")
+        db.set_session_title("chat", "Bot Chat")
+        db.close()
+    queue.defer("b" * 64, {"id": "bad"}, "bad output", "", blocked_home)
+    queue.defer("a" * 64, {"id": "good"}, "good output", "", tmp_path)
+    calls = []
+    original_is_dir = Path.is_dir
+    armed = False
+
+    def resolve_cli(_):
+        nonlocal armed
+        armed = True
+        return "/bin/hermes"
+
+    def is_dir(self):
+        if armed and self == blocked_home:
+            raise PermissionError("target traversal denied after discovery")
+        return original_is_dir(self)
+
+    def run(*args, **kwargs):
+        calls.append(kwargs["env"]["HERMES_HOME"])
+        return subprocess.CompletedProcess([], 0, "", "")
+
+    monkeypatch.setattr(delivery.shutil, "which", resolve_cli)
+    monkeypatch.setattr(delivery.subprocess, "run", run)
+    monkeypatch.setattr(Path, "is_dir", is_dir)
+    queue.drain()
+    assert queue.read_pending("b" * 64)["status"] == "ambiguous"
+    assert "PermissionError" in queue.read_pending("b" * 64)["error"]
+    assert queue.read_pending("a" * 64)["status"] == "settled"
+    queue.drain()
+    assert calls == [str(tmp_path)]
+
+
+def test_pending_queue_uses_admission_order_and_keeps_claims(tmp_path, monkeypatch):
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    job = {"id": "job"}
+    queue.defer("f" * 64, job, "older", "", tmp_path)
+    queue.defer("a" * 64, job, "newer", "", tmp_path)
+    seen = []
+
+    def interrupted(job, content, profile, **kwargs):
+        seen.append(content)
+        raise KeyboardInterrupt
+
+    monkeypatch.setattr(delivery, "_deliver_to_bot_chat", interrupted)
+    with pytest.raises(KeyboardInterrupt):
+        queue.drain()
+    assert seen == ["older"]
+    assert queue.read_pending("f" * 64)["status"] == "claimed"
+    monkeypatch.setattr(delivery, "_deliver_to_bot_chat", lambda j, c, p, **kw: seen.append(c))
+    queue.drain()
+    queue.drain()
+    assert seen == ["older", "newer"]
diff --git a/tests/cron/test_bot_chat_pending_identity.py b/tests/cron/test_bot_chat_pending_identity.py
new file mode 100644
index 0000000000000..748a01728f4d4
--- /dev/null
+++ b/tests/cron/test_bot_chat_pending_identity.py
@@ -0,0 +1,83 @@
+"""A deferred delivery keeps its original destination and admission identity."""
+import subprocess
+from pathlib import Path
+from unittest.mock import Mock
+
+import pytest
+
+from cron import bot_chat_delivery as queue
+from cron import scheduler_delivery as delivery
+from hermes_cli.active_sessions import try_acquire_active_session
+from hermes_state import SessionDB
+from tools.bot_live_delivery import read_delivery_result
+
+
+@pytest.mark.parametrize("recipient", ["cli", "desktop", "renamed"])
+def test_deferred_destination_does_not_follow_root_changes(tmp_path, monkeypatch, recipient):
+    source = tmp_path / "source"
+    home = tmp_path / "original" / "profiles" / "beta"
+    other = tmp_path / "other" / "profiles" / "beta"
+    home.mkdir(parents=True)
+    other.mkdir(parents=True)
+    monkeypatch.setenv("HERMES_HOME", str(source))
+    monkeypatch.setattr("hermes_cli.profiles.get_profile_dir", lambda _: home)
+    db = SessionDB(db_path=home / "state.db")
+    db.create_session(session_id="chat", source="cli")
+    db.set_session_title("chat", "Bot Chat")
+    lease, refusal = try_acquire_active_session(
+        session_id="chat", surface="cli", config={}, registry_home=home)
+    assert refusal is None
+    job = {"id": "job", "execution_id": "execution"}
+    try:
+        assert "queued" in delivery._deliver_to_bot_chat(job, "output", "beta")
+        key = job["_bot_chat_delivery_receipts"]["bot-chat:beta"]["delivery_id"]
+    finally:
+        lease.release()
+        db.close()
+    monkeypatch.setattr("hermes_cli.profiles.get_profile_dir", lambda _: other)
+    run = Mock(return_value=subprocess.CompletedProcess([], 0, "", ""))
+    monkeypatch.setattr(delivery.subprocess, "run", run)
+    monkeypatch.setattr(delivery.shutil, "which", lambda _: "/bin/hermes")
+    if recipient == "desktop":
+        lease, refusal = try_acquire_active_session(
+            session_id="chat", surface="desktop", config={}, registry_home=home,
+            metadata={"bot_live_delivery_consumer": True, "live_session_id": "live"})
+        assert refusal is None
+    elif recipient == "renamed":
+        home.rename(home.with_name("renamed"))
+    try:
+        with monkeypatch.context() as changed:
+            changed.setenv("HERMES_HOME", str(tmp_path / "new-source"))
+            queue.drain(source / "cron" / "bot_chat_pending")
+            queue.drain(source / "cron" / "bot_chat_pending")
+        if recipient == "cli":
+            assert run.call_count == 1
+            argv = run.call_args.args[0]
+            assert "-p" not in argv
+            assert Path(run.call_args.kwargs["env"]["HERMES_HOME"]) == home
+        elif recipient == "desktop":
+            run.assert_not_called()
+            receipt = read_delivery_result(home, key)
+            assert receipt is not None and receipt["status"] == "queued"
+            assert queue.read_pending(key)["status"] == "transferred"
+        else:
+            run.assert_not_called()
+            assert not home.exists()
+            assert queue.read_pending(key)["status"] == "ambiguous"
+        assert read_delivery_result(other, key) is None
+    finally:
+        lease.release()
+
+
+def test_corrupt_record_is_retained_without_blocking_other_admissions(tmp_path, monkeypatch):
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    queue.defer("a" * 64, {"id": "job"}, "first", "", tmp_path)
+    broken = tmp_path / "cron" / "bot_chat_pending" / "broken.json"
+    broken.write_text("{", encoding="utf-8")
+    seen = []
+    monkeypatch.setattr(delivery, "_deliver_to_bot_chat", lambda j, c, p, **kw: seen.append(c))
+    queue.drain()
+    queue.defer("b" * 64, {"id": "next"}, "second", "", tmp_path)
+    queue.drain()
+    assert seen == ["first", "second"]
+    assert broken.read_text(encoding="utf-8") == "{"
diff --git a/tests/cron/test_cron_bot_chat_delivery.py b/tests/cron/test_cron_bot_chat_delivery.py
index e8bc4a7983f9a..fbfa7939cb8bc 100644
--- a/tests/cron/test_cron_bot_chat_delivery.py
+++ b/tests/cron/test_cron_bot_chat_delivery.py
@@ -135,7 +135,7 @@ def fake_run(argv, **kwargs):
     assert err is None
     argv = calls["argv"]
     assert argv[0] == "/usr/bin/hermes"
-    assert "-p" not in argv  # own profile: subprocess inherits HERMES_HOME
+    assert argv[1:3] == ["-p", "default"]  # do not follow active_profile
     assert "chat" in argv
     assert "Bot Chat" in argv
     assert "--create-if-missing" in argv
@@ -145,26 +145,6 @@ def fake_run(argv, **kwargs):
     assert not any("the output" in str(a) for a in argv)
 
 
-def test_deliver_named_profile_uses_p_flag_and_clears_home():
-    calls = {}
-
-    def fake_run(argv, **kwargs):
-        calls["argv"] = argv
-        calls["kwargs"] = kwargs
-        return _completed()
-
-    with mock.patch.object(sched.subprocess, "run", side_effect=fake_run), \
-         mock.patch.object(sched_delivery.shutil, "which", return_value="/usr/bin/hermes"), \
-         mock.patch.dict(sched.os.environ, {"HERMES_HOME": "/tmp/other-profile"}):
-        err = _deliver_to_bot_chat({"id": "j1", "name": "n"}, "out", "research")
-
-    assert err is None
-    argv = calls["argv"]
-    assert argv[1:3] == ["-p", "research"]
-    # -p owns resolution; the scheduler's own HERMES_HOME must not leak in.
-    assert "HERMES_HOME" not in calls["kwargs"]["env"]
-
-
 def test_deliver_failure_returns_error_string():
     with mock.patch.object(
         sched.subprocess, "run", return_value=_completed(returncode=1, stderr="boom")
diff --git a/tools/bot_live_delivery.py b/tools/bot_live_delivery.py
index 54d71c4d2b351..ebec4efeecd57 100644
--- a/tools/bot_live_delivery.py
+++ b/tools/bot_live_delivery.py
@@ -25,12 +25,8 @@
 _TERMINAL = frozenset({"settled", "failed", "cancelled", "ambiguous"})
 
 
-def find_canonical_live_owner(profile_home: Path | str) -> dict[str, Any] | None:
-    """Resolve exact Bot Chat's compression tip without creating/migrating its DB.
-
-    Capability advertisement is mandatory; old Desktop/TUI processes must not
-    receive work they cannot consume. Registry errors propagate, failing closed.
-    """
+def find_canonical_owner(profile_home: Path | str) -> dict[str, Any] | None:
+    """Return the exact Bot Chat tip's lease, including unsupported CLI owners."""
     from hermes_cli.active_sessions import active_session_registry_snapshot
     from hermes_state import SessionDB
 
@@ -46,12 +42,18 @@ def find_canonical_live_owner(profile_home: Path | str) -> dict[str, Any] | None
     if not session_id:
         return None
     for entry in active_session_registry_snapshot(registry_home=home):
-        meta = entry.get("metadata") or {}
-        if (entry["session_id"] == session_id
-                and meta.get("bot_live_delivery_consumer") is True
-                and meta.get("live_session_id")):
-            return dict(profile_home=str(home), session_id=session_id,
-                        lease_id=entry["lease_id"], live_session_id=meta["live_session_id"])
+        if entry["session_id"] == session_id:
+            return {**entry, "profile_home": str(home)}
+    return None
+
+
+def find_canonical_live_owner(profile_home: Path | str) -> dict[str, Any] | None:
+    """Only advertised consumers may receive owner-pinned mailbox deliveries."""
+    entry = find_canonical_owner(profile_home)
+    meta = (entry or {}).get("metadata") or {}
+    if entry and meta.get("bot_live_delivery_consumer") is True and meta.get("live_session_id"):
+        return {key: entry[key] for key in ("profile_home", "session_id", "lease_id")} | {
+            "live_session_id": meta["live_session_id"]}
     return None
 
 
diff --git a/website/docs/user-guide/features/cron.md b/website/docs/user-guide/features/cron.md
index 73d96cdd54beb..c9f691deb85a9 100644
--- a/website/docs/user-guide/features/cron.md
+++ b/website/docs/user-guide/features/cron.md
@@ -536,7 +536,8 @@ error. A delivery failure does not count toward the job's `failure_streak`
 - `bot-chat:<profile>` targets another profile **on the same machine**. Names are validated against `hermes profile list` when the job is created; profiles on other gateways or machines can never be targeted, so same-named profiles across machines are unambiguous.
 - Each delivery costs the target bot one full agent turn — mind the schedule frequency.
 - Composes with other targets (`bot-chat,telegram`) but is never included in `all`.
-- If the canonical chat is open in a mailbox-capable Desktop/TUI backend, delivery is **durably queued immediately**, whether the bot is idle or busy. Only that live owner runs the incoming turn; cron does not start a competing CLI writer. Without a live mailbox owner, the existing `hermes chat -c "Bot Chat" --create-if-missing` lane remains available (normal session ownership checks still apply).
+- If the canonical chat is open in a mailbox-capable Desktop/TUI backend, delivery is **durably queued immediately**, whether the bot is idle or busy. Only that live owner runs the incoming turn; cron does not start a competing CLI writer. If a CLI-only or older unsupported owner holds the chat, cron retains the never-started output under the sending profile's `cron/bot_chat_pending/<receipt-id>.json`. Later scheduler ticks deliver after that owner releases the chat, in admission order. Deferred work retains its admitted destination home and receipt ID even if the scheduler's launch root changes; a missing/renamed destination is not recreated or resolved to another profile. A `transferred` pending record points to the live-owner receipt, not a failed turn. Malformed JSON records are retained and logged without blocking other queued outputs. With no owner, the existing `hermes chat -c "Bot Chat" --create-if-missing` lane remains available (normal session ownership checks still apply). That child uses the exact destination home already checked by cron, including custom roots; inherited `HOME` or a changed active profile cannot redirect it. A missing destination directory is refused before launch, not recreated. A deferred request is claimed before launching that lane; interruption or an uncertain subprocess result never causes an automatic resend.
+- Never-started outputs have no TTL: if an unsupported owner never releases, they remain queued rather than being silently dropped. Receipts retain their payloads indefinitely. An unexpected delivery exception is logged and retained as `ambiguous`, without stopping sibling deliveries in that drain; claimed/ambiguous attempts are never automatically replayed.
 - **Queued is not completed.** Cron records receipt IDs and `queued`/`claimed` statuses in `last_delivery_queued`, with delivery outcome `queued` (neither delivered nor failed). A successful job shows `delivery_queued`; genuine errors on other targets still take precedence as delivery failures. The bot may complete later. The durable receipt in the target profile's `runtime/bot_live_delivery/<receipt-id>.json` is authoritative; cron's historical status is not automatically refreshed.
 - Rechecking the same execution inspects its existing receipt, even if the owner has disappeared. It never falls back to another writer after acceptance. `failed`, `cancelled`, or `ambiguous` receipts are not automatically replayed; inspect the chat and receipt before intentionally starting new work. Each new cron execution has a distinct delivery ID.