fix(gateway): keep long turns controllable without blocking Telegram
This commit is contained in:
@@ -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).
|
||||
|
||||
+26
-8
@@ -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:
|
||||
|
||||
@@ -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):
|
||||
|
||||
@@ -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):
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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"
|
||||
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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.
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user