* fix(gateway): serialize concurrent turns per resolved session_id with a turn lease Closes the serialization half of #64934. The busy guards are keyed by routing key, but the durable transcript is owned by session_id — and switch_session() makes the key→id mapping many-to-one (/resume from a second chat/topic, CLI-continuity rebinding, async-delegation pinning, topic-binding tip-walks). Two routing keys mapped to one session_id ran concurrent turns on two different agent objects, invisible to every per-key guard: flushes persisted in completion order, the identity-marker dedup swallowed rows, and the second turn ran on a stale history base — leaving a permanent user;user alternation wedge. The fix: an asyncio lease keyed by RESOLVED session_id (gateway/turn_lease.py), acquired in _handle_message_with_agent after session resolution is final (post switch_session/tip-walk), immediately before the transcript load, and released in _handle_message's finally on every exit path. Tokens are granted per (routing key, run generation) so a stale unwind can never release a newer turn's lease (#28686 ownership lesson). Same-key messages never reach the acquisition point mid-turn (both routing-key guards hold them), so the lock is uncontended outside the alias-key route — where the second turn now waits for the first turn's flush and logs one WARNING naming the session and both routing keys (pairs with the #67371 tripwire). Fail-open: a stuck holder degrades to today's unserialized behavior with a loud ERROR after agent.gateway_timeout — never a wedged session; a degraded token holds nothing and can't steal the lease. Registry is size-capped and never evicts a live lease. Persist-disabled review forks never dispatch through _handle_message, so they cannot contend. Known limits (tracked on #64934): CLI-continuity cross-process pairs need a DB-level lease; mid-turn compression rotation leaves a small alias window for a follow-up at the binding-sync sites. Validation: 8 behavior tests (alias-key wait + flush order, no cross-session contention, generation-scoped idempotent release, timeout fail-open without lease theft, bounded registry, bare-runner-safe release wiring) + E2E against a real SessionStore reproducing the issue's switch_session alias route — strict alternation and arrival order preserved. * refactor(gateway): conversation-scope funnel + mid-turn lease rebind Completes the #64934 system beyond the point fix. Two structural changes, both eliminating whole bug classes rather than instances: 1. _clear_conversation_scope — THE single conversation-boundary funnel. /new, /resume, auto-reset, expiry finalization, and the compression-exhausted reset each carried a hand-copied pop-list of the per-session dicts, and the lists drifted every time a new dict was added (#48031, #58403, #10702, #35809 were all 'boundary X forgot dict Y' bugs). All five sites now make one funnel call driven by the _CONVERSATION_SCOPED_STATE registry; adding a new conversation-scoped dict means adding one name to the registry, and every boundary picks it up automatically. Scope rules documented at the registry: turn-scoped state, the monotonic generation counter, and the agent cache are deliberately excluded (different lifecycles). 2. SessionTurnLeaseRegistry.rebind — the held turn lease now FOLLOWS mid-turn compression rotation. Both rotation sites (session-hygiene pre-compression, agent-result session_id swap) alias the same _SessionLease object under the new id, so an alias routing key resolving the fresh child (topic tip-walk) still serializes against the in-flight turn. Closes the rotation-alias window flagged as a known limit on #64934. Ownership-checked like release; when the target id already has a live lease the rebind fails open with a loud WARNING (never a mid-turn deadlock). Tests: 3 new rebind behavior tests + 5 funnel behavior tests (including a real-setter drift guard); the two AST change-detector pins in test_10710/test_48031 were re-pointed at the funnel and the #58403 pin converted to a behavioral test. E2E: rotation-alias scenario against a real SessionStore + SessionDB — turn B on the fresh child waits behind the rotated holder, sees its rows, alternation intact.
This commit is contained in:
Teknium
+211
-56
@@ -1921,6 +1921,7 @@ from gateway.session import (
|
||||
neutralize_untrusted_inline_text,
|
||||
)
|
||||
from gateway.delivery import DeliveryRouter, looks_like_telegram_private_chat_id
|
||||
from gateway.turn_lease import SessionTurnLeaseRegistry
|
||||
from gateway.authz_mixin import GatewayAuthorizationMixin
|
||||
from gateway.kanban_watchers import GatewayKanbanWatchersMixin
|
||||
from gateway.slash_commands import GatewaySlashCommandsMixin
|
||||
@@ -2006,6 +2007,39 @@ def _own_policy_open_startup_violation(config) -> Optional[str]:
|
||||
# between the guard check and actual agent creation.
|
||||
_AGENT_PENDING_SENTINEL = object()
|
||||
|
||||
# Conversation-scoped per-session state registry. Every GatewayRunner dict
|
||||
# keyed by session_key whose entries must NOT survive a conversation boundary
|
||||
# (/new, /resume, auto-reset, expiry finalization, compression-exhausted
|
||||
# reset) is listed here, and _clear_conversation_scope() pops them all.
|
||||
# Boundaries used to each carry a hand-copied pop-list that drifted whenever
|
||||
# a new dict was added (#48031, #58403, #10702, #35809). Adding a new
|
||||
# conversation-scoped dict means adding its attribute name HERE — every
|
||||
# boundary then handles it automatically.
|
||||
#
|
||||
# NOT in this list (different lifecycles):
|
||||
# - _running_agents/_running_agents_ts/_active_session_leases/_busy_ack_ts/
|
||||
# _turn_lease_tokens: turn-scoped, owned by _release_running_agent_state
|
||||
# and the dispatch finally.
|
||||
# - _session_run_generation: monotonic by design; clearing it would reset
|
||||
# the counter and break stale-run detection (#28686).
|
||||
# - _agent_cache: has its own eviction path (_evict_cached_agent) with
|
||||
# resource cleanup; boundaries call it explicitly.
|
||||
# - _pending_approvals/_update_prompt_pending/slash-confirm/tool-approval
|
||||
# state: cleared via _clear_session_boundary_security_state, which
|
||||
# _clear_conversation_scope calls.
|
||||
_CONVERSATION_SCOPED_STATE: tuple = (
|
||||
"_session_model_overrides",
|
||||
"_pending_one_turn_model_restores",
|
||||
"_session_reasoning_overrides",
|
||||
"_pending_model_notes",
|
||||
"_last_resolved_model",
|
||||
"_queued_events",
|
||||
# Staged-but-never-consumed sidecar notes (turn aborted between staging
|
||||
# and run_sync) must not leak into a future conversation's first user
|
||||
# message — session keys are source-derived and REUSED.
|
||||
"_pending_turn_sidecar_notes",
|
||||
)
|
||||
|
||||
# Sentinel for "caller did not pass metadata" vs "caller passed None".
|
||||
_UNSET = object()
|
||||
|
||||
@@ -3152,6 +3186,17 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
||||
self._running_agents: Dict[str, Any] = {}
|
||||
self._running_agents_ts: Dict[str, float] = {} # start timestamp per session
|
||||
self._active_session_leases: Dict[str, Any] = {}
|
||||
# Per-SESSION_ID turn lease (#64934): serializes the
|
||||
# [load history → run → flush] region when two ROUTING KEYS resolve
|
||||
# to one session_id (switch_session's many-to-one mapping). The
|
||||
# routing-key guards above cannot see that overlap. Acquired in
|
||||
# _handle_message_with_agent after session resolution is final,
|
||||
# released via _release_turn_lease in the same method's finally.
|
||||
self._turn_leases = SessionTurnLeaseRegistry()
|
||||
# Tokens for held turn leases, keyed by (routing key, run generation)
|
||||
# so release is granted per-turn and a stale unwind can never free a
|
||||
# newer turn's lease (#28686 ownership lesson).
|
||||
self._turn_lease_tokens: Dict[tuple, Any] = {}
|
||||
self._pending_messages: Dict[str, str] = {} # Queued messages during interrupt
|
||||
# Last successfully-resolved (non-empty) model, keyed by session. Used
|
||||
# as a fallback when a fresh config read transiently returns an empty
|
||||
@@ -8335,39 +8380,19 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
||||
# be garbage-collected. Otherwise the cache grows
|
||||
# unbounded across the gateway's lifetime.
|
||||
self._evict_cached_agent(key)
|
||||
# Permanently finalizing this session — drop its
|
||||
# per-session control state so the dicts don't grow
|
||||
# Permanently finalizing this session — one funnel
|
||||
# call drops every conversation-scoped dict AND the
|
||||
# boundary security state (approvals, update
|
||||
# prompts, slash-confirm) so the dicts don't grow
|
||||
# unbounded across the gateway's lifetime. (Idle
|
||||
# agent-cache eviction must NOT prune these: the
|
||||
# agent-cache eviction must NOT do this: the
|
||||
# session is still alive and a resumed turn rebuilds
|
||||
# its agent from these overrides. Only true session
|
||||
# finalization, /new, and /reset clear them.)
|
||||
self._session_model_overrides.pop(key, None)
|
||||
self._pending_one_turn_model_restores.pop(key, None)
|
||||
self._set_session_reasoning_override(key, None)
|
||||
if hasattr(self, "_pending_model_notes"):
|
||||
self._pending_model_notes.pop(key, None)
|
||||
# Session keys are source-derived and REUSED by the
|
||||
# next conversation: a staged-but-never-consumed
|
||||
# sidecar note (turn aborted between staging and
|
||||
# run_sync) must not leak into a future session's
|
||||
# first user message.
|
||||
_psn = getattr(self, "_pending_turn_sidecar_notes", None)
|
||||
if isinstance(_psn, dict):
|
||||
_psn.pop(key, None)
|
||||
# Clear per-session model cache so a resumed turn
|
||||
# resolves from current config, not a stale fallback
|
||||
# cached before the session went idle (mirrors /new
|
||||
# and the compression-exhausted auto-reset, #58403).
|
||||
_lrm = getattr(self, "_last_resolved_model", None)
|
||||
if _lrm is not None:
|
||||
_lrm.pop(key, None)
|
||||
_pending_approvals = getattr(self, "_pending_approvals", None)
|
||||
if isinstance(_pending_approvals, dict):
|
||||
_pending_approvals.pop(key, None)
|
||||
_update_prompt_pending = getattr(self, "_update_prompt_pending", None)
|
||||
if isinstance(_update_prompt_pending, dict):
|
||||
_update_prompt_pending.pop(key, None)
|
||||
# finalization, /new, and /reset clear them.) See
|
||||
# _CONVERSATION_SCOPED_STATE.
|
||||
self._clear_conversation_scope(
|
||||
key, reason="expiry_finalized"
|
||||
)
|
||||
# Persist the finalized flag to sessions.json AND
|
||||
# state.db (single write-path, #9006) — also drops
|
||||
# the persisted /model override, since finalization
|
||||
@@ -11371,6 +11396,10 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
||||
# inside _run_agent returns False, and the old sentinel-only check here
|
||||
# missed the leftover real agent — locking the session out forever (#28686).
|
||||
self._release_running_agent_state(_quick_key)
|
||||
# Turn lease (#64934): release THIS turn's lease token — keyed by
|
||||
# (routing key, run generation) so this unwind can only ever free
|
||||
# the lease its own turn acquired, never a newer turn's.
|
||||
self._release_turn_lease(_quick_key, _run_generation)
|
||||
|
||||
def _restore_moa_one_shot(self, event: "MessageEvent", quick_key: str) -> None:
|
||||
"""Revert a ``/moa <prompt>`` one-shot model override after its turn.
|
||||
@@ -11998,22 +12027,13 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
||||
# wiping model/reasoning overrides set between turns (Closes #48031).
|
||||
_was_auto_reset = getattr(session_entry, "was_auto_reset", False)
|
||||
if _was_auto_reset:
|
||||
# Treat auto-reset as a full conversation boundary — drop every
|
||||
# session-scoped transient state so the fresh session does not
|
||||
# inherit the previous conversation's model/reasoning overrides
|
||||
# or a queued "/model switched" note.
|
||||
self._session_model_overrides.pop(session_key, None)
|
||||
self._pending_one_turn_model_restores.pop(session_key, None)
|
||||
self._set_session_reasoning_override(session_key, None)
|
||||
if hasattr(self, "_pending_model_notes"):
|
||||
self._pending_model_notes.pop(session_key, None)
|
||||
# Clear per-session model cache so the fresh session resolves
|
||||
# from current config, not a stale fallback cached before the
|
||||
# auto-reset (mirrors /new and the compression-exhausted
|
||||
# auto-reset, #58403).
|
||||
_lrm = getattr(self, "_last_resolved_model", None)
|
||||
if _lrm is not None:
|
||||
_lrm.pop(session_key, None)
|
||||
# Treat auto-reset as a full conversation boundary — clear every
|
||||
# conversation-scoped per-session dict in one funnel call so the
|
||||
# fresh session does not inherit the previous conversation's
|
||||
# model/reasoning overrides, a queued "/model switched" note, or
|
||||
# a stale resolved-model cache (#48031, #58403). See
|
||||
# _CONVERSATION_SCOPED_STATE.
|
||||
self._clear_conversation_scope(session_key, reason="auto_reset")
|
||||
# Evict the cached agent so the fresh session does not inherit the
|
||||
# previous conversation's context_compressor._previous_summary —
|
||||
# the cache is keyed on the stable session_key, so an auto-reset
|
||||
@@ -12183,6 +12203,33 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
||||
except Exception as e:
|
||||
logger.warning("[Gateway] Failed to auto-load skill(s) %s: %s", _skill_names, e)
|
||||
|
||||
# ── Turn lease (#64934) ────────────────────────────────────────
|
||||
# Session resolution is FINAL here (get_or_create → async-delegation
|
||||
# pinning → topic tip-walk switch_session are all above). Serialize
|
||||
# the [load history → run → flush] region per resolved SESSION_ID:
|
||||
# when a second routing key is mapped to this same session_id, its
|
||||
# turn waits here for the previous turn's flush instead of loading a
|
||||
# stale history base and interleaving transcript writes. Same-key
|
||||
# messages never reach this point mid-turn (adapter + runner guards
|
||||
# hold them), so the lock is uncontended outside the alias-key route.
|
||||
# Fail-open: on timeout the token comes back degraded and the turn
|
||||
# proceeds unserialized (never a wedged session). 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.
|
||||
_lease_registry = getattr(self, "_turn_leases", None)
|
||||
if _lease_registry is not None:
|
||||
_lease_token = await _lease_registry.acquire(
|
||||
session_entry.session_id,
|
||||
owner_key=_quick_key,
|
||||
generation=run_generation,
|
||||
timeout=_float_env("HERMES_AGENT_TIMEOUT", 1800),
|
||||
)
|
||||
if _lease_token is not None:
|
||||
if not hasattr(self, "_turn_lease_tokens"):
|
||||
self._turn_lease_tokens = {}
|
||||
self._turn_lease_tokens[(_quick_key, run_generation)] = _lease_token
|
||||
|
||||
# Load conversation history from transcript
|
||||
history = await self.async_session_store.load_transcript(session_entry.session_id)
|
||||
|
||||
@@ -12447,6 +12494,13 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
||||
)
|
||||
if _hyg_rotated:
|
||||
session_entry.session_id = _hyg_new_sid
|
||||
# The held turn lease follows the
|
||||
# rotation so an alias key resolving
|
||||
# the fresh child still serializes
|
||||
# against this turn (#64934).
|
||||
self._rebind_turn_lease(
|
||||
_quick_key, run_generation, _hyg_new_sid
|
||||
)
|
||||
await self.async_session_store._save()
|
||||
await asyncio.to_thread(
|
||||
self._sync_telegram_topic_binding,
|
||||
@@ -12927,6 +12981,13 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
||||
if agent_result.get("session_id") and agent_result["session_id"] != session_entry.session_id:
|
||||
if session_entry.session_id == _run_start_session_id:
|
||||
session_entry.session_id = agent_result["session_id"]
|
||||
# The held turn lease follows the rotation: the transcript
|
||||
# persistence below writes to the NEW id, so the
|
||||
# serialization boundary must move with it or an alias
|
||||
# key resolving the fresh child could interleave (#64934).
|
||||
self._rebind_turn_lease(
|
||||
_quick_key, run_generation, session_entry.session_id
|
||||
)
|
||||
await self.async_session_store._save()
|
||||
await self.async_session_store._record_gateway_session_peer(
|
||||
session_entry.session_id,
|
||||
@@ -13138,16 +13199,12 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
||||
)
|
||||
new_entry = await self.async_session_store.reset_session(session_key)
|
||||
self._evict_cached_agent(session_key)
|
||||
self._session_model_overrides.pop(session_key, None)
|
||||
self._pending_one_turn_model_restores.pop(session_key, None)
|
||||
self._set_session_reasoning_override(session_key, None)
|
||||
if hasattr(self, "_pending_model_notes"):
|
||||
self._pending_model_notes.pop(session_key, None)
|
||||
# Clear per-session model cache so the post-reset turn
|
||||
# resolves from current config, not a stale fallback.
|
||||
_lrm = getattr(self, "_last_resolved_model", None)
|
||||
if _lrm is not None:
|
||||
_lrm.pop(session_key, None)
|
||||
# Conversation boundary: one funnel call clears every
|
||||
# conversation-scoped per-session dict (#58403 and siblings).
|
||||
# See _CONVERSATION_SCOPED_STATE.
|
||||
self._clear_conversation_scope(
|
||||
session_key, reason="compression_exhausted_reset"
|
||||
)
|
||||
if new_entry is not None:
|
||||
# Drop the stale reference to the bloated compressed child and
|
||||
# re-point the Telegram topic binding at the fresh session.
|
||||
@@ -17459,6 +17516,104 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
|
||||
self._persist_active_agents()
|
||||
return True
|
||||
|
||||
def _release_turn_lease(self, session_key: str, run_generation: int) -> bool:
|
||||
"""Release the turn lease acquired by (``session_key``, ``run_generation``).
|
||||
|
||||
Companion to the acquisition in ``_handle_message_with_agent``
|
||||
(#64934). The token map is keyed by (routing key, run generation), so
|
||||
this can only ever free the lease its own turn acquired — a stale
|
||||
unwind whose generation was bumped by /stop or /new pops ITS token,
|
||||
and the registry's identity check refuses it if a newer turn already
|
||||
holds the lease. Idempotent and safe for bare test runners built via
|
||||
``object.__new__`` (getattr defaults).
|
||||
"""
|
||||
if not session_key:
|
||||
return False
|
||||
tokens = getattr(self, "_turn_lease_tokens", None)
|
||||
registry = getattr(self, "_turn_leases", None)
|
||||
if tokens is None or registry is None:
|
||||
return False
|
||||
token = tokens.pop((session_key, run_generation), None)
|
||||
if token is None:
|
||||
return False
|
||||
try:
|
||||
return registry.release(token)
|
||||
except Exception:
|
||||
logger.debug("Failed to release turn lease", exc_info=True)
|
||||
return False
|
||||
|
||||
def _rebind_turn_lease(
|
||||
self, session_key: str, run_generation: int, new_session_id: str
|
||||
) -> bool:
|
||||
"""Follow a mid-turn session_id rotation with the held turn lease.
|
||||
|
||||
Compression (session-hygiene pre-compression or the agent's own
|
||||
compressor) can rotate ``session_entry.session_id`` while this turn
|
||||
is in flight. The turn's flush targets the NEW id, so the
|
||||
serialization boundary must follow it — otherwise an alias routing
|
||||
key resolving the new id (topic tip-walk onto the fresh child) could
|
||||
start a concurrent turn the lease never sees (#64934 rotation-alias
|
||||
window). Call at every site that reassigns session_entry.session_id
|
||||
mid-turn. Fail-open no-op when there is no held token.
|
||||
"""
|
||||
if not session_key or not new_session_id:
|
||||
return False
|
||||
tokens = getattr(self, "_turn_lease_tokens", None)
|
||||
registry = getattr(self, "_turn_leases", None)
|
||||
if tokens is None or registry is None:
|
||||
return False
|
||||
token = tokens.get((session_key, run_generation))
|
||||
if token is None:
|
||||
return False
|
||||
try:
|
||||
return registry.rebind(token, new_session_id)
|
||||
except Exception:
|
||||
logger.debug("Failed to rebind turn lease", exc_info=True)
|
||||
return False
|
||||
|
||||
def _clear_conversation_scope(self, session_key: str, *, reason: str) -> None:
|
||||
"""Clear ALL conversation-scoped per-session state for ``session_key``.
|
||||
|
||||
THE single conversation-boundary funnel. Call this — and nothing
|
||||
else — whenever a session_key crosses a conversation boundary:
|
||||
/new, /resume, auto-reset (idle/daily/suspended), expiry
|
||||
finalization, and the compression-exhausted auto-reset.
|
||||
|
||||
Why a funnel: these boundaries used to each carry a hand-copied
|
||||
pop-list of the per-session dicts, and the lists drifted every time
|
||||
a new dict was added (#48031, #58403, #10702, #35809 were all
|
||||
"boundary X forgot dict Y" bugs — e.g. /new cleared the /model
|
||||
override but not the /model --once restore snapshot). Adding a new
|
||||
conversation-scoped dict now means adding its attribute name to
|
||||
_CONVERSATION_SCOPED_STATE below; every boundary picks it up
|
||||
automatically.
|
||||
|
||||
Scope rules:
|
||||
- Conversation-scoped (cleared here): model/reasoning overrides,
|
||||
one-turn restore snapshots, pending model notes, last-resolved
|
||||
model cache, queued follow-up events, and the boundary security
|
||||
state (approvals, /yolo, slash-confirm, update prompts).
|
||||
- Turn-scoped (NOT cleared here): _running_agents/_ts, slot leases,
|
||||
turn-lease tokens — owned by _release_running_agent_state and the
|
||||
dispatch finally.
|
||||
- Idle agent-cache eviction is NOT a conversation boundary: the
|
||||
session is still alive and a resumed turn rebuilds from these
|
||||
overrides. Only true boundaries call this.
|
||||
|
||||
Safe on bare test runners built via ``object.__new__`` (every
|
||||
access is getattr-guarded).
|
||||
"""
|
||||
if not session_key:
|
||||
return
|
||||
for attr in _CONVERSATION_SCOPED_STATE:
|
||||
store = getattr(self, attr, None)
|
||||
if isinstance(store, dict):
|
||||
store.pop(session_key, None)
|
||||
self._clear_session_boundary_security_state(session_key)
|
||||
logger.debug(
|
||||
"Cleared conversation scope for %s (%s)", session_key, reason
|
||||
)
|
||||
|
||||
def _clear_session_boundary_security_state(self, session_key: str) -> None:
|
||||
"""Clear per-session control state that must not survive a boundary switch."""
|
||||
if not session_key:
|
||||
|
||||
+13
-45
@@ -164,12 +164,11 @@ class GatewaySlashCommandsMixin:
|
||||
)
|
||||
self._evict_cached_agent(session_key)
|
||||
|
||||
# Discard any /queue overflow for this session — /new is a
|
||||
# conversation-boundary operation, queued follow-ups from the
|
||||
# previous conversation must not bleed into the new one.
|
||||
_qe = getattr(self, "_queued_events", None)
|
||||
if _qe is not None:
|
||||
_qe.pop(session_key, None)
|
||||
# Conversation boundary: clear ALL conversation-scoped per-session
|
||||
# state (model/reasoning overrides, one-turn restores, model notes,
|
||||
# last-resolved cache, /queue overflow) + security state in one
|
||||
# funnel call. See _CONVERSATION_SCOPED_STATE in gateway/run.py.
|
||||
self._clear_conversation_scope(session_key, reason="session_reset")
|
||||
|
||||
# The old conversation's in-flight async delegations end WITH it
|
||||
# (#55578): after the reset rotates the session id, their completions
|
||||
@@ -204,24 +203,8 @@ class GatewaySlashCommandsMixin:
|
||||
# Reset the session
|
||||
new_entry = await self.async_session_store.reset_session(session_key)
|
||||
|
||||
# Clear any session-scoped model/reasoning overrides so the next agent
|
||||
# picks up configured defaults instead of previous session switches.
|
||||
self._session_model_overrides.pop(session_key, None)
|
||||
self._set_session_reasoning_override(session_key, None)
|
||||
if hasattr(self, "_pending_model_notes"):
|
||||
self._pending_model_notes.pop(session_key, None)
|
||||
|
||||
# Clear the per-session last-resolved-model cache so the next turn
|
||||
# reads from current config instead of falling back to a stale model
|
||||
# after a config change (#58403).
|
||||
_lrm = getattr(self, "_last_resolved_model", None)
|
||||
if _lrm is not None:
|
||||
_lrm.pop(session_key, None)
|
||||
|
||||
# Clear session-scoped dangerous-command approvals and /yolo state.
|
||||
# /new is a conversation-boundary operation — approval state from the
|
||||
# previous conversation must not survive the reset.
|
||||
self._clear_session_boundary_security_state(session_key)
|
||||
# (Conversation-scoped overrides + security state were already
|
||||
# cleared via _clear_conversation_scope above.)
|
||||
|
||||
_old_sid = old_entry.session_id if old_entry else None
|
||||
|
||||
@@ -3914,28 +3897,13 @@ class GatewaySlashCommandsMixin:
|
||||
new_entry = await self.async_session_store.switch_session(session_key, target_id)
|
||||
if not new_entry:
|
||||
return t("gateway.resume.switch_failed")
|
||||
self._clear_session_boundary_security_state(session_key)
|
||||
|
||||
# Clear session-scoped model/reasoning overrides so the resumed
|
||||
# conversation picks up configured defaults instead of a /model
|
||||
# switch made in the previous session under the same chat
|
||||
# session_key. /resume is a conversation boundary just like /new
|
||||
# (which clears these too); without this, a stale override leaks
|
||||
# across the switch. See #10702.
|
||||
_overrides = getattr(self, "_session_model_overrides", None)
|
||||
if isinstance(_overrides, dict):
|
||||
_overrides.pop(session_key, None)
|
||||
self._set_session_reasoning_override(session_key, None)
|
||||
_pending_notes = getattr(self, "_pending_model_notes", None)
|
||||
if isinstance(_pending_notes, dict):
|
||||
_pending_notes.pop(session_key, None)
|
||||
# Clear per-session model cache too, for the same reason — the
|
||||
# resumed conversation must resolve from current config, not a
|
||||
# stale value cached under this session_key before the switch
|
||||
# (mirrors /new and the compression-exhausted auto-reset, #58403).
|
||||
_lrm = getattr(self, "_last_resolved_model", None)
|
||||
if isinstance(_lrm, dict):
|
||||
_lrm.pop(session_key, None)
|
||||
# Conversation boundary: clear ALL conversation-scoped per-session
|
||||
# state (model/reasoning overrides #10702, one-turn restores, model
|
||||
# notes, last-resolved cache #58403, /queue overflow) + security
|
||||
# state in one funnel call. See _CONVERSATION_SCOPED_STATE in
|
||||
# gateway/run.py.
|
||||
self._clear_conversation_scope(session_key, reason="resume")
|
||||
|
||||
# Evict any cached agent for this session so the next message
|
||||
# rebuilds with the correct session_id end-to-end — mirrors
|
||||
|
||||
@@ -0,0 +1,302 @@
|
||||
"""Per-session turn lease — serializes the [load history → run → flush] region.
|
||||
|
||||
Why this exists (#64934): the gateway's busy guards are keyed by ROUTING KEY
|
||||
(``_active_sessions`` in the adapter, ``_running_agents`` in the runner), but
|
||||
the durable transcript is owned by SESSION_ID — and ``switch_session()`` makes
|
||||
the key→id mapping many-to-one (``/resume`` of a named session from a second
|
||||
chat/topic, CLI-continuity rebinding, async-delegation completion pinning,
|
||||
Telegram topic-binding tip-walks). Two routing keys mapped to one session_id
|
||||
run concurrent turns on two different agent objects, so no per-key guard ever
|
||||
sees the collision. The two turns then interleave their flushes on one
|
||||
transcript: rows persist in completion order instead of arrival order, the
|
||||
identity-marker dedup over shared history dicts can swallow a row outright,
|
||||
and the second turn runs on a history base that never saw the first turn's
|
||||
exchange — leaving a permanent ``user;user`` alternation wedge that
|
||||
``repair_message_sequence`` re-repairs on every request forever.
|
||||
|
||||
The lease closes that route by serializing per RESOLVED session_id: it is
|
||||
acquired after session resolution is final (post ``switch_session``/tip-walk),
|
||||
immediately before the transcript load, and released in the dispatch layer's
|
||||
``finally`` on every exit path. Same-key messages never reach the acquisition
|
||||
point while a turn runs (both routing-key guards hold them), so the lock is
|
||||
uncontended everywhere except the alias-key route — where the second turn now
|
||||
waits for the first turn's flush and logs one WARNING naming the session and
|
||||
both routing keys (pairing with the cross-agent tripwire in
|
||||
``agent/agent_runtime_helpers.note_turn_start``).
|
||||
|
||||
Safety properties:
|
||||
|
||||
- **Generation-scoped, identity-checked release.** A token records its owner
|
||||
(routing key, run generation) and release only frees the lease when that
|
||||
exact token is the current holder — a stale unwind can never release a
|
||||
newer turn's lease (the #28686 ownership lesson applied). Release is
|
||||
idempotent.
|
||||
- **Fail-open on timeout.** A stuck holder degrades to today's unserialized
|
||||
behavior with a loud ERROR after the configured wait — never a wedged
|
||||
session. A degraded token holds nothing and releases nothing.
|
||||
- **Bounded registry.** The per-session lease map is size-capped; eviction
|
||||
only ever removes idle (unheld, uncontended) entries, never a live lease.
|
||||
|
||||
Known limits (deliberate, flagged on #64934):
|
||||
|
||||
- A CLI process sharing the session via CLI-continuity is outside any
|
||||
in-process lock — that pair needs a DB-level lease (separate design).
|
||||
- Mid-turn compression rotation leaves a small alias window: the tip-walk can
|
||||
resolve a fresh child id while the parent-holding turn is still in flight.
|
||||
The mid-turn binding-sync sites are the right place to alias the lease in a
|
||||
follow-up.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
import time
|
||||
from typing import Dict, Optional
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Upper bound on tracked per-session leases. Idle entries (no holder, no
|
||||
# waiter) are evicted oldest-first once the cap is reached; live leases are
|
||||
# never evicted, so a burst of distinct sessions can transiently exceed the
|
||||
# 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 so a stuck holder fails open
|
||||
# on the same clock the turn itself would be declared stuck on.
|
||||
DEFAULT_LEASE_WAIT = 1800.0
|
||||
|
||||
|
||||
class TurnLeaseToken:
|
||||
"""Handle returned by :meth:`SessionTurnLeaseRegistry.acquire`.
|
||||
|
||||
``degraded`` means the acquire timed out and the turn is proceeding
|
||||
UNSERIALIZED (fail-open); such a token holds nothing and its release is a
|
||||
no-op. ``released`` makes release idempotent.
|
||||
"""
|
||||
|
||||
__slots__ = ("session_id", "owner_key", "generation", "degraded", "released")
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
session_id: str,
|
||||
owner_key: str,
|
||||
generation: int,
|
||||
degraded: bool = False,
|
||||
) -> None:
|
||||
self.session_id = session_id
|
||||
self.owner_key = owner_key
|
||||
self.generation = generation
|
||||
self.degraded = degraded
|
||||
self.released = False
|
||||
|
||||
def __repr__(self) -> str: # pragma: no cover - debug aid
|
||||
return (
|
||||
f"TurnLeaseToken(session_id={self.session_id!r}, "
|
||||
f"owner_key={self.owner_key!r}, generation={self.generation}, "
|
||||
f"degraded={self.degraded}, released={self.released})"
|
||||
)
|
||||
|
||||
|
||||
class _SessionLease:
|
||||
__slots__ = ("lock", "holder", "acquired_at", "last_used")
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.lock = asyncio.Lock()
|
||||
self.holder: Optional[TurnLeaseToken] = None
|
||||
self.acquired_at = 0.0
|
||||
self.last_used = time.time()
|
||||
|
||||
@property
|
||||
def idle(self) -> bool:
|
||||
"""True when this lease can be evicted: nobody holds or awaits it."""
|
||||
return self.holder is None and not self.lock.locked()
|
||||
|
||||
|
||||
class SessionTurnLeaseRegistry:
|
||||
"""Asyncio lease per resolved session_id serializing transcript turns.
|
||||
|
||||
Process-local and single-event-loop by design — the same visibility scope
|
||||
as the routing-key guards it extends. All methods must be called from the
|
||||
gateway's event loop.
|
||||
"""
|
||||
|
||||
def __init__(self, max_entries: int = DEFAULT_MAX_LEASES) -> None:
|
||||
self._leases: Dict[str, _SessionLease] = {}
|
||||
self._max_entries = max(1, int(max_entries))
|
||||
|
||||
def __len__(self) -> int:
|
||||
return len(self._leases)
|
||||
|
||||
def _get_or_create(self, session_id: str) -> _SessionLease:
|
||||
lease = self._leases.get(session_id)
|
||||
if lease is None:
|
||||
self._evict_idle()
|
||||
lease = _SessionLease()
|
||||
self._leases[session_id] = lease
|
||||
lease.last_used = time.time()
|
||||
return lease
|
||||
|
||||
def _evict_idle(self) -> None:
|
||||
"""Drop oldest idle entries so a new lease fits under the cap.
|
||||
|
||||
Never evicts a held or contended lease — correctness beats the cap.
|
||||
"""
|
||||
overflow = len(self._leases) - self._max_entries + 1
|
||||
if overflow <= 0:
|
||||
return
|
||||
idle_ids = sorted(
|
||||
(sid for sid, lease in self._leases.items() if lease.idle),
|
||||
key=lambda sid: self._leases[sid].last_used,
|
||||
)
|
||||
for sid in idle_ids[:overflow]:
|
||||
self._leases.pop(sid, None)
|
||||
|
||||
async def acquire(
|
||||
self,
|
||||
session_id: str,
|
||||
*,
|
||||
owner_key: str,
|
||||
generation: int,
|
||||
timeout: Optional[float] = None,
|
||||
) -> Optional[TurnLeaseToken]:
|
||||
"""Acquire the turn lease for ``session_id``, waiting if held.
|
||||
|
||||
Returns a :class:`TurnLeaseToken` — degraded when the wait timed out
|
||||
(fail-open: caller proceeds unserialized). Returns ``None`` for a
|
||||
falsy ``session_id``.
|
||||
"""
|
||||
if not session_id:
|
||||
return None
|
||||
wait = float(timeout) if timeout and timeout > 0 else DEFAULT_LEASE_WAIT
|
||||
token = TurnLeaseToken(session_id, owner_key, int(generation))
|
||||
lease = self._get_or_create(session_id)
|
||||
|
||||
if lease.lock.locked():
|
||||
holder = lease.holder
|
||||
logger.warning(
|
||||
"turn lease contention on session %s: routing key %s (gen %s) "
|
||||
"waiting behind in-flight turn held by routing key %s (gen %s, "
|
||||
"held %.0fs) — two routing keys are mapped to one session_id "
|
||||
"(#64934); serializing this turn behind the previous turn's "
|
||||
"flush",
|
||||
session_id,
|
||||
owner_key,
|
||||
generation,
|
||||
holder.owner_key if holder else "?",
|
||||
holder.generation if holder else "?",
|
||||
time.time() - lease.acquired_at if lease.acquired_at else -1.0,
|
||||
)
|
||||
|
||||
try:
|
||||
await asyncio.wait_for(lease.lock.acquire(), timeout=wait)
|
||||
except asyncio.TimeoutError:
|
||||
holder = lease.holder
|
||||
logger.error(
|
||||
"turn lease wait timed out after %.0fs on session %s "
|
||||
"(waiter: routing key %s gen %s; holder: routing key %s "
|
||||
"gen %s) — failing open: this turn runs UNSERIALIZED against "
|
||||
"the stuck holder rather than wedging the session; transcript "
|
||||
"writes may interleave",
|
||||
wait,
|
||||
session_id,
|
||||
owner_key,
|
||||
generation,
|
||||
holder.owner_key if holder else "?",
|
||||
holder.generation if holder else "?",
|
||||
)
|
||||
token.degraded = True
|
||||
return token
|
||||
|
||||
lease.holder = token
|
||||
lease.acquired_at = time.time()
|
||||
lease.last_used = lease.acquired_at
|
||||
return token
|
||||
|
||||
def rebind(self, token: Optional[TurnLeaseToken], new_session_id: str) -> bool:
|
||||
"""Alias a HELD lease onto ``new_session_id`` after mid-turn rotation.
|
||||
|
||||
Compression can rotate the durable session_id while a turn is in
|
||||
flight (session-hygiene pre-compression, in-agent compression). The
|
||||
turn's flush then targets the NEW id — so the serialization boundary
|
||||
must follow it, or an alias routing key resolving the new id (e.g. a
|
||||
topic tip-walk landing on the fresh child) could start a concurrent
|
||||
turn the lease never sees. This closes the rotation-alias window
|
||||
flagged on #64934.
|
||||
|
||||
Mechanism: the SAME ``_SessionLease`` object is registered under the
|
||||
new id (the old mapping stays until it goes idle and is evicted), so
|
||||
acquirers on either id serialize against one lock — no lock state is
|
||||
moved, no asyncio internals are touched. Only the current holder can
|
||||
rebind (identity-checked like release), and the token follows to the
|
||||
new id so release frees the shared object.
|
||||
|
||||
Edge: if the new id already has a live lease of its own (another
|
||||
turn is running on the target session), the two serialization
|
||||
domains cannot be merged mid-wait — log loudly and keep the token on
|
||||
the old id. Fail-open, never deadlock: a holder cannot wait mid-turn.
|
||||
"""
|
||||
if (
|
||||
token is None
|
||||
or token.degraded
|
||||
or token.released
|
||||
or not new_session_id
|
||||
or new_session_id == token.session_id
|
||||
):
|
||||
return False
|
||||
lease = self._leases.get(token.session_id)
|
||||
if lease is None or lease.holder is not token:
|
||||
return False
|
||||
|
||||
existing = self._leases.get(new_session_id)
|
||||
if existing is not None and existing is not lease and not existing.idle:
|
||||
holder = existing.holder
|
||||
logger.warning(
|
||||
"turn lease rebind blocked: session %s rotated to %s mid-turn "
|
||||
"(holder: routing key %s gen %s) but the target session's "
|
||||
"lease is already live (holder: routing key %s gen %s) — "
|
||||
"keeping the lease on the old id; transcript writes on %s "
|
||||
"may interleave (#64934 rotation-alias edge)",
|
||||
token.session_id,
|
||||
new_session_id,
|
||||
token.owner_key,
|
||||
token.generation,
|
||||
holder.owner_key if holder else "?",
|
||||
holder.generation if holder else "?",
|
||||
new_session_id,
|
||||
)
|
||||
return False
|
||||
|
||||
self._leases[new_session_id] = lease
|
||||
lease.last_used = time.time()
|
||||
token.session_id = new_session_id
|
||||
return True
|
||||
|
||||
def release(self, token: Optional[TurnLeaseToken]) -> bool:
|
||||
"""Release ``token``'s lease. Idempotent; ownership-checked.
|
||||
|
||||
Returns True only when this exact token was the current holder and
|
||||
the lock was freed. A degraded token, a re-release, or a stale token
|
||||
whose slot has since been granted to a newer turn are all safe
|
||||
no-ops — a stale unwind can never release a newer turn's lease.
|
||||
"""
|
||||
if token is None or token.degraded or token.released:
|
||||
return False
|
||||
token.released = True
|
||||
lease = self._leases.get(token.session_id)
|
||||
if lease is None:
|
||||
return False
|
||||
if lease.holder is not token:
|
||||
logger.debug(
|
||||
"turn lease release skipped on session %s: token (key %s "
|
||||
"gen %s) is not the current holder",
|
||||
token.session_id,
|
||||
token.owner_key,
|
||||
token.generation,
|
||||
)
|
||||
return False
|
||||
lease.holder = None
|
||||
lease.acquired_at = 0.0
|
||||
lease.last_used = time.time()
|
||||
if lease.lock.locked():
|
||||
lease.lock.release()
|
||||
return True
|
||||
@@ -56,16 +56,18 @@ def test_auto_reset_cleanup_evicts_cached_agent():
|
||||
tree = ast.parse(inspect.getsource(gateway_run))
|
||||
|
||||
# Fingerprint the cleanup branch: the `if <was_auto_reset>:` block that
|
||||
# drops transient session state (calls the reasoning-override setter AND
|
||||
# consumes the flag by setting was_auto_reset = False). The eviction must
|
||||
# live in that same block.
|
||||
# clears the conversation scope via the funnel (post-#64934 refactor:
|
||||
# one _clear_conversation_scope call replaced the inline pops) and
|
||||
# consumes the flag by setting was_auto_reset = False. The eviction must
|
||||
# live in that same block — the funnel deliberately does NOT evict the
|
||||
# agent cache (it has its own resource-cleanup path).
|
||||
found = False
|
||||
for node in ast.walk(tree):
|
||||
if not isinstance(node, ast.If):
|
||||
continue
|
||||
calls = _calls(node)
|
||||
if (
|
||||
"_set_session_reasoning_override" in calls
|
||||
"_clear_conversation_scope" in calls
|
||||
and _assigns_false(node, "was_auto_reset")
|
||||
):
|
||||
assert "_evict_cached_agent" in calls, (
|
||||
@@ -79,7 +81,7 @@ def test_auto_reset_cleanup_evicts_cached_agent():
|
||||
break
|
||||
assert found, (
|
||||
"could not locate the auto-reset transient-state cleanup block in "
|
||||
"gateway/run.py (fingerprint: _set_session_reasoning_override + "
|
||||
"gateway/run.py (fingerprint: _clear_conversation_scope + "
|
||||
"was_auto_reset = False)."
|
||||
)
|
||||
|
||||
@@ -102,37 +104,18 @@ def _references_name(node: ast.AST, literal: str) -> bool:
|
||||
def test_auto_reset_cleanup_clears_last_resolved_model():
|
||||
"""Regression test for #58403.
|
||||
|
||||
The auto-reset cleanup block (daily/idle/suspended, fingerprinted by
|
||||
`_set_session_reasoning_override` + `was_auto_reset = False`) already
|
||||
pops `_session_model_overrides` and `_pending_model_notes` — the same
|
||||
"full conversation boundary" treatment /new and the compression-exhausted
|
||||
auto-reset give `_last_resolved_model`. Without also popping
|
||||
`_last_resolved_model` here, the fresh auto-reset session could serve a
|
||||
model cached before the reset on a transient config-cache miss.
|
||||
Auto-reset is a full conversation boundary and now routes through the
|
||||
`_clear_conversation_scope` funnel. The funnel must clear
|
||||
`_last_resolved_model` — without it, the fresh auto-reset session could
|
||||
serve a model cached before the reset on a transient config-cache miss.
|
||||
Behavioral: exercises the real funnel instead of pinning source shape.
|
||||
"""
|
||||
tree = ast.parse(inspect.getsource(gateway_run))
|
||||
|
||||
found = False
|
||||
for node in ast.walk(tree):
|
||||
if not isinstance(node, ast.If):
|
||||
continue
|
||||
calls = _calls(node)
|
||||
if (
|
||||
"_set_session_reasoning_override" in calls
|
||||
and _assigns_false(node, "was_auto_reset")
|
||||
):
|
||||
assert _references_name(node, "_last_resolved_model") and "pop" in _calls(
|
||||
node
|
||||
), (
|
||||
"gateway/run.py auto-reset cleanup block must also pop the "
|
||||
"session's entry from `_last_resolved_model`, mirroring the "
|
||||
"existing `_session_model_overrides`/`_pending_model_notes` "
|
||||
"pops in the same block (#58403)."
|
||||
)
|
||||
found = True
|
||||
break
|
||||
assert found, (
|
||||
"could not locate the auto-reset transient-state cleanup block in "
|
||||
"gateway/run.py (fingerprint: _set_session_reasoning_override + "
|
||||
"was_auto_reset = False)."
|
||||
runner = object.__new__(gateway_run.GatewayRunner)
|
||||
key = "agent:main:telegram:dm:58403"
|
||||
runner._last_resolved_model = {key: "stale/model", "other": "keep/me"}
|
||||
runner._clear_conversation_scope(key, reason="auto_reset")
|
||||
assert key not in runner._last_resolved_model, (
|
||||
"the conversation-boundary funnel must pop the session's entry from "
|
||||
"`_last_resolved_model` (#58403) — auto-reset routes through it"
|
||||
)
|
||||
assert runner._last_resolved_model.get("other") == "keep/me"
|
||||
|
||||
@@ -49,26 +49,22 @@ def test_run_consumes_was_auto_reset_in_cleanup_block():
|
||||
wipe an override stored between turns (#48031)."""
|
||||
tree = ast.parse(inspect.getsource(gateway_run))
|
||||
|
||||
# Find the cleanup branch: an `if <flag>:` block that pops a model/reasoning
|
||||
# override AND clears the flag. We assert at least one such block sets
|
||||
# was_auto_reset False.
|
||||
# Find the cleanup branch: an `if <flag>:` block that clears the
|
||||
# conversation scope (post-funnel: one _clear_conversation_scope call
|
||||
# replaced the inline override pops) AND clears the flag. We assert at
|
||||
# least one such block sets was_auto_reset False.
|
||||
found = False
|
||||
for node in ast.walk(tree):
|
||||
if not isinstance(node, ast.If):
|
||||
continue
|
||||
names = {
|
||||
n.attr
|
||||
for n in ast.walk(node)
|
||||
if isinstance(n, ast.Attribute)
|
||||
}
|
||||
calls = {
|
||||
n.func.attr
|
||||
for n in ast.walk(node)
|
||||
if isinstance(n, ast.Call) and isinstance(n.func, ast.Attribute)
|
||||
}
|
||||
# The cleanup block references the reasoning-override setter and pops
|
||||
# pending model notes — fingerprint of the transient-state cleanup.
|
||||
if "_set_session_reasoning_override" in calls and _assigns_false(node, "was_auto_reset"):
|
||||
# The cleanup block routes through the conversation-scope funnel —
|
||||
# fingerprint of the transient-state cleanup.
|
||||
if "_clear_conversation_scope" in calls and _assigns_false(node, "was_auto_reset"):
|
||||
found = True
|
||||
break
|
||||
assert found, (
|
||||
|
||||
@@ -0,0 +1,78 @@
|
||||
"""Behavior tests for _clear_conversation_scope — the single conversation-
|
||||
boundary funnel (#64934 follow-up).
|
||||
|
||||
Boundaries (/new, /resume, auto-reset, expiry finalization,
|
||||
compression-exhausted reset) used to each carry a hand-copied pop-list of the
|
||||
per-session dicts, and the lists drifted whenever a new dict was added
|
||||
(#48031, #58403, #10702, #35809 were all "boundary X forgot dict Y" bugs).
|
||||
The funnel clears every dict registered in _CONVERSATION_SCOPED_STATE plus
|
||||
the boundary security state, in one call.
|
||||
"""
|
||||
|
||||
from gateway.run import _CONVERSATION_SCOPED_STATE, GatewayRunner
|
||||
|
||||
KEY = "agent:main:telegram:dm:777"
|
||||
OTHER = "agent:main:discord:dm:888"
|
||||
|
||||
|
||||
def _bare_runner() -> GatewayRunner:
|
||||
runner = object.__new__(GatewayRunner)
|
||||
for attr in _CONVERSATION_SCOPED_STATE:
|
||||
setattr(runner, attr, {KEY: object(), OTHER: object()})
|
||||
# Turn-scoped state that the funnel must NOT touch.
|
||||
runner._running_agents = {KEY: object()}
|
||||
runner._running_agents_ts = {KEY: 1.0}
|
||||
runner._session_run_generation = {KEY: 7}
|
||||
return runner
|
||||
|
||||
|
||||
def test_funnel_clears_every_registered_dict_for_key_only():
|
||||
runner = _bare_runner()
|
||||
runner._clear_conversation_scope(KEY, reason="test")
|
||||
for attr in _CONVERSATION_SCOPED_STATE:
|
||||
store = getattr(runner, attr)
|
||||
assert KEY not in store, f"{attr} not cleared by funnel"
|
||||
assert OTHER in store, f"{attr} cleared the wrong session"
|
||||
|
||||
|
||||
def test_funnel_leaves_turn_scoped_and_generation_state_alone():
|
||||
runner = _bare_runner()
|
||||
runner._clear_conversation_scope(KEY, reason="test")
|
||||
# Turn-scoped: owned by _release_running_agent_state / dispatch finally.
|
||||
assert KEY in runner._running_agents
|
||||
assert KEY in runner._running_agents_ts
|
||||
# Generation counter is monotonic by design (#28686) — never reset.
|
||||
assert runner._session_run_generation[KEY] == 7
|
||||
|
||||
|
||||
def test_funnel_is_bare_runner_safe_and_empty_key_noop():
|
||||
runner = object.__new__(GatewayRunner)
|
||||
# No dicts initialized at all — must not raise (pitfall #17).
|
||||
runner._clear_conversation_scope(KEY, reason="test")
|
||||
runner._clear_conversation_scope("", reason="test")
|
||||
|
||||
|
||||
def test_funnel_clears_state_written_by_real_setters():
|
||||
"""Behavioral invariant: state written through the runner's real setter
|
||||
paths is cleared by the funnel. Guards against a registry entry drifting
|
||||
out of sync with the attribute the setter actually writes (a typo'd
|
||||
registry name would silently clear nothing and resurrect the
|
||||
boundary-drift bug class the funnel exists to kill)."""
|
||||
runner = object.__new__(GatewayRunner)
|
||||
# Real setter: lazily creates _session_reasoning_overrides.
|
||||
runner._set_session_reasoning_override(KEY, {"effort": "high"})
|
||||
assert runner._session_reasoning_overrides.get(KEY) == {"effort": "high"}
|
||||
runner._clear_conversation_scope(KEY, reason="test")
|
||||
assert KEY not in runner._session_reasoning_overrides
|
||||
|
||||
|
||||
def test_funnel_also_clears_boundary_security_state():
|
||||
runner = _bare_runner()
|
||||
runner._pending_approvals = {KEY: {"cmd": "rm -rf"}, OTHER: {}}
|
||||
runner._update_prompt_pending = {KEY: True}
|
||||
runner._pending_skills_reload_notes = {KEY: "note"}
|
||||
runner._clear_conversation_scope(KEY, reason="test")
|
||||
assert KEY not in runner._pending_approvals
|
||||
assert OTHER in runner._pending_approvals
|
||||
assert KEY not in runner._update_prompt_pending
|
||||
assert KEY not in runner._pending_skills_reload_notes
|
||||
@@ -0,0 +1,305 @@
|
||||
"""Behavior tests for the per-session turn lease (#64934).
|
||||
|
||||
The lease serializes the [load history → run → flush] region per RESOLVED
|
||||
session_id, closing the alias-key overlap route: two routing keys mapped to
|
||||
one session_id via switch_session() run turns on two different agent objects,
|
||||
invisible to every routing-key guard.
|
||||
|
||||
Covers:
|
||||
- alias-key turn waits until the first turn's flush, and flush order is
|
||||
preserved (the second turn loads history AFTER the first turn's release)
|
||||
- distinct sessions do not contend
|
||||
- generation-scoped, idempotent release: a stale unwind can never free a
|
||||
newer turn's lease; double-release is a no-op
|
||||
- timeout fail-open: a stuck holder degrades to unserialized with a degraded
|
||||
token, never a wedged session, and the degraded token releases nothing
|
||||
- registry stays bounded; live leases are never evicted
|
||||
- GatewayRunner._release_turn_lease wiring (bare-runner safe, token-scoped)
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
|
||||
import pytest
|
||||
|
||||
from gateway.turn_lease import SessionTurnLeaseRegistry
|
||||
|
||||
|
||||
def _run(coro):
|
||||
return asyncio.run(coro)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Serialization behavior
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_alias_key_turn_waits_and_order_is_preserved():
|
||||
"""Second routing key on the same session_id waits for the first turn's
|
||||
release; events interleave in strict [load1, flush1, load2, flush2] order."""
|
||||
|
||||
async def scenario():
|
||||
registry = SessionTurnLeaseRegistry()
|
||||
events = []
|
||||
|
||||
async def turn(owner_key, generation, hold):
|
||||
token = await registry.acquire(
|
||||
"sess-1", owner_key=owner_key, generation=generation, timeout=5
|
||||
)
|
||||
assert token is not None and not token.degraded
|
||||
events.append(f"load:{owner_key}")
|
||||
await asyncio.sleep(hold) # simulate run + flush
|
||||
events.append(f"flush:{owner_key}")
|
||||
registry.release(token)
|
||||
|
||||
t1 = asyncio.create_task(turn("key-a", 1, hold=0.05))
|
||||
await asyncio.sleep(0.01) # let turn 1 take the lease
|
||||
t2 = asyncio.create_task(turn("key-b", 1, hold=0))
|
||||
await asyncio.gather(t1, t2)
|
||||
return events
|
||||
|
||||
events = _run(scenario())
|
||||
assert events == ["load:key-a", "flush:key-a", "load:key-b", "flush:key-b"]
|
||||
|
||||
|
||||
def test_distinct_sessions_do_not_contend():
|
||||
async def scenario():
|
||||
registry = SessionTurnLeaseRegistry()
|
||||
order = []
|
||||
|
||||
async def turn(session_id, owner_key):
|
||||
token = await registry.acquire(
|
||||
session_id, owner_key=owner_key, generation=1, timeout=5
|
||||
)
|
||||
order.append(f"start:{session_id}")
|
||||
await asyncio.sleep(0.05)
|
||||
order.append(f"end:{session_id}")
|
||||
registry.release(token)
|
||||
|
||||
await asyncio.gather(turn("sess-a", "key-a"), turn("sess-b", "key-b"))
|
||||
return order
|
||||
|
||||
order = _run(scenario())
|
||||
# Both started before either finished — no serialization across sessions.
|
||||
assert order[:2] == ["start:sess-a", "start:sess-b"]
|
||||
|
||||
|
||||
def test_contention_logs_named_warning(caplog):
|
||||
async def scenario():
|
||||
registry = SessionTurnLeaseRegistry()
|
||||
t1 = await registry.acquire("sess-w", owner_key="key-a", generation=3, timeout=5)
|
||||
|
||||
async def second():
|
||||
t2 = await registry.acquire(
|
||||
"sess-w", owner_key="key-b", generation=7, timeout=5
|
||||
)
|
||||
registry.release(t2)
|
||||
|
||||
task = asyncio.create_task(second())
|
||||
await asyncio.sleep(0.01)
|
||||
registry.release(t1)
|
||||
await task
|
||||
|
||||
with caplog.at_level("WARNING", logger="gateway.turn_lease"):
|
||||
_run(scenario())
|
||||
warnings = [r for r in caplog.records if "turn lease contention" in r.getMessage()]
|
||||
assert len(warnings) == 1
|
||||
msg = warnings[0].getMessage()
|
||||
assert "sess-w" in msg and "key-a" in msg and "key-b" in msg
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Release semantics
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_generation_scoped_idempotent_release():
|
||||
"""A released token re-released is a no-op, and a stale token cannot free
|
||||
a newer turn's lease."""
|
||||
|
||||
async def scenario():
|
||||
registry = SessionTurnLeaseRegistry()
|
||||
stale = await registry.acquire("sess-g", owner_key="key-a", generation=1, timeout=5)
|
||||
assert stale is not None
|
||||
assert registry.release(stale) is True
|
||||
# Double release: no-op.
|
||||
assert registry.release(stale) is False
|
||||
|
||||
newer = await registry.acquire("sess-g", owner_key="key-a", generation=2, timeout=5)
|
||||
# Stale token (already released, older generation) must not free the
|
||||
# newer holder even if some unwind calls release again.
|
||||
stale.released = False # simulate a buggy double-unwind resurrecting it
|
||||
assert registry.release(stale) is False
|
||||
# Newer turn still holds the lease: a third acquire must wait.
|
||||
waiter = asyncio.create_task(
|
||||
registry.acquire("sess-g", owner_key="key-b", generation=3, timeout=5)
|
||||
)
|
||||
await asyncio.sleep(0.02)
|
||||
assert not waiter.done()
|
||||
assert registry.release(newer) is True
|
||||
third = await waiter
|
||||
assert third is not None and not third.degraded
|
||||
registry.release(third)
|
||||
|
||||
_run(scenario())
|
||||
|
||||
|
||||
def test_release_none_and_empty_session_are_noops():
|
||||
async def scenario():
|
||||
registry = SessionTurnLeaseRegistry()
|
||||
assert registry.release(None) is False
|
||||
assert await registry.acquire("", owner_key="k", generation=1) is None
|
||||
|
||||
_run(scenario())
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fail-open on timeout
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_timeout_fails_open_with_degraded_token(caplog):
|
||||
async def scenario():
|
||||
registry = SessionTurnLeaseRegistry()
|
||||
holder = await registry.acquire(
|
||||
"sess-t", owner_key="key-stuck", generation=1, timeout=5
|
||||
)
|
||||
degraded = await registry.acquire(
|
||||
"sess-t", owner_key="key-b", generation=2, timeout=0.05
|
||||
)
|
||||
assert degraded is not None
|
||||
assert degraded.degraded is True
|
||||
# Degraded release must NOT free the stuck holder's lock (no lease theft).
|
||||
assert registry.release(degraded) is False
|
||||
third = asyncio.create_task(
|
||||
registry.acquire("sess-t", owner_key="key-c", generation=3, timeout=5)
|
||||
)
|
||||
await asyncio.sleep(0.02)
|
||||
assert not third.done() # still held by the original holder
|
||||
registry.release(holder)
|
||||
t3 = await third
|
||||
assert t3 is not None and not t3.degraded
|
||||
registry.release(t3)
|
||||
|
||||
with caplog.at_level("ERROR", logger="gateway.turn_lease"):
|
||||
_run(scenario())
|
||||
errors = [r for r in caplog.records if "failing open" in r.getMessage()]
|
||||
assert len(errors) == 1
|
||||
assert "sess-t" in errors[0].getMessage()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Bounded registry
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_registry_bounded_and_never_evicts_live_lease():
|
||||
async def scenario():
|
||||
registry = SessionTurnLeaseRegistry(max_entries=5)
|
||||
live = await registry.acquire("live", owner_key="k", generation=1, timeout=5)
|
||||
# Churn far past the cap with idle leases.
|
||||
for i in range(50):
|
||||
t = await registry.acquire(f"s{i}", owner_key="k", generation=1, timeout=5)
|
||||
registry.release(t)
|
||||
assert len(registry) <= 6 # cap + the transient new entry
|
||||
# The live lease survived every eviction pass: releasing it works and
|
||||
# it still serializes.
|
||||
assert registry.release(live) is True
|
||||
|
||||
_run(scenario())
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Mid-turn rotation rebind
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_rebind_moves_serialization_to_new_session_id():
|
||||
"""After a mid-turn compression rotation, an alias key acquiring the NEW
|
||||
id must wait behind the holder; release under the new id frees it."""
|
||||
|
||||
async def scenario():
|
||||
registry = SessionTurnLeaseRegistry()
|
||||
token = await registry.acquire("parent", owner_key="key-a", generation=1, timeout=5)
|
||||
assert registry.rebind(token, "child") is True
|
||||
assert token is not None and token.session_id == "child"
|
||||
|
||||
# Alias key resolving the fresh child id serializes behind the holder.
|
||||
waiter = asyncio.create_task(
|
||||
registry.acquire("child", owner_key="key-b", generation=1, timeout=5)
|
||||
)
|
||||
await asyncio.sleep(0.02)
|
||||
assert not waiter.done()
|
||||
|
||||
assert registry.release(token) is True
|
||||
t2 = await waiter
|
||||
assert t2 is not None and not t2.degraded
|
||||
registry.release(t2)
|
||||
|
||||
_run(scenario())
|
||||
|
||||
|
||||
def test_rebind_is_ownership_checked_and_noop_safe():
|
||||
async def scenario():
|
||||
registry = SessionTurnLeaseRegistry()
|
||||
token = await registry.acquire("s1", owner_key="k", generation=1, timeout=5)
|
||||
assert token is not None
|
||||
# Same id → no-op.
|
||||
assert registry.rebind(token, "s1") is False
|
||||
# Empty target → no-op.
|
||||
assert registry.rebind(token, "") is False
|
||||
# None / released tokens → no-op.
|
||||
assert registry.rebind(None, "s2") is False
|
||||
registry.release(token)
|
||||
assert registry.rebind(token, "s2") is False
|
||||
|
||||
_run(scenario())
|
||||
|
||||
|
||||
def test_rebind_blocked_when_target_lease_is_live():
|
||||
"""Two live serialization domains can't be merged mid-wait — rebind
|
||||
fails open (token stays on the old id) with a loud warning."""
|
||||
|
||||
async def scenario():
|
||||
registry = SessionTurnLeaseRegistry()
|
||||
t_a = await registry.acquire("sess-a", owner_key="key-a", generation=1, timeout=5)
|
||||
t_b = await registry.acquire("sess-b", owner_key="key-b", generation=1, timeout=5)
|
||||
assert t_a is not None and t_b is not None
|
||||
assert registry.rebind(t_a, "sess-b") is False
|
||||
assert t_a.session_id == "sess-a" # unchanged
|
||||
# Both still release cleanly under their own ids.
|
||||
assert registry.release(t_a) is True
|
||||
assert registry.release(t_b) is True
|
||||
|
||||
_run(scenario())
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# GatewayRunner wiring
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_runner_release_turn_lease_is_token_scoped_and_bare_safe():
|
||||
from gateway.run import GatewayRunner
|
||||
|
||||
runner = object.__new__(GatewayRunner)
|
||||
# Bare runner without __init__: must be a safe no-op (pitfall #17).
|
||||
assert runner._release_turn_lease("key-a", 1) is False
|
||||
|
||||
async def scenario():
|
||||
runner._turn_leases = SessionTurnLeaseRegistry()
|
||||
runner._turn_lease_tokens = {}
|
||||
token = await runner._turn_leases.acquire(
|
||||
"sess-r", owner_key="key-a", generation=1, timeout=5
|
||||
)
|
||||
runner._turn_lease_tokens[("key-a", 1)] = token
|
||||
# Wrong generation: pops nothing, releases nothing.
|
||||
assert runner._release_turn_lease("key-a", 2) is False
|
||||
assert runner._turn_leases._leases["sess-r"].holder is token
|
||||
# Right (key, generation): releases.
|
||||
assert runner._release_turn_lease("key-a", 1) is True
|
||||
# Idempotent.
|
||||
assert runner._release_turn_lease("key-a", 1) is False
|
||||
# Empty key guard.
|
||||
assert runner._release_turn_lease("", 1) is False
|
||||
|
||||
_run(scenario())
|
||||
Reference in New Issue
Block a user