From 11d6448b474ba9a81a104ad76eed577c3ad1f0da Mon Sep 17 00:00:00 2001 From: sefsefsef10 Date: Thu, 10 Sep 2026 11:30:24 -0500 Subject: [PATCH 1/5] fix(cron): retry bot-chat delivery when the recipient is mid-turn A bot-chat delivery refused because the recipient's Bot Chat already has a live owner was treated as a terminal failure and dropped. Bot-chat is deliberately excluded from the durable delivery queue in _deliver_result (a bot turn costs money and must never be replayed by a second writer), so that drop is total: the payload exists nowhere afterwards. The sender records delivery_outcome=failed, the recipient never learns anything was sent, and a scheduled finding is lost silently. SESSION_NOT_OWNED is a typed CAPACITY refusal -- active_sessions.py documents it as "busy, come back later" -- so the correct response is to wait, not to give up. Reproduced live: two concurrent `hermes chat -c "Bot Chat"` sends against one profile return exit 1 / SESSION_NOT_OWNED for the second, which is exactly what a bot routine firing during another bot's delivery turn produces. Observed in the field with two routines 30 minutes apart whose recipient turn ran 15 minutes. Retry only the typed capacity refusals (SESSION_NOT_OWNED, MAX_CONCURRENT_SESSIONS) with exponential backoff; genuine failures stay terminal and loud on the first attempt, and an exhausted ladder still returns the refusal rather than a phantom success. Defaults (3 retries; 30s doubling to 30/60/120s) are sized for an agent turn and live in config.yaml, not env vars; bot_chat_busy_retries: 0 restores the previous drop-on-busy behaviour. Tests: a busy recipient is retried and delivers; a genuine error is attempted exactly once; a persistently busy recipient reports the refusal instead of succeeding. --- cron/scheduler_delivery.py | 73 ++++++++++++++++++++--- hermes_cli/config_defaults.py | 10 ++++ tests/cron/test_cron_bot_chat_delivery.py | 69 +++++++++++++++++++++ 3 files changed, 143 insertions(+), 9 deletions(-) diff --git a/cron/scheduler_delivery.py b/cron/scheduler_delivery.py index 92f99b4c9e853..c4b50fae172e2 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,35 @@ 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 _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 +762,19 @@ 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. + # Genuine errors are NOT retried — they stay terminal and loud. + _CAPACITY_REFUSALS = ("SESSION_NOT_OWNED", "MAX_CONCURRENT_SESSIONS") + + def _is_capacity_refusal(text: str) -> bool: + return any(reason in text for reason in _CAPACITY_REFUSALS) + from agent.delegation_context import delegated_child_subprocess_env env = delegated_child_subprocess_env(os.environ) if profile: @@ -754,16 +797,28 @@ 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: + attempts = max(1, _get_bot_chat_busy_retries() + 1) + backoff = _get_bot_chat_busy_backoff_seconds() + for attempt in range(attempts): + 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: + logger.info( + "Job '%s': delivered to Bot Chat of profile '%s'", job_id, profile_label) + return None 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 + last_attempt = attempt == attempts - 1 + if last_attempt or not _is_capacity_refusal(tail): + return _fail( + f"bot-chat delivery to profile '{profile_label}' failed " + f"(exit {result.returncode})" + (f": {tail}" if tail else "")) + delay = backoff * (2 ** attempt) + logger.info( + "Job '%s': Bot Chat of profile '%s' busy; retrying in %.1fs (%d/%d)", + job_id, profile_label, delay, attempt + 1, attempts - 1) + time.sleep(delay) except subprocess.TimeoutExpired: return _fail( f"bot-chat delivery to profile '{profile_label}' timed out " diff --git a/hermes_cli/config_defaults.py b/hermes_cli/config_defaults.py index 466d5f99b03e1..90d5212d017ea 100644 --- a/hermes_cli/config_defaults.py +++ b/hermes_cli/config_defaults.py @@ -1623,6 +1623,16 @@ 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, # 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..2379e63365ea8 100644 --- a/tests/cron/test_cron_bot_chat_delivery.py +++ b/tests/cron/test_cron_bot_chat_delivery.py @@ -184,6 +184,75 @@ 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.time, "sleep", lambda _s: None): + 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.time, "sleep", lambda _s: None): + 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.time, "sleep", lambda _s: None): + 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_deliver_message_carries_cron_attribution(tmp_path): """The injected turn must self-identify as scheduled output, not the user.""" captured = {} From b795c4192736f7bc9e4c9e4caac05545af8a20dd Mon Sep 17 00:00:00 2001 From: sefsefsef10 Date: Thu, 10 Sep 2026 11:34:37 -0500 Subject: [PATCH 2/5] fix(cron): classify bot-chat refusals by typed marker, not substring Two defects in the busy-retry gate, both found by adversarial review: 1. Substring matching meant ANY child output containing "SESSION_NOT_OWNED" was retried -- including a job whose own payload or traceback merely mentions the reason name. Parse the `hermes-refusal-reason:` marker line the CLI actually emits (hermes_cli/active_sessions.py) instead. 2. The output was truncated to its last 500 chars BEFORE classification. The refusal marker is printed FIRST, ahead of the CLI's long explanatory text, so a real refusal behind a verbose message was classified as a genuine failure and the payload dropped -- the precise bug the retry exists to prevent. Classify on the full output; truncate only what is displayed. SESSION_COORDINATION_UNAVAILABLE is now explicitly excluded and pinned by a test: it means ownership could not be PROVEN, and retrying an unprovable state is the fail-open hole that lets two writers share one session. Duplicate delivery was investigated and does not occur: a capacity refusal is issued at lease acquisition, before any turn runs. Verified live -- a refused concurrent send wrote 0 rows to the recipient's state.db. A timeout after acceptance exits through `except subprocess.TimeoutExpired` and never reaches the retry path. Tests: 23 pass. Reverting either fix fails exactly the two tests that guard it (mutation-verified). --- cron/scheduler_delivery.py | 28 +++++++-- tests/cron/test_cron_bot_chat_delivery.py | 74 +++++++++++++++++++++++ 2 files changed, 98 insertions(+), 4 deletions(-) diff --git a/cron/scheduler_delivery.py b/cron/scheduler_delivery.py index c4b50fae172e2..6e17730a0f5fc 100644 --- a/cron/scheduler_delivery.py +++ b/cron/scheduler_delivery.py @@ -769,11 +769,27 @@ def _fail(msg: str, **log_kwargs) -> str: # 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. - _CAPACITY_REFUSALS = ("SESSION_NOT_OWNED", "MAX_CONCURRENT_SESSIONS") + # + # 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: - return any(reason in text for reason in _CAPACITY_REFUSALS) + """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) @@ -808,9 +824,10 @@ def _is_capacity_refusal(text: str) -> bool: logger.info( "Job '%s': delivered to Bot Chat of profile '%s'", job_id, profile_label) return None - tail = (result.stderr or result.stdout or "").strip()[-500:] + output = (result.stderr or result.stdout or "").strip() + tail = output[-500:] last_attempt = attempt == attempts - 1 - if last_attempt or not _is_capacity_refusal(tail): + 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 "")) @@ -853,6 +870,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/tests/cron/test_cron_bot_chat_delivery.py b/tests/cron/test_cron_bot_chat_delivery.py index 2379e63365ea8..8bb9885734460 100644 --- a/tests/cron/test_cron_bot_chat_delivery.py +++ b/tests/cron/test_cron_bot_chat_delivery.py @@ -253,6 +253,80 @@ def fake_run(argv, **kwargs): 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.time, "sleep", lambda _s: None): + 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.time, "sleep", lambda _s: None): + 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.time, "sleep", lambda _s: None): + 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_deliver_message_carries_cron_attribution(tmp_path): """The injected turn must self-identify as scheduled output, not the user.""" captured = {} From ca44cc41153a04acc6389f5d9c9929f8033e8075 Mon Sep 17 00:00:00 2001 From: sefsefsef10 Date: Thu, 10 Sep 2026 11:39:48 -0500 Subject: [PATCH 3/5] fix(cron): bound total wall time of bot-chat busy retries Each retry attempt carries its own delivery timeout, so the retry ladder had no overall ceiling: a recipient whose turns HANG rather than refuse could stack 4x600s of timeout plus 210s of backoff -- ~44 minutes inside a single delivery, holding a cron worker the whole time. Measured live on the real CLI: a capacity refusal returns in ~3.4s (a successful delivery ~9.8s), so the realistic ladder costs ~4 minutes and is unaffected. The ceiling exists for the pathological case only. cron.bot_chat_busy_budget_seconds (default 900) is checked BEFORE starting another sleep, so it never truncates a delivery that is progressing -- it only declines to sleep into time it cannot honour. 0 disables it. Tests: 25 pass. Removing the guard fails exactly the budget test; a fast ladder at the measured 3.4s latency still retries and delivers. --- cron/scheduler_delivery.py | 28 ++++++++++++ hermes_cli/config_defaults.py | 7 +++ tests/cron/test_cron_bot_chat_delivery.py | 55 +++++++++++++++++++++++ 3 files changed, 90 insertions(+) diff --git a/cron/scheduler_delivery.py b/cron/scheduler_delivery.py index 6e17730a0f5fc..57ff1f3ce95a2 100644 --- a/cron/scheduler_delivery.py +++ b/cron/scheduler_delivery.py @@ -683,6 +683,25 @@ def _get_bot_chat_busy_backoff_seconds() -> float: 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 _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. @@ -815,6 +834,8 @@ def _is_capacity_refusal(text: str) -> bool: ] 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() for attempt in range(attempts): result = subprocess.run( argv, capture_output=True, text=True, @@ -832,6 +853,13 @@ def _is_capacity_refusal(text: str) -> bool: 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) diff --git a/hermes_cli/config_defaults.py b/hermes_cli/config_defaults.py index 90d5212d017ea..f6d49e38892c2 100644 --- a/hermes_cli/config_defaults.py +++ b/hermes_cli/config_defaults.py @@ -1633,6 +1633,13 @@ def _aux(timeout, *, reasoning_effort=True, **extra): # 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 8bb9885734460..cfa3d4a53c0eb 100644 --- a/tests/cron/test_cron_bot_chat_delivery.py +++ b/tests/cron/test_cron_bot_chat_delivery.py @@ -327,6 +327,61 @@ def fake_run(argv, **kwargs): assert err is None +def test_busy_retries_stop_at_the_wall_clock_budget(): + """A hanging recipient must not hold a cron worker for the full retry ladder. + + Each attempt carries its own delivery timeout, so without an overall budget a + recipient whose turns hang stacks 4x600s + 210s of backoff (~44 min) inside a + single delivery. The budget is checked BEFORE sleeping, so it bounds the + pathological case without truncating a delivery that is progressing. + """ + 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"] += 600.0 # every attempt hangs to its delivery 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.time, "sleep", lambda s: clock.__setitem__("t", clock["t"] + s)): + err = _deliver_to_bot_chat({"id": "j1", "name": "n"}, "out", "") + + assert len(attempts) < 4, "the budget must stop the ladder before every retry is spent" + assert err is not None + assert "budget" in err.lower() + + +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_deliver_message_carries_cron_attribution(tmp_path): """The injected turn must self-identify as scheduled output, not the user.""" captured = {} From ea2d1f2d4c298946ea076745600be9b58d8d0b19 Mon Sep 17 00:00:00 2001 From: sefsefsef10 Date: Thu, 10 Sep 2026 11:51:39 -0500 Subject: [PATCH 4/5] fix(cron): let a bot-chat retry wait yield to gateway shutdown The busy-retry backoff slept up to 210s in one uninterruptible time.sleep. Gateway stop marks in-flight cron executions interrupted and then drains for agent.cron_drain_timeout (default 30s), so a worker merely WAITING to retry was still sleeping when the drain gave up -- SIGKILLed mid-wait and left wedged instead of cleanly interrupted. The retry that exists to save a delivery could therefore cost a clean shutdown. Sleep in <=1s slices and poll cron.scheduler._is_interrupted (the existing non-destructive peek used elsewhere before deciding what to deliver), keyed by the execution's fire token so a recurring job's later run is not poisoned. Verified live: a 120s wait aborts in 0.00s once the flag is set, well inside the 30s drain. Two review-process notes, both self-inflicted and fixed here: - I first called a helper that does not exist (_is_job_interrupted) behind a bare `except Exception: pass`, which would have swallowed the mistake forever and silently disabled the guard. The except is gone; the real name is _is_interrupted. - Tests stubbing time.sleep as a no-op busy-spun against the real clock once the sleep became sliced. They now patch the _sleep_unless_interrupted seam, which is what the retry path actually calls. Tests: 27 pass in ~6s. Removing the interrupt check makes the shutdown test hang until the runner is killed (exit 124) -- the wedged-worker bug reproduced. --- cron/scheduler_delivery.py | 33 ++++++++++++- tests/cron/test_cron_bot_chat_delivery.py | 56 ++++++++++++++++++++--- 2 files changed, 82 insertions(+), 7 deletions(-) diff --git a/cron/scheduler_delivery.py b/cron/scheduler_delivery.py index 57ff1f3ce95a2..b23b264f4b70f 100644 --- a/cron/scheduler_delivery.py +++ b/cron/scheduler_delivery.py @@ -702,6 +702,27 @@ def _get_bot_chat_busy_budget_seconds() -> float: 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. @@ -863,7 +884,17 @@ def _is_capacity_refusal(text: str) -> bool: logger.info( "Job '%s': Bot Chat of profile '%s' busy; retrying in %.1fs (%d/%d)", job_id, profile_label, delay, attempt + 1, attempts - 1) - time.sleep(delay) + # 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 " diff --git a/tests/cron/test_cron_bot_chat_delivery.py b/tests/cron/test_cron_bot_chat_delivery.py index cfa3d4a53c0eb..041249a185da5 100644 --- a/tests/cron/test_cron_bot_chat_delivery.py +++ b/tests/cron/test_cron_bot_chat_delivery.py @@ -207,7 +207,7 @@ def fake_run(argv, **kwargs): 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, "sleep", lambda _s: None): + 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" @@ -224,7 +224,7 @@ def fake_run(argv, **kwargs): 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, "sleep", lambda _s: None): + 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" @@ -245,7 +245,7 @@ def fake_run(argv, **kwargs): 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, "sleep", lambda _s: None): + 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" @@ -271,7 +271,7 @@ def fake_run(argv, **kwargs): 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, "sleep", lambda _s: None): + 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" @@ -291,7 +291,7 @@ def fake_run(argv, **kwargs): 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, "sleep", lambda _s: None): + 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" @@ -320,7 +320,7 @@ def fake_run(argv, **kwargs): 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, "sleep", lambda _s: None): + 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" @@ -382,6 +382,50 @@ def fake_run(argv, **kwargs): 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 = {} From 99fef31da3fee6b42632c99c1c1956fd3e383e79 Mon Sep 17 00:00:00 2001 From: sefsefsef10 Date: Thu, 10 Sep 2026 11:53:47 -0500 Subject: [PATCH 5/5] fix(cron): make the bot-chat retry budget a real deadline Independent review (different model family) returned BLOCK on ca44cc4115 and was correct. The budget was checked only before SLEEPING, so the next attempt still started with a full per-attempt timeout regardless of time remaining. Reproduced: two attempts at 600s each against a 900s budget = 1230s elapsed, a 330s overshoot of the ceiling the config advertises. The budget is now a deadline. Each attempt's timeout is clamped to the time actually left, and a TimeoutExpired caused by that clamp is reported as budget exhaustion rather than as the recipient hanging for the full configured timeout -- the two are different failures and must not be conflated. The more serious finding was the test, not the code: the previous budget test asserted only `attempts < 4`, which passed while elapsed time was 1230s. It tested a proxy instead of the invariant. It now asserts elapsed <= budget and that a later attempt is clamped, and a second test pins the budget-vs-hang error distinction. Reverting the clamp fails both. Tests: 28 pass. --- cron/scheduler_delivery.py | 33 +++++++++++++-- tests/cron/test_cron_bot_chat_delivery.py | 49 ++++++++++++++++++----- 2 files changed, 68 insertions(+), 14 deletions(-) diff --git a/cron/scheduler_delivery.py b/cron/scheduler_delivery.py index b23b264f4b70f..e8c3b76828398 100644 --- a/cron/scheduler_delivery.py +++ b/cron/scheduler_delivery.py @@ -857,11 +857,36 @@ def _is_capacity_refusal(text: str) -> bool: 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): - result = subprocess.run( - argv, capture_output=True, text=True, - timeout=_get_bot_chat_delivery_timeout(), env=env, - creationflags=windows_hide_flags()) + 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) diff --git a/tests/cron/test_cron_bot_chat_delivery.py b/tests/cron/test_cron_bot_chat_delivery.py index 041249a185da5..8da7cb1b7c601 100644 --- a/tests/cron/test_cron_bot_chat_delivery.py +++ b/tests/cron/test_cron_bot_chat_delivery.py @@ -328,36 +328,65 @@ def fake_run(argv, **kwargs): def test_busy_retries_stop_at_the_wall_clock_budget(): - """A hanging recipient must not hold a cron worker for the full retry ladder. + """Elapsed wall time must not exceed the budget -- the invariant, not attempt count. - Each attempt carries its own delivery timeout, so without an overall budget a - recipient whose turns hang stacks 4x600s + 210s of backoff (~44 min) inside a - single delivery. The budget is checked BEFORE sleeping, so it bounds the - pathological case without truncating a delivery that is progressing. + 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", ) - attempts = [] + timeouts = [] clock = {"t": 0.0} def fake_run(argv, **kwargs): - attempts.append(argv) - clock["t"] += 600.0 # every attempt hangs to its delivery timeout + 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.time, "sleep", lambda s: clock.__setitem__("t", clock["t"] + s)): + 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 len(attempts) < 4, "the budget must stop the ladder before every retry is spent" + 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(