diff --git a/cron/scheduler_delivery.py b/cron/scheduler_delivery.py index 92f99b4c9e853..e8c3b76828398 100644 --- a/cron/scheduler_delivery.py +++ b/cron/scheduler_delivery.py @@ -17,6 +17,7 @@ import shutil import subprocess import sys +import time from dataclasses import dataclass from typing import Any, List, Optional @@ -653,6 +654,75 @@ def _get_bot_chat_delivery_timeout() -> int: return 600 +def _get_bot_chat_busy_retries() -> int: + """Extra attempts when the recipient's Bot Chat is BUSY (not when it errors). + + ``cron.bot_chat_busy_retries``; default 3. Bot-chat has no durable delivery + queue, so without this a finding delivered while the recipient is mid-turn is + lost outright. 0 restores the old drop-on-busy behaviour. + """ + try: + cfg = _sched.load_config() + value = int(cfg.get("cron", {}).get("bot_chat_busy_retries", 3)) + return max(0, value) + except Exception: + return 3 + + +def _get_bot_chat_busy_backoff_seconds() -> float: + """Base seconds for exponential backoff between busy retries. + + ``cron.bot_chat_busy_backoff_seconds``; default 30. Doubles per attempt, so + the default ladder waits 30s, 60s, 120s — sized for a recipient's agent turn. + """ + try: + cfg = _sched.load_config() + value = float(cfg.get("cron", {}).get("bot_chat_busy_backoff_seconds", 30.0)) + return value if value > 0 else 30.0 + except Exception: + return 30.0 + + +def _get_bot_chat_busy_budget_seconds() -> float: + """Overall wall-clock ceiling for one bot-chat delivery, retries included. + + ``cron.bot_chat_busy_budget_seconds``; default 900. A refusal returns in + seconds (measured ~3.4s), so the realistic retry ladder costs ~4 minutes. + But each attempt carries its own delivery timeout, so a recipient whose + turns HANG could otherwise stack 4x600s + 210s of backoff ~= 44 minutes in + one delivery. The budget is checked before starting another attempt and + before sleeping, so it bounds the pathological case without truncating a + delivery that is actually progressing. 0 disables the ceiling. + """ + try: + cfg = _sched.load_config() + value = float(cfg.get("cron", {}).get("bot_chat_busy_budget_seconds", 900.0)) + return max(0.0, value) + except Exception: + return 900.0 + + +def _sleep_unless_interrupted( + delay: float, job_id: str, token: Optional[object] = None, +) -> bool: + """Sleep ``delay`` in slices; return True if shutdown interrupted this execution. + + Gateway stop marks in-flight cron executions interrupted and drains for + ``agent.cron_drain_timeout`` (default 30s). A bare ``time.sleep`` of up to + 210s ignores that signal, so the worker is still sleeping when the drain + gives up and is SIGKILLed — leaving a wedged job instead of a cleanly + interrupted one. Slicing lets the ladder abandon itself promptly. + """ + deadline = time.monotonic() + delay + while True: + remaining = deadline - time.monotonic() + if remaining <= 0: + return False + if _sched._is_interrupted(job_id, token): + return True + time.sleep(min(1.0, remaining)) + + def _deliver_to_bot_chat(job: dict, content: str, profile: str) -> Optional[str]: """Hand output to the live Bot Chat owner, or use the legacy unowned CLI lane. @@ -732,6 +802,35 @@ def _fail(msg: str, **log_kwargs) -> str: logger.warning("Job '%s': %s", job_id, msg, **log_kwargs) return msg + # A live Bot Chat owner refuses a concurrent write with the typed + # SESSION_NOT_OWNED refusal, which means "busy, come back later" — capacity, + # not failure. Bot-chat is deliberately excluded from the durable delivery + # queue below (a bot turn costs money and must not be replayed by a second + # writer), so a dropped refusal loses the payload with no retry anywhere: + # a scheduled bot finding vanishes while the sender records a bare failure. + # Retrying here is the only place that recoverable refusal can be honoured. + # A refusal is issued at lease acquisition, BEFORE any turn runs, so it + # writes nothing to the recipient and a retry cannot duplicate a delivery. + # Genuine errors are NOT retried — they stay terminal and loud. + # + # SESSION_COORDINATION_UNAVAILABLE is deliberately NOT here: it means + # ownership could not be PROVEN, and retrying an unprovable state is the + # fail-open hole that lets two writers share one session. + _CAPACITY_REFUSALS = frozenset({"SESSION_NOT_OWNED", "MAX_CONCURRENT_SESSIONS"}) + + def _is_capacity_refusal(text: str) -> bool: + """True only when the CLI emitted a typed capacity refusal marker. + + Parses the ``hermes-refusal-reason:`` line rather than substring-matching + the output: a job whose own payload merely mentions a reason name must + never be mistaken for a refusal. + """ + for line in text.splitlines(): + line = line.strip() + if line.startswith(_REFUSAL_REASON_PREFIX): + return line.removeprefix(_REFUSAL_REASON_PREFIX).strip() in _CAPACITY_REFUSALS + return False + from agent.delegation_context import delegated_child_subprocess_env env = delegated_child_subprocess_env(os.environ) if profile: @@ -754,16 +853,73 @@ def _fail(msg: str, **log_kwargs) -> str: "chat", "--in", "~", "-c", "Bot Chat", "--create-if-missing", "-Q", "--query-file", query_file, ] - result = subprocess.run( - argv, capture_output=True, text=True, timeout=_get_bot_chat_delivery_timeout(), env=env, - creationflags=windows_hide_flags()) - 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": {tail}" if tail else "")) - logger.info("Job '%s': delivered to Bot Chat of profile '%s'", job_id, profile_label) - return None + attempts = max(1, _get_bot_chat_busy_retries() + 1) + backoff = _get_bot_chat_busy_backoff_seconds() + budget = _get_bot_chat_busy_budget_seconds() + started = time.monotonic() + # The budget is a DEADLINE, not just a gate on sleeping. Clamping each + # attempt's timeout to the time actually left is what makes the ceiling + # real: checking only before a sleep lets the NEXT attempt start with a + # full per-attempt timeout and overshoot by up to that timeout again + # (measured: 1230s elapsed against a 900s budget). + for attempt in range(attempts): + per_attempt = _get_bot_chat_delivery_timeout() + if budget: + remaining = budget - (time.monotonic() - started) + if remaining <= 0: + return _fail( + f"bot-chat delivery to profile '{profile_label}' gave up after " + f"{time.monotonic() - started:.0f}s (budget {budget:.0f}s); " + f"recipient stayed busy") + per_attempt = min(per_attempt, remaining) + budget_deadline = budget and per_attempt < _get_bot_chat_delivery_timeout() + try: + result = subprocess.run( + argv, capture_output=True, text=True, + timeout=per_attempt, env=env, + creationflags=windows_hide_flags()) + except subprocess.TimeoutExpired: + # A timeout caused by the BUDGET deadline is budget exhaustion, not + # evidence the recipient hung for the full configured timeout. + if budget_deadline: + return _fail( + f"bot-chat delivery to profile '{profile_label}' gave up after " + f"{time.monotonic() - started:.0f}s (budget {budget:.0f}s); " + f"recipient stayed busy") + raise + if result.returncode == 0: + logger.info( + "Job '%s': delivered to Bot Chat of profile '%s'", job_id, profile_label) + return None + output = (result.stderr or result.stdout or "").strip() + tail = output[-500:] + last_attempt = attempt == attempts - 1 + if last_attempt or not _is_capacity_refusal(output): + return _fail( + f"bot-chat delivery to profile '{profile_label}' failed " + f"(exit {result.returncode})" + (f": {tail}" if tail else "")) + delay = backoff * (2 ** attempt) + # Stop before sleeping into a budget we cannot honour: an attempt that + # cannot start is wasted wall time held by the worker for nothing. + if budget and (time.monotonic() - started) + delay >= budget: + return _fail( + f"bot-chat delivery to profile '{profile_label}' gave up after " + f"{time.monotonic() - started:.0f}s (budget {budget:.0f}s); " + f"recipient stayed busy" + (f": {tail}" if tail else "")) + logger.info( + "Job '%s': Bot Chat of profile '%s' busy; retrying in %.1fs (%d/%d)", + job_id, profile_label, delay, attempt + 1, attempts - 1) + # Sleep in slices and abandon the ladder when shutdown marks this + # execution interrupted. Gateway stop drains cron for + # agent.cron_drain_timeout (default 30s); an uninterruptible sleep of + # up to 210s outlives that drain, so the worker gets SIGKILLed while + # merely WAITING to retry and the job is left wedged rather than + # cleanly interrupted. + if _sleep_unless_interrupted(delay, job_id, job.get("_fire_token")): + return _fail( + f"bot-chat delivery to profile '{profile_label}' abandoned: " + f"shutdown interrupted the retry wait" + + (f" (last: {tail})" if tail else "")) except subprocess.TimeoutExpired: return _fail( f"bot-chat delivery to profile '{profile_label}' timed out " @@ -798,6 +954,9 @@ def _normalize_deliver_value(deliver) -> str: # ``bot-chat`` = own profile; ``bot-chat:`` = named profile on THIS machine. BOT_CHAT_PLATFORM = "bot-chat" +# Marker the CLI prints ahead of a typed refusal reason (hermes_cli/active_sessions.py). +_REFUSAL_REASON_PREFIX = "hermes-refusal-reason:" + def parse_bot_chat_deliver_token(part: str) -> Optional[str]: """``bot-chat[:]`` → ``""`` (own profile), the name, or ``None`` if not a bot-chat diff --git a/hermes_cli/config_defaults.py b/hermes_cli/config_defaults.py index 466d5f99b03e1..f6d49e38892c2 100644 --- a/hermes_cli/config_defaults.py +++ b/hermes_cli/config_defaults.py @@ -1623,6 +1623,23 @@ def _aux(timeout, *, reasoning_effort=True, **extra): # created this way are user-owned in the same flat jobs table. Interactive toolsets # (messaging/clarify) stay denied in cron regardless. "allow_agent_scheduling": False, + # Bot Chat has no durable delivery queue: a bot turn costs money and must never be + # replayed by a second writer, so a delivery refused because the recipient is + # MID-TURN is otherwise dropped and the finding is lost with only a bare + # delivery_outcome=failed on the sender. These retry only the typed capacity + # refusals (SESSION_NOT_OWNED / MAX_CONCURRENT_SESSIONS); genuine errors stay + # terminal and loud. 0 restores the old drop-on-busy behaviour. + "bot_chat_busy_retries": 3, + # Base seconds between busy retries; doubles per attempt (30s, 60s, 120s), sized + # for a recipient agent turn rather than a network blip. + "bot_chat_busy_backoff_seconds": 30.0, + # Overall wall-clock ceiling for one bot-chat delivery including retries. A refusal + # returns in seconds, so the realistic ladder costs ~4 min; but each attempt carries + # its own delivery timeout, so a recipient whose turns HANG could otherwise stack + # 4x600s + 210s of backoff (~44 min) inside a single delivery while holding a cron + # worker. Checked before starting another attempt, so it never truncates a delivery + # that is progressing. 0 disables the ceiling. + "bot_chat_busy_budget_seconds": 900.0, # Pre-dispatch validation: before building any agent machinery, verify the provider API key # resolves (unless a fallback chain exists), attached skills are ready, and delivery # platforms are configured. Failure -> last_status=blocked_config, ONE alert, no LLM call. diff --git a/tests/cron/test_cron_bot_chat_delivery.py b/tests/cron/test_cron_bot_chat_delivery.py index e8bc4a7983f9a..8da7cb1b7c601 100644 --- a/tests/cron/test_cron_bot_chat_delivery.py +++ b/tests/cron/test_cron_bot_chat_delivery.py @@ -184,6 +184,277 @@ def test_deliver_timeout_returns_error_string(): assert "timed out" in err +def test_deliver_retries_when_recipient_session_is_busy(): + """A busy recipient is capacity, not failure: retry rather than drop the finding. + + Bot-chat is excluded from the durable delivery queue, so a refusal that is + discarded loses the payload permanently. `SESSION_NOT_OWNED` is the typed + "busy, come back later" refusal a live Bot Chat owner raises while its turn + runs; treating it as terminal is what silently lost scheduled bot findings. + """ + busy = _completed( + returncode=1, + stderr=( + "hermes-refusal-reason: SESSION_NOT_OWNED\n" + "Session 20260101_000000_abcdef already has a live owner (cli, pid 1)." + ), + ) + attempts = [] + + def fake_run(argv, **kwargs): + attempts.append(argv) + return busy if len(attempts) == 1 else _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.object(sched_delivery, "_sleep_unless_interrupted", lambda *_a, **_k: False): + err = _deliver_to_bot_chat({"id": "j1", "name": "n"}, "out", "") + + assert len(attempts) == 2, "a busy recipient must be retried, not dropped" + assert err is None + + +def test_deliver_does_not_retry_a_genuine_failure(): + """Only capacity refusals retry; a real error must stay terminal and loud.""" + attempts = [] + + def fake_run(argv, **kwargs): + attempts.append(argv) + return _completed(returncode=1, stderr="boom") + + 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.object(sched_delivery, "_sleep_unless_interrupted", lambda *_a, **_k: False): + err = _deliver_to_bot_chat({"id": "j1", "name": "n"}, "out", "") + + assert len(attempts) == 1, "a non-capacity failure must not be retried" + assert err is not None and "boom" in err + + +def test_deliver_reports_error_when_recipient_stays_busy(): + """Exhausted retries must surface the refusal, never report a phantom success.""" + busy = _completed( + returncode=1, + stderr="hermes-refusal-reason: SESSION_NOT_OWNED\nSession busy.", + ) + attempts = [] + + def fake_run(argv, **kwargs): + attempts.append(argv) + return busy + + 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.object(sched_delivery, "_sleep_unless_interrupted", lambda *_a, **_k: False): + err = _deliver_to_bot_chat({"id": "j1", "name": "n"}, "out", "") + + assert len(attempts) > 1, "a persistently busy recipient must be retried" + assert err is not None + assert "SESSION_NOT_OWNED" in err or "busy" in err.lower() + + +def test_payload_mentioning_a_refusal_reason_is_not_retried(): + """A reason NAME inside ordinary output must not be mistaken for a refusal. + + Substring-matching the child's output would retry any job whose payload or + traceback merely mentions SESSION_NOT_OWNED. Only the typed + `hermes-refusal-reason:` marker line means the CLI actually refused. + """ + attempts = [] + + def fake_run(argv, **kwargs): + attempts.append(argv) + return _completed( + returncode=1, + stderr="ValueError: audit found SESSION_NOT_OWNED in 3 log lines", + ) + + 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.object(sched_delivery, "_sleep_unless_interrupted", lambda *_a, **_k: False): + err = _deliver_to_bot_chat({"id": "j1", "name": "n"}, "out", "") + + assert len(attempts) == 1, "output merely naming a reason must not be retried" + assert err is not None + + +def test_coordination_unavailable_is_not_retried(): + """Unprovable ownership is not capacity: retrying it is the fail-open hole.""" + attempts = [] + + def fake_run(argv, **kwargs): + attempts.append(argv) + return _completed( + returncode=1, + stderr="hermes-refusal-reason: SESSION_COORDINATION_UNAVAILABLE\nregistry unreadable", + ) + + 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.object(sched_delivery, "_sleep_unless_interrupted", lambda *_a, **_k: False): + err = _deliver_to_bot_chat({"id": "j1", "name": "n"}, "out", "") + + assert len(attempts) == 1, "SESSION_COORDINATION_UNAVAILABLE must never retry" + assert err is not None + + +def test_capacity_refusal_is_detected_behind_a_long_preamble(): + """Classification reads the whole output; only the DISPLAYED error is truncated. + + The refusal marker is printed before the CLI's long explanatory text, so + truncating to the last 500 chars before matching loses it and drops the + payload — the exact bug this retry exists to prevent. + """ + stderr = ( + "hermes-refusal-reason: SESSION_NOT_OWNED\n" + + "Session already has a live owner. " * 40 + ) + assert len(stderr) > 500 + attempts = [] + + def fake_run(argv, **kwargs): + attempts.append(argv) + return busy if len(attempts) == 1 else _completed() + + busy = _completed(returncode=1, stderr=stderr) + + 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.object(sched_delivery, "_sleep_unless_interrupted", lambda *_a, **_k: False): + err = _deliver_to_bot_chat({"id": "j1", "name": "n"}, "out", "") + + assert len(attempts) == 2, "a refusal marker before 500 chars of text must still retry" + assert err is None + + +def test_busy_retries_stop_at_the_wall_clock_budget(): + """Elapsed wall time must not exceed the budget -- the invariant, not attempt count. + + An earlier version of this test asserted only `attempts < 4`. It passed while + the implementation overshot to 1230s against a 900s budget, because each new + attempt was still handed a FULL per-attempt timeout regardless of time left. + Assert the clock, and assert a later attempt is clamped to what remains, or + the budget is advertised without being enforced. + """ + busy = _completed( + returncode=1, + stderr="hermes-refusal-reason: SESSION_NOT_OWNED\nbusy", + ) + timeouts = [] + clock = {"t": 0.0} + + def fake_run(argv, **kwargs): + timeouts.append(kwargs["timeout"]) + # Every attempt hangs for exactly the timeout it was granted. + clock["t"] += kwargs["timeout"] + return busy + + 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.object(sched_delivery.time, "monotonic", lambda: clock["t"]), \ + mock.patch.object(sched_delivery, "_sleep_unless_interrupted", + lambda d, *_a, **_k: (clock.__setitem__("t", clock["t"] + d), False)[1]): + err = _deliver_to_bot_chat({"id": "j1", "name": "n"}, "out", "") + + budget = sched_delivery._get_bot_chat_busy_budget_seconds() + assert clock["t"] <= budget, ( + f"elapsed {clock['t']}s exceeded the advertised budget {budget}s") + assert timeouts[-1] < timeouts[0], ( + "a later attempt must be clamped to the remaining budget, not given a full timeout") + assert err is not None + assert "budget" in err.lower() + + +def test_budget_deadline_timeout_reports_budget_not_recipient_hang(): + """A timeout caused by OUR deadline must not be reported as the recipient hanging.""" + clock = {"t": 0.0} + + def fake_run(argv, **kwargs): + clock["t"] += kwargs["timeout"] + raise subprocess.TimeoutExpired(cmd="hermes", timeout=kwargs["timeout"]) + + 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.object(sched_delivery.time, "monotonic", lambda: clock["t"]), \ + mock.patch.object(sched_delivery, "_get_bot_chat_busy_budget_seconds", lambda: 100.0), \ + mock.patch.object(sched_delivery, "_get_bot_chat_delivery_timeout", lambda: 600), \ + mock.patch.object(sched_delivery, "_sleep_unless_interrupted", + lambda d, *_a, **_k: (clock.__setitem__("t", clock["t"] + d), False)[1]): + err = _deliver_to_bot_chat({"id": "j1", "name": "n"}, "out", "") + + assert err is not None + assert "budget" in err.lower(), "a budget-clamped timeout must be reported as budget exhaustion" + assert clock["t"] <= 100.0 + + +def test_budget_does_not_interrupt_a_fast_retry_ladder(): + """The realistic path (refusals return in seconds) must still retry normally.""" + busy = _completed( + returncode=1, + stderr="hermes-refusal-reason: SESSION_NOT_OWNED\nbusy", + ) + attempts = [] + clock = {"t": 0.0} + + def fake_run(argv, **kwargs): + attempts.append(argv) + clock["t"] += 3.4 # measured live refusal latency + return busy if len(attempts) == 1 else _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.object(sched_delivery.time, "monotonic", lambda: clock["t"]), \ + mock.patch.object(sched_delivery.time, "sleep", lambda s: clock.__setitem__("t", clock["t"] + s)): + err = _deliver_to_bot_chat({"id": "j1", "name": "n"}, "out", "") + + assert err is None + assert len(attempts) == 2 + + +def test_retry_wait_abandons_on_gateway_shutdown(): + """A retry sleep must yield to shutdown, not outlive the cron drain. + + Gateway stop marks in-flight executions interrupted and drains cron for + agent.cron_drain_timeout (default 30s). The backoff ladder sleeps up to 210s, + so a bare time.sleep is still waiting when the drain gives up and the worker + is SIGKILLed -- a wedged job instead of a cleanly interrupted one. + """ + busy = _completed( + returncode=1, + stderr="hermes-refusal-reason: SESSION_NOT_OWNED\nbusy", + ) + attempts = [] + + def fake_run(argv, **kwargs): + attempts.append(argv) + # Shutdown lands while the first attempt is in flight. + sched._interrupted_job_ids.add("j-shutdown") + return busy + + try: + with mock.patch.object(sched.subprocess, "run", side_effect=fake_run), \ + mock.patch.object(sched_delivery.shutil, "which", return_value="/usr/bin/hermes"): + err = _deliver_to_bot_chat({"id": "j-shutdown", "name": "n"}, "out", "") + finally: + sched._interrupted_job_ids.discard("j-shutdown") + + assert len(attempts) == 1, "shutdown must stop the ladder, not run it to exhaustion" + assert err is not None + assert "shutdown" in err.lower() + + +def test_retry_wait_sleeps_normally_without_shutdown(): + """The interruptible sleep must still wait when nothing interrupts it.""" + slept = [] + + with mock.patch.object(sched_delivery.time, "sleep", lambda s: slept.append(s)): + interrupted = sched_delivery._sleep_unless_interrupted(2.5, "quiet-job") + + assert interrupted is False + assert slept, "it must actually sleep when not interrupted" + assert max(slept) <= 1.0, "sleep must be sliced so shutdown is noticed promptly" + + def test_deliver_message_carries_cron_attribution(tmp_path): """The injected turn must self-identify as scheduled output, not the user.""" captured = {}