chore(cron): retire the deferred hermes-chat pending lane main kept evolving

cron/bot_chat_delivery.py, find_canonical_owner and their tests only served the
subprocess delivery path this branch replaced with the authority door; the merge
brought them back as dead code with no production caller.

cron/bot_chat_delivery.py

@@ -1,128 +0,0 @@
-"""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()

evals/botmode-dm-delivery/README.md

@@ -61,20 +61,12 @@ 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.
+Retired with the deferred `hermes chat` pending lane (`cron/bot_chat_delivery.py`): cron
+output now reaches Bot Chat through the gateway authority door (`gateway/session_bot.py`),
+which records a per-target receipt on the job and creates the Bot Chat when missing, so a
+per-record file-scan exception has no counterpart. The regression case lives in
+`tests/cron/test_cron_bot_chat_create_if_missing.py` and
+`tests/gateway/test_cron_canonical_receipt_regression.py`.
 
 ## Ordinary custom-root fallback (#104066 / #104055)
 

evals/botmode-dm-delivery/exception-tick.py

@@ -1,41 +0,0 @@
-"""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")

tests/cron/test_bot_chat_cli_home.py

@@ -1,79 +0,0 @@
-"""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())))')
-        # Everything after the launcher (binary or ``python -m hermes_cli.main``) is the CLI argv.
-        cli_argv = argv[3:] if argv[1:3] == ["-m", "hermes_cli.main"] else argv[1:]
-        result = real_run([sys.executable, "-c", code, json.dumps(cli_argv)],
-                          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.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)

tests/cron/test_bot_chat_pending.py

@@ -1,106 +0,0 @@
-"""Only never-started cron delivery may wait for a CLI owner's release."""
-import importlib.util
-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(name, *args, **kwargs):
-        # Delivery resolves the CLI (the running install's ``hermes_cli``) right after discovery.
-        nonlocal armed
-        armed = True
-        return object()
-
-    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(importlib.util, "find_spec", 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"]

tests/cron/test_bot_chat_pending_identity.py

@@ -1,82 +0,0 @@
-"""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)
-    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") == "{"

tools/bot_live_delivery.py

@@ -38,23 +38,6 @@ def _bot_chat_tip(home: Path) -> str | None:
         db.close()
 
 
-def find_canonical_owner(profile_home: Path | str) -> dict[str, Any] | None:
-    """Return the exact Bot Chat tip's active-session lease, including unsupported CLI owners.
-
-    A CLI process holding the tip live cannot consume authority deliveries (the door refuses with
-    ``runtime_coordination_required``); cron uses this to retain output until that owner releases."""
-    from hermes_cli.active_sessions import active_session_registry_snapshot
-
-    home = Path(profile_home).resolve()
-    session_id = _bot_chat_tip(home)
-    if not session_id:
-        return None
-    for entry in active_session_registry_snapshot(registry_home=home):
-        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:
     """Discover the profile authority and its exact Bot Chat without acquiring a lease.