diff --git a/gateway/run.py b/gateway/run.py index 21ab492bfa36c..e17721c091ae3 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -2366,7 +2366,11 @@ from gateway.delivery import ( looks_like_telegram_private_chat_id, resolve_delivery_transport, ) -from gateway.turn_lease import SessionTurnLeaseRegistry, TurnLeaseTimeoutError +from gateway.turn_lease import ( + DEFAULT_LEASE_WAIT, + SessionTurnLeaseRegistry, + TurnLeaseTimeoutError, +) from gateway.session_state import ( SERVICE_TIER_UNSET as _SERVICE_TIER_UNSET, SessionState, @@ -15738,7 +15742,26 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew _run_generation = self._begin_session_run_generation(_quick_key) try: - _agent_result = await self._handle_message_with_agent(event, source, _quick_key, _run_generation) + try: + _agent_result = await self._handle_message_with_agent( + event, source, _quick_key, _run_generation + ) + except TurnLeaseTimeoutError as exc: + # This is a rejected message, not a completed agent turn. Return + # before the /goal judge below so it cannot consume the resend + # notice and enqueue a synthetic continuation loop. + logger.error( + "Rejecting turn for routing key %s on session %s after " + "turn-lease timeout; transcript load was not started and " + "the user must resend", + _quick_key, + exc.session_id, + ) + return ( + "⏳ Another turn is still running on this session. To " + "protect the transcript, this message was not processed. " + "Wait for the active turn to finish, then resend it." + ) # Goal continuation: after the agent returns a final response # for this turn, check any standing /goal — the judge will # either mark it done, pause it (budget), or enqueue a @@ -16625,8 +16648,9 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew # messages never reach this point mid-turn (adapter + runner guards # hold them), so the lock is uncontended outside the alias-key route. # Fail-closed on timeout: never enter the transcript region without a - # lease. A bounded retry response is safer than recreating the exact - # concurrent-turn corruption this lease exists to prevent. Released + # lease. Outer dispatch returns a bounded rejection/resend notice rather + # than recreating the exact concurrent-turn corruption this lease exists + # to prevent. Released # in _handle_message's finally via _release_turn_lease — granted per # (routing key, run generation) so a stale unwind can't release a # newer turn's lease. @@ -16637,20 +16661,16 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew session_entry.session_id, owner_key=_quick_key, generation=run_generation, - timeout=_float_env("HERMES_AGENT_TIMEOUT", 1800), + timeout=_float_env( + "HERMES_TURN_LEASE_TIMEOUT", DEFAULT_LEASE_WAIT + ), ) except TurnLeaseTimeoutError: - logger.error( - "Deferring turn for routing key %s on session %s after " - "turn-lease timeout; transcript load was not started", - _quick_key, - session_entry.session_id, - ) - return ( - "⏳ Another turn is still running on this session. To " - "protect the transcript, this message was not processed. " - "Wait for the active turn to finish, then resend it." - ) + # The broad session-context cleanup finally starts later in this + # method. Restore the tokens here before propagating the rejection + # to outer dispatch, or this early exit leaks task-local identity. + self._clear_session_env(_session_env_tokens) + raise if _lease_token is not None: _lease_state = self._session_state(_quick_key).turn _lease_state.lease_token = _lease_token diff --git a/gateway/turn_lease.py b/gateway/turn_lease.py index 9b4349f4a270a..26688fdc68d92 100644 --- a/gateway/turn_lease.py +++ b/gateway/turn_lease.py @@ -32,9 +32,9 @@ Safety properties: newer turn's lease (the #28686 ownership lesson applied). Release is idempotent. - **Fail-closed on timeout.** A timed-out waiter raises - :class:`TurnLeaseTimeoutError` and must be deferred by the dispatch layer. - It never runs concurrently against the still-live holder and therefore - cannot defeat the serialization invariant this lease exists to enforce. + :class:`TurnLeaseTimeoutError` and must be rejected by the dispatch layer + with a visible resend notice. It never runs concurrently against the + still-held lease and therefore cannot defeat the serialization invariant. - **Bounded registry.** The per-session lease map is size-capped; eviction only ever removes idle (unheld, uncontended) entries, never a live lease. @@ -61,9 +61,10 @@ logger = logging.getLogger(__name__) # cap rather than break serialization. DEFAULT_MAX_LEASES = 512 -# Fallback wait (seconds) when the caller passes no positive timeout. Matches -# the gateway's default agent inactivity timeout. A caller that reaches this -# bound must defer the turn rather than run it concurrently with the holder. +# Fallback wait (seconds) when the caller passes no positive timeout. The +# gateway exposes this independently as HERMES_TURN_LEASE_TIMEOUT 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 @@ -190,7 +191,7 @@ class SessionTurnLeaseRegistry: Returns a held :class:`TurnLeaseToken`. Raises :class:`TurnLeaseTimeoutError` when the wait budget expires; the caller - must defer rather than enter the serialized region. Returns ``None`` + must reject rather than enter the serialized region. Returns ``None`` for a falsy ``session_id``. """ if not session_id: @@ -223,7 +224,7 @@ class SessionTurnLeaseRegistry: "turn lease wait timed out after %.0fs on session %s " "(waiter: routing key %s gen %s; holder: routing key %s " "gen %s) — failing closed: refusing to run this turn " - "UNSERIALIZED against the still-live holder", + "UNSERIALIZED against the still-held lease", wait, session_id, owner_key, diff --git a/tests/gateway/test_turn_lease.py b/tests/gateway/test_turn_lease.py index f9379f87400c6..5f25b1f4369f5 100644 --- a/tests/gateway/test_turn_lease.py +++ b/tests/gateway/test_turn_lease.py @@ -12,16 +12,18 @@ Covers: - generation-scoped, idempotent release: a stale unwind can never free a newer turn's lease; double-release is a no-op - timeout fail-closed: a timed-out waiter never enters the transcript region, - and the gateway returns a safe retry response before loading history + and outer dispatch returns a visible rejection/resend notice without invoking + goal continuation - registry stays bounded; live leases are never evicted - GatewayRunner._release_turn_lease wiring (bare-runner safe, token-scoped) """ import asyncio +from unittest.mock import AsyncMock, MagicMock import pytest -from gateway.turn_lease import SessionTurnLeaseRegistry +from gateway.turn_lease import SessionTurnLeaseRegistry, TurnLeaseTimeoutError def _run(coro): @@ -100,8 +102,6 @@ def test_timeout_fails_closed_instead_of_authorizing_an_unserialized_turn(): The two turns could then load the same history base and interleave their transcript writes, defeating the serialization invariant this lease owns. """ - from gateway.turn_lease import TurnLeaseTimeoutError - async def scenario(): registry = SessionTurnLeaseRegistry() holder = await registry.acquire( @@ -129,13 +129,14 @@ def test_timeout_fails_closed_instead_of_authorizing_an_unserialized_turn(): @pytest.mark.asyncio -async def test_gateway_defers_timed_out_lease_before_loading_transcript( +async def test_agent_path_propagates_timed_out_lease_before_loading_transcript( monkeypatch, tmp_path ): - """The dispatch layer turns a lease timeout into a safe retry response. + """The agent path propagates timeout before transcript work can begin. - Most importantly, transcript loading and agent execution must not start: - both would operate without the per-session serialization guarantee. + Outer dispatch owns the visible rejection/resend notice. Most importantly, + transcript loading and agent execution must not start: both would operate + without the per-session serialization guarantee. """ from tests.gateway.test_42039_duplicate_user_message import ( _bootstrap, @@ -149,7 +150,7 @@ async def test_gateway_defers_timed_out_lease_before_loading_transcript( "sess-dedup", owner_key="holder-key", generation=1, timeout=1 ) assert holder is not None - monkeypatch.setenv("HERMES_AGENT_TIMEOUT", "0.02") + monkeypatch.setenv("HERMES_TURN_LEASE_TIMEOUT", "0.02") runner.session_store.load_transcript.side_effect = AssertionError( "transcript must not load after a turn-lease timeout" @@ -157,17 +158,56 @@ async def test_gateway_defers_timed_out_lease_before_loading_transcript( runner._run_agent = pytest.fail try: - response = await runner._handle_message_with_agent( - _event(), _source(), "agent:main:telegram:group:-1001:12345", 1 - ) + with pytest.raises(TurnLeaseTimeoutError): + await runner._handle_message_with_agent( + _event(), _source(), "agent:main:telegram:group:-1001:12345", 1 + ) + finally: + assert runner._turn_leases.release(holder) is True + + runner.session_store.load_transcript.assert_not_called() + + +@pytest.mark.asyncio +async def test_full_dispatch_rejects_lease_timeout_without_running_goal_hook( + monkeypatch, tmp_path +): + """A lease rejection is not a completed turn for `/goal` evaluation. + + The lease wait also has its own clock: a short lease budget must reject + promptly even while the normal agent inactivity timeout remains long. + """ + from tests.gateway.test_42039_duplicate_user_message import _bootstrap, _event + + runner = _bootstrap(monkeypatch, tmp_path) + runner._turn_leases = SessionTurnLeaseRegistry() + holder = await runner._turn_leases.acquire( + "sess-dedup", owner_key="holder-key", generation=1, timeout=1 + ) + assert holder is not None + monkeypatch.setenv("HERMES_AGENT_TIMEOUT", "5") + monkeypatch.setenv("HERMES_TURN_LEASE_TIMEOUT", "0.02") + + runner.session_store.load_transcript.side_effect = AssertionError( + "transcript must not load after a turn-lease timeout" + ) + session_env_tokens = object() + runner._set_session_env = MagicMock(return_value=session_env_tokens) + runner._clear_session_env = MagicMock() + runner._run_agent = pytest.fail + runner._post_turn_goal_continuation = AsyncMock() + + try: + response = await asyncio.wait_for(runner._handle_message(_event()), timeout=1) finally: assert runner._turn_leases.release(holder) is True assert isinstance(response, str) - assert "still running" in response.lower() assert "not processed" in response.lower() assert "resend" in response.lower() runner.session_store.load_transcript.assert_not_called() + runner._clear_session_env.assert_called_once_with(session_env_tokens) + runner._post_turn_goal_continuation.assert_not_awaited() # --------------------------------------------------------------------------- diff --git a/website/docs/reference/environment-variables.md b/website/docs/reference/environment-variables.md index bae4a6166c20a..7ee3b7de79784 100644 --- a/website/docs/reference/environment-variables.md +++ b/website/docs/reference/environment-variables.md @@ -805,6 +805,7 @@ Advanced per-platform knobs for throttling the outbound message batcher. Most us | `HERMES_STREAM_RETRIES` | Number of mid-stream reconnect attempts on transient network errors (default: `3`). | | `HERMES_STREAM_STALE_GIVEUP` | Cross-turn circuit breaker: after this many consecutive stale kills (streaming or non-streaming) with no completed response, abort each call immediately with an actionable error instead of re-waiting out the stale timeout (default: `5`, `0` disables). Resets on any completed response, `/model` switch, fallback activation, or turn-start primary restore. | | `HERMES_AGENT_TIMEOUT` | Gateway inactivity timeout for a running agent in seconds (default: `1800`, 30 minutes). Resets on every tool call and streamed token. Set to `0` to disable. | +| `HERMES_TURN_LEASE_TIMEOUT` | Maximum time in seconds an alias routing key waits for the active turn on the same resolved session before Hermes rejects the message with a resend notice (default: `1800`). This is independent of agent inactivity; non-positive values use the default. | | `HERMES_GATEWAY_MAX_STARTS` | Respawn-storm circuit breaker: maximum gateway (re)starts allowed within the window before an exponential backoff is slept to break the storm (default: `5`, `0` disables). Also configurable via `gateway.respawn_storm.max_starts` in `config.yaml`. | | `HERMES_GATEWAY_START_WINDOW_S` | Respawn-storm breaker window in seconds (default: `120`). Also configurable via `gateway.respawn_storm.window_seconds` in `config.yaml`. | | `HERMES_AGENT_TIMEOUT_WARNING` | Gateway: send a warning message after this many seconds of inactivity (default: 75% of `HERMES_AGENT_TIMEOUT`). |