diff --git a/cron/bot_chat_delivery.py b/cron/bot_chat_delivery.py deleted file mode 100644 index 5d38f96f24..0000000000 --- a/cron/bot_chat_delivery.py +++ /dev/null @@ -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() diff --git a/evals/botmode-dm-delivery/README.md b/evals/botmode-dm-delivery/README.md index 7d4b9bbdb8..414a8ba42f 100644 --- a/evals/botmode-dm-delivery/README.md +++ b/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) diff --git a/evals/botmode-dm-delivery/exception-tick.py b/evals/botmode-dm-delivery/exception-tick.py deleted file mode 100644 index c0482dac9d..0000000000 --- a/evals/botmode-dm-delivery/exception-tick.py +++ /dev/null @@ -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") diff --git a/tests/cron/test_bot_chat_cli_home.py b/tests/cron/test_bot_chat_cli_home.py deleted file mode 100644 index dadcf5f923..0000000000 --- a/tests/cron/test_bot_chat_cli_home.py +++ /dev/null @@ -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) diff --git a/tests/cron/test_bot_chat_pending.py b/tests/cron/test_bot_chat_pending.py deleted file mode 100644 index 8aa2096d6e..0000000000 --- a/tests/cron/test_bot_chat_pending.py +++ /dev/null @@ -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"] diff --git a/tests/cron/test_bot_chat_pending_identity.py b/tests/cron/test_bot_chat_pending_identity.py deleted file mode 100644 index bdb044fdb8..0000000000 --- a/tests/cron/test_bot_chat_pending_identity.py +++ /dev/null @@ -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") == "{" diff --git a/tools/bot_live_delivery.py b/tools/bot_live_delivery.py index 530362fa3c..c7fcfbd56a 100644 --- a/tools/bot_live_delivery.py +++ b/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.