diff --git a/cli-config.yaml.example b/cli-config.yaml.example index b2d09dc5dd..575a03e5c5 100644 --- a/cli-config.yaml.example +++ b/cli-config.yaml.example @@ -1041,8 +1041,8 @@ agent: # Maximum time an alias routing key waits for an active turn holding the same # resolved session lease. On expiry Hermes rejects this inbound message and # asks the user to resend rather than running it without serialization. - # Non-positive values fall back to the 1800-second default. - # gateway_turn_lease_timeout: 1800 + # Non-positive values fall back to the 5-second default. + # gateway_turn_lease_timeout: 5 # Staged warning: send a warning before escalating to full timeout. # Fires once per run when inactivity reaches this threshold (seconds). diff --git a/gateway/run.py b/gateway/run.py index de8c680ace..8264eedc47 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -18552,26 +18552,44 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew # immediately evict them, racing with the setup path. _stale_idle = float("inf") # assume idle if we can't check _stale_detail = "" + _activity_summary_valid = False if _stale_agent and hasattr(_stale_agent, "get_activity_summary"): try: _sa = _stale_agent.get_activity_summary() - _stale_idle = _sa.get("seconds_since_activity", float("inf")) + from gateway.session_stall import ( + resolve_session_idle_seconds_from_activity, + ) + + _resolved_idle = resolve_session_idle_seconds_from_activity( + _sa if isinstance(_sa, dict) else None, + now=time.time(), + ) + if _resolved_idle is not None: + _stale_idle = _resolved_idle + _activity_summary_valid = True _stale_detail = ( - f" | last_activity={_sa.get('last_activity_desc', 'unknown')} " + f" | last_activity={_sa.get('last_activity_desc', 'unknown') if isinstance(_sa, dict) else 'unknown'} " f"({_stale_idle:.0f}s ago) " - f"| iteration={_sa.get('api_call_count', 0)}/{_sa.get('max_iterations', 0)}" + f"| iteration={_sa.get('api_call_count', 0) if isinstance(_sa, dict) else 0}/{_sa.get('max_iterations', 0) if isinstance(_sa, dict) else 0}" ) except Exception: pass - # Evict if: agent is idle beyond timeout, OR wall-clock age is - # extreme (10x timeout or 2h, whichever is larger — catches - # cases where the agent object was garbage-collected). + # A valid activity clock is authoritative: total age alone never + # makes an actively progressing turn stale. The emergency wall TTL + # is only a fallback when the agent cannot report usable activity. _wall_ttl = max(_raw_stale_timeout * 10, 7200) if _raw_stale_timeout > 0 else float("inf") _should_evict = ( _stale_agent is not _AGENT_PENDING_SENTINEL and ( - (_raw_stale_timeout > 0 and _stale_idle >= _raw_stale_timeout) - or _stale_age > _wall_ttl + ( + _activity_summary_valid + and _raw_stale_timeout > 0 + and _stale_idle >= _raw_stale_timeout + ) + or ( + not _activity_summary_valid + and _stale_age > _wall_ttl + ) ) ) if _should_evict: diff --git a/gateway/session_stall.py b/gateway/session_stall.py index f95405283a..76dafcbb9b 100644 --- a/gateway/session_stall.py +++ b/gateway/session_stall.py @@ -85,6 +85,8 @@ def resolve_session_idle_seconds_from_activity( return None elapsed = activity.get("seconds_since_activity") + if isinstance(elapsed, bool): + elapsed = None if elapsed is not None: try: idle = float(elapsed) @@ -102,6 +104,8 @@ def resolve_session_idle_seconds_from_activity( ts = activity.get("last_activity_ts") if ts is None: return None + if isinstance(ts, bool): + return None try: when = float(ts) except (TypeError, ValueError): diff --git a/gateway/turn_lease.py b/gateway/turn_lease.py index a47ad65a43..0f640f9fd3 100644 --- a/gateway/turn_lease.py +++ b/gateway/turn_lease.py @@ -66,7 +66,10 @@ DEFAULT_MAX_LEASES = 512 # HERMES_TURN_LEASE_TIMEOUT bridge because lease contention is not agent # inactivity. A caller that reaches this bound must reject the turn rather than # run it concurrently with the holder. -DEFAULT_LEASE_WAIT = 1800.0 +# Keep contention fail-closed, but never pin a sequential platform updater for +# minutes. A waiter that cannot acquire promptly is rejected with a resend +# notice; it is never authorized to run without serialization. +DEFAULT_LEASE_WAIT = 5.0 class TurnLeaseTimeoutError(TimeoutError): diff --git a/hermes_cli/config_defaults.py b/hermes_cli/config_defaults.py index f0af27e8ca..7990d88ad2 100644 --- a/hermes_cli/config_defaults.py +++ b/hermes_cli/config_defaults.py @@ -62,8 +62,10 @@ DEFAULT_CONFIG = { # Maximum time an alias routing key waits for the active turn holding # the same resolved session lease. On expiry the inbound message is # rejected with a resend notice rather than run without serialization. - # Non-positive values fall back to 1800 seconds. - "gateway_turn_lease_timeout": 1800, + # Keep this short: Telegram dispatches updates sequentially, so an + # inline lease waiter also delays unrelated topics. Non-positive values + # fall back to the five-second safety default. + "gateway_turn_lease_timeout": 5, # Per-session AIAgent cache in the gateway. Each cached agent keeps a # warm prompt prefix AND the session's full transcript, so the cache # trades memory for cost: too small and every turn re-pays an uncached diff --git a/tests/gateway/test_config_env_bridge_authority.py b/tests/gateway/test_config_env_bridge_authority.py index 23824ba69e..594f5182ae 100644 --- a/tests/gateway/test_config_env_bridge_authority.py +++ b/tests/gateway/test_config_env_bridge_authority.py @@ -172,7 +172,7 @@ def test_default_turn_lease_timeout_overrides_stale_env_when_key_is_omitted( env = _run_gateway_import(hermes_home, initial_env={}) - assert env.get("HERMES_TURN_LEASE_TIMEOUT") == "1800" + assert env.get("HERMES_TURN_LEASE_TIMEOUT") == "5" def test_default_turn_lease_timeout_matches_the_runtime_fallback() -> None: diff --git a/tests/gateway/test_running_agent_session_toggles.py b/tests/gateway/test_running_agent_session_toggles.py index e9636bba23..09663b1bab 100644 --- a/tests/gateway/test_running_agent_session_toggles.py +++ b/tests/gateway/test_running_agent_session_toggles.py @@ -135,3 +135,34 @@ async def test_verbose_dispatches_mid_run(monkeypatch): assert "can't run mid-turn" not in (result or "") +@pytest.mark.asyncio +async def test_fresh_ancient_turn_remains_controllable(monkeypatch): + """Total turn age must not evict an agent with fresh activity. + + A long autonomous turn can legitimately run beyond the emergency wall TTL. + Evicting it solely by age forgets the control handle while its session-ID + lease remains held, so /steer and /stop fall into the lease waiter instead + of reaching the live agent. + """ + import time + + runner = _make_runner() + sk = build_session_key(_make_source()) + agent = runner._running_agents[sk] + agent.get_activity_summary.return_value = { + "seconds_since_activity": 3.0, + "last_activity_desc": "receiving stream response", + "api_call_count": 1186, + "max_iterations": 2000, + } + runner._running_agents_ts[sk] = time.time() - 31_471 + runner._handle_verbose_command = AsyncMock(return_value="tool progress: new") + monkeypatch.setenv("HERMES_AGENT_TIMEOUT", "1800") + + result = await runner._handle_message(_make_event("/verbose")) + + runner._handle_verbose_command.assert_awaited_once() + assert runner._running_agents[sk] is agent + assert result == "tool progress: new" + + diff --git a/tests/gateway/test_session_stall_watchdog.py b/tests/gateway/test_session_stall_watchdog.py index 06bae8d2f8..8336212a70 100644 --- a/tests/gateway/test_session_stall_watchdog.py +++ b/tests/gateway/test_session_stall_watchdog.py @@ -454,6 +454,18 @@ def test_resolve_idle_rejects_nonfinite_seconds_since_activity(): assert idle == 15.0 +def test_resolve_idle_rejects_boolean_seconds_and_uses_timestamp(): + now = 1_000_000.0 + idle = resolve_session_idle_seconds_from_activity( + { + "seconds_since_activity": True, + "last_activity_ts": now - 3, + }, + now=now, + ) + assert idle == 3.0 + + def test_session_stall_timeout_in_default_config(): from hermes_cli.config import DEFAULT_CONFIG diff --git a/tests/gateway/test_steer_command.py b/tests/gateway/test_steer_command.py index ea4cd3224c..6dbe368b5c 100644 --- a/tests/gateway/test_steer_command.py +++ b/tests/gateway/test_steer_command.py @@ -13,6 +13,7 @@ interrupting. The gateway runner must: from __future__ import annotations from datetime import datetime +import time from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock @@ -116,6 +117,31 @@ async def test_steer_calls_agent_steer_and_does_not_interrupt(): assert adapter._pending_messages == {} +@pytest.mark.asyncio +@pytest.mark.parametrize("elapsed", [float("nan"), True, "not-a-number"]) +async def test_steer_reaches_ancient_turn_via_fresh_timestamp_fallback( + monkeypatch, elapsed +): + runner, _adapter = _make_runner(_session_entry()) + sk = build_session_key(_make_source()) + running_agent = MagicMock() + running_agent.steer.return_value = True + running_agent.get_activity_summary.return_value = { + "seconds_since_activity": elapsed, + "last_activity_at": time.time() - 3, + "last_activity_desc": "receiving stream response", + } + runner._running_agents[sk] = running_agent + runner._running_agents_ts[sk] = time.time() - 31_471 + monkeypatch.setenv("HERMES_AGENT_TIMEOUT", "1800") + + result = await runner._handle_message(_make_event("/steer pause safely")) + + running_agent.steer.assert_called_once_with("pause safely") + assert runner._running_agents[sk] is running_agent + assert result is not None + + @pytest.mark.asyncio async def test_steer_agent_without_steer_method_falls_back(): """If the running agent somehow lacks the steer() method (older build, diff --git a/tests/gateway/test_turn_lease.py b/tests/gateway/test_turn_lease.py index 552fc98b2d..1295720c1f 100644 --- a/tests/gateway/test_turn_lease.py +++ b/tests/gateway/test_turn_lease.py @@ -24,7 +24,11 @@ from unittest.mock import AsyncMock, MagicMock import pytest -from gateway.turn_lease import SessionTurnLeaseRegistry, TurnLeaseTimeoutError +from gateway.turn_lease import ( + DEFAULT_LEASE_WAIT, + SessionTurnLeaseRegistry, + TurnLeaseTimeoutError, +) def _run(coro): @@ -96,6 +100,16 @@ def test_distinct_sessions_do_not_contend(): # --------------------------------------------------------------------------- +def test_default_wait_cannot_head_of_line_block_platform_updates_for_minutes(): + """A contended topic must fail/queue promptly, not pin Telegram's updater. + + Telegram dispatches updates sequentially. Awaiting a held session lease for + 1,800 seconds blocks unrelated topics behind the waiter even though their + sessions do not share a transcript. + """ + assert DEFAULT_LEASE_WAIT == 5.0 + + def test_timeout_fails_closed_instead_of_authorizing_an_unserialized_turn(): """A timed-out waiter must never run against the still-live holder. diff --git a/website/docs/user-guide/configuration.md b/website/docs/user-guide/configuration.md index 89f6e78e8f..3989fc20f9 100644 --- a/website/docs/user-guide/configuration.md +++ b/website/docs/user-guide/configuration.md @@ -981,14 +981,14 @@ lease wait independently of the ordinary agent inactivity timeout: ```yaml agent: - gateway_turn_lease_timeout: 1800 + gateway_turn_lease_timeout: 5 ``` If another turn still holds the session lease when this budget expires, Hermes fails closed: it does not load the transcript or run the model for the waiting message. The user receives a rejection notice and must resend. Hermes does not automatically requeue the message because doing so without durable ordering and -idempotency could process it twice. Non-positive values use the 1800-second +idempotency could process it twice. Non-positive values use the 5-second default. ## Session Stall Watchdog