Gateway session slot and lock handling for stale runs

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #14513 fix(gateway): heal split-brain session locks — salvage #8472 + #12219 (closes #11016)

  merged to main

  files: gateway/platforms/base.py, gateway/run.py, scripts/release.py, tests/gateway/test_session_split_brain_11016.py

  Show the diff

  • gateway/platforms/base.py

    @@ -900,10 +900,16 @@ def __init__(self, config: PlatformConfig, platform: Platform):
             self._fatal_error_retryable = True
             self._fatal_error_handler: Optional[Callable[["BasePlatformAdapter"], Awaitable[None] | None]] = None
             
    -        # Track active message handlers per session for interrupt support
    -        # Key: session_key (e.g., chat_id), Value: (event, asyncio.Event for interrupt)
    +        # Track active message handlers per session for interrupt support.
    +        # _active_sessions stores the per-session interrupt Event; _session_tasks
    +        # maps session → the specific Task currently processing it so that
    +        # session-terminating commands (/stop, /new, /reset) can cancel the
    +        # right task and release the adapter-level guard deterministically.
    +        # Without the owner-task map, an old task's finally block could delete
    +        # a newer task's guard, leaving stale busy state.
             self._active_sessions: Dict[str, asyncio.Event] = {}
             self._pending_messages: Dict[str, MessageEvent] = {}
    +        self._session_tasks: Dict[str, asyncio.Task] = {}
             # Background message-processing tasks spawned by handle_message().
             # Gateway shutdown cancels these so an old gateway instance doesn't keep
             # working on a task after --replace or manual restarts.
    @@ -1680,6 +1686,222 @@ def _merge_caption(existing_text: Optional[str], new_text: str) -> str:
                 return f"{existing_text}\n\n{new_text}".strip()
             return existing_text
     
    +    # ------------------------------------------------------------------
    +    # Session task + guard ownership helpers
    +    # ------------------------------------------------------------------
    +    # These were introduced together with the _session_tasks owner map to
    +    # make session lifecycle reconciliation deterministic across (a) the
    +    # normal completion path, (b) /stop/ /new/ /reset bypass commands,
    +    # and (c) stale-lock self-heal on the next inbound message.
    +
    +    def _release_session_guard(
    +        self,
    +        session_key: str,
    +        *,
    +        guard: Optional[asyncio.Event] = None,
    +    ) -> None:
    +        """Release the adapter-level guard for a session.
    +
    +        When ``guard`` is provided, only release the entry if it still points
    +        at that exact Event.  This lets reset-like commands swap in a temporary
    +        guard while the old processing task unwinds, without having the old
    +        task's cleanup accidentally clear the replacement guard.
    +        """
    +        current_guard = self._active_sessions.get(session_key)
    +        if current_guard is None:
    +            return
    +        if guard is not None and current_guard is not guard:
    +            return
    +        del self._active_sessions[session_key]
    +
    +    def _session_task_is_stale(self, session_key: str) -> bool:
    +        """Return True if the owner task for ``session_key`` is done/cancelled.
    +
    +        A lock is "stale" when the adapter still has ``_active_sessions[key]``
    +        AND a known owner task in ``_session_tasks`` that has already exited.
    +        When there is no owner task at all, that usually means the guard was
    +        installed by some path other than handle_message() (tests sometimes
    +        install guards directly) — don't treat that as stale.  The on-entry
    +        self-heal only needs to handle the production split-brain case where
    +        an owner task was recorded, then exited without clearing its guard.
    +        """
    +        task = self._session_tasks.get(session_key)
    +        if task is None:
    +            return False
    +        done = getattr(task, "done", None)
    +        return bool(done and done())
    +
    +    def _heal_stale_session_lock(self, session_key: str) -> bool:
    +        """Clear a stale session lock if the owner task is already gone.
    +
    +        Returns True if a stale lock was healed.  Returns False if there is
    +        no lock, or the owner task is still alive (the normal busy case).
    +
    +        This is the on-entry safety net sidbin's issue #11016 analysis calls
    +        for: without it, a split-brain — adapter still thinks the session is
    +        active, but nothing is actually processing — traps the chat in
    +        infinite "Interrupting current task..." until the gateway is
    +        restarted.
    +        """
    +        if session_key not in self._active_sessions:
    +            return False
    +        if not self._session_task_is_stale(session_key):
    +            return False
    +        logger.warning(
    +            "[%s] Healing stale session lock for %s (owner task is done/absent)",
    +            self.name,
    +            session_key,
    +        )
    +        self._active_sessions.pop(session_key, None)
    +        self._pending_messages.pop(session_key, None)
    +        self._session_tasks.pop(session_key, None)
    +        return True
    +
    +    def _start_session_processing(
    +        self,
    +        event: MessageEvent,
    +        session_key: str,
    +        *,
    +        interrupt_event: Optional[asyncio.Event] = None,
    +    ) -> bool:
    +        """Spawn a background processing task under the given session guard.
    +
    +        Returns True on success.  If the runtime stubs ``create_task`` with a
    +        non-Task sentinel (some tests do this), the guard is rolled back and
    +        False is returned so the caller isn't left holding a half-installed
    +        session lock.
    +        """
    +        guard = interrupt_event or asyncio.Event()
    +        self._active_sessions[session_key] = guard
    +
    +        task = asyncio.create_task(self._process_message_background(event, session_key))
    +        self._session_tasks[session_key] = task
    +        try:
    +            self._background_tasks.add(task)
    +        except TypeError:
    +            # Tests stub create_task() with lightweight sentinels that are not
    +            # hashable and do not support lifecycle callbacks.
    +            self._session_tasks.pop(session_key, None)
    +            self._release_session_guard(session_key, guard=guard)
    +            return False
    +        if hasattr(task, "add_done_callback"):
    +            task.add_done_callback(self._background_tasks.discard)
    +            task.add_done_callback(self._expected_cancelled_tasks.discard)
    +        return True
    +
    +    async def cancel_session_processing(
    +        self,
    +        session_key: str,
    +        *,
    +        release_guard: bool = True,
    +        discard_pending: bool = True,
    +    ) -> None:
    +        """Cancel in-flight processing for a single session.
    +
    +        ``release_guard=False`` keeps the adapter-level session guard in place
    +        so reset-like commands can finish atomically before follow-up messages
    +        are allowed to start a fresh background task.
    +        """
    +        task = self._session_tasks.pop(session_key, None)
    +        if task is not None and not task.done():
    +            logger.debug(
    +                "[%s] Cancelling active processing for session %s",
    +                self.name,
    +                session_key,
    +            )
    +            self._expected_cancelled_tasks.add(task)
    +            task.cancel()
    +            try:
    +                await task
    +            except asyncio.CancelledError:
    +                pass
    +            except Exception:
    +                logger.debug(
    +                    "[%s] Session cancellation raised while unwinding %s",
    +                    self.name,
    +                    session_key,
    +                    exc_info=True,
    +                )
    +        if discard_pending:
    +            self._pending_messages.pop(session_key, None)
    +        if release_guard:
    +            self._release_session_guard(session_key)
    +
    +    async def _drain_pending_after_session_command(
    +        self,
    +        session_key: str,
    +        command_guard: asyncio.Event,
    +    ) -> None:
    +        """Resume the latest queued follow-up once a session command completes.
    +
    +        Called at the tail of /stop, /new, and /reset dispatch.  Releases the
    +        command-scoped guard, then — if a follow-up message landed while the
    +        command was running — spawns a fresh processing task for it.
    +        """
    +        pending_event = self._pending_messages.pop(session_key, None)
    +        self._release_session_guard(session_key, guard=command_guard)
    +        if pending_event is None:
    +            return
    +        self._start_session_processing(pending_event, session_key)
    +
    +    async def _dispatch_active_session_command(
    +        self,
    +        event: MessageEvent,
    +        session_key: str,
    +        cmd: str,
    +    ) -> None:
    +        """Dispatch a reset-like bypass command while preserving guard ordering.
    +
    +        /stop, /new, and /reset must:
    +          1. Keep the session guard installed while the runner processes the
    +             command (so a racing follow-up message stays queued, not
    +             dispatched as a second parallel run).
    +          2. Cancel the old in-flight adapter task only AFTER the runner has
    +             finished handling the command (so the runner sees consistent
    +             state and its response is sent in order).
    +          3. Release the command-scoped guard and drain the latest queued
    +             follow-up exactly once, after 1 and 2 complete.
    +        """
    +        logger.debug(
    +            "[%s] Command '/%s' bypassing active-session guard for %s",
    +            self.name,
    +            cmd,
    +            session_key,
    +        )
    +
    +        current_guard = self._active_sessions.get(session_key)
    +        command_guard = asyncio.Event()
    +        self._active_sessions[session_key] = command_guard
    +        thread_meta = {"thread_id": event.source.thread_id} if event.source.thread_id else None
    +
    +        try:
    +            response = await self._message_handler(event)
    +            # Old adapter task (if any) is cancelled AFTER the runner has
    +            # fully handled the command — keeps ordering deterministic.
    +            await self.cancel_session_processing(
    +                session_key,
    +                release_guard=False,
    +                discard_pending=False,
    +            )
    +            if response:
    +                await self._send_with_retry(
    +                    chat_id=event.source.chat_id,
    +                    content=response,
    +                    reply_to=event.message_id,
    +                    metadata=thread_meta,
    +                )
    +        except Exception:
    +            # On failure, restore the original guard if one still exists so
    +            # we don't leave the session in a half-reset state.
    +            if self._active_sessions.get(session_key) is command_guard:
    +                if session_key in self._session_tasks and current_guard is not None:
    +                    self._active_sessions[session_key] = current_guard
    +                else:
    +                    self._release_session_guard(session_key, guard=command_guard)
    +            raise
    +
    +        await self._drain_pending_after_session_command(session_key, command_guard)
    +
         async def handle_message(self, event: MessageEvent) -> None:
             """
             Process an incoming message.
    @@ -1696,7 +1918,15 @@ async def handle_message(self, event: MessageEvent) -> None:
                 group_sessions_per_user=self.config.extra.get("group_sessions_per_user", True),
                 thread_sessions_per_user=self.config.extra.get("thread_sessions_per_user", False),
             )
    -        
    +
    +        # On-entry self-heal: if the adapter still has an _active_sessions
    +        # entry for this key but the owner task has already exited (done or
    +        # cancelled), the lock is stale.  Clear it and fall through to
    +        # normal dispatch so the user isn't trapped behind a dead guard —
    +        # this is the split-brain tail described in issue #11016.
    +        if session_key in self._active_sessions:
    +            self._heal_stale_session_lock(session_key)
    +
             # Check if there's already an active handler for this session
             if session_key in self._active_sessions:
                 # Certain commands must bypass the active-session guard and be
    @@ -1713,6 +1943,23 @@ async def handle_message(self, event: MessageEvent) -> None:
                 from hermes_cli.commands import should_bypass_active_session
     
                 if should_bypass_active_session(cmd):
    +                # /stop, /new, /reset must cancel the in-flight adapter task
    +                # and preserve ordering of queued follow-ups.  Route those
    +                # through the dedicated handoff path that serializes
    +                # cancellation + runner response + pending drain.
    +                if cmd in ("stop", "new", "reset"):
    +                    try:
    +                        await self._dispatch_active_session_command(event, session_key, cmd)
    +                    except Exception as e:
    +                        logger.error(
    +                            "[%s] Command '/%s' dispatch failed: %s",
    +                            self.name, cmd, e, exc_info=True,
    +                        )
    +                    return
    +
    +                # Other bypass commands (/approve, /deny, /status,
    +                # /background, /restart) just need direct dispatch — they
    +                # don't cancel the running task.
                     logger.debug(
                         "[%s] Command '/%s' bypassing active-session guard for %s",
                         self.name, cmd, session_key,
    @@ -1758,19 +2005,9 @@ async def handle_message(self, event: MessageEvent) -> None:
             # starts would also pass the _active_sessions check and spawn a
             # duplicate task.  (grammY sequentialize / aiogram EventIsolation
             # pattern — set the guard synchronously, not inside the task.)
    -        self._active_sessions[session_key] = asyncio.Event()
    -
    -        # Spawn background task to process this message
    -        task = asyncio.create_task(self._process_message_background(event, session_key))
    -        try:
    -            self._background_tasks.add(task)
    -        except TypeError:
    -            # Some tests stub create_task() with lightweight sentinels that are not
    -            # hashable and do not support lifecycle callbacks.
    -            return
    -        if hasattr(task, "add_done_callback"):
    -            task.add_done_callback(self._background_tasks.discard)
    -            task.add_done_callback(self._expected_cancelled_tasks.discard)
    +        # _start_session_processing installs the guard AND the owner-task
    +        # mapping atomically so stale-lock detection works.
    +        self._start_session_processing(event, session_key)
         
         @staticmethod
         def _get_human_delay() -> float:
    @@ -2130,6 +2367,9 @@ def _record_delivery(result):
                     drain_task = asyncio.create_task(
                         self._process_message_background(late_pending, session_key)
                     )
    +                # Hand ownership of the session to the drain task so stale-lock
    +                # detection keeps working while it runs.
    +                self._session_tasks[session_key] = drain_task
                     try:
                         self._background_tasks.add(drain_task)
                         drain_task.add_done_callback(self._background_tasks.discard)
    @@ -2139,9 +2379,14 @@ def _record_delivery(result):
                     # Leave _active_sessions[session_key] populated — the drain
                     # task's own lifecycle will clean it up.
                 else:
    -                # Clean up session tracking
    -                if session_key in self._active_sessions:
    -                    del self._active_sessions[session_key]
    +                # Clean up session tracking.  Guard-match both deletes so a
    +                # reset-like command that already swapped in its own
    +                # command_guard (and cancelled us) can't be accidentally
    +                # cleared by our unwind.  The command owns the session now.
    +                current_task = asyncio.current_task()
    +                if current_task is not None and self._session_tasks.get(session_key) is current_task:
    +                    del self._session_tasks[session_key]
    +                self._release_session_guard(session_key, guard=interrupt_event)
         
         async def cancel_background_tasks(self) -> None:
             """Cancel any in-flight background message-processing tasks.
    @@ -2171,6 +2416,7 @@ async def cancel_background_tasks(self) -> None:
                 # will be in self._background_tasks now.  Re-check.
             self._background_tasks.clear()
             self._expected_cancelled_tasks.clear()
    +        self._session_tasks.clear()
             self._pending_messages.clear()
             self._active_sessions.clear()
     

  • gateway/run.py

    @@ -8665,7 +8665,12 @@ def _is_intentional_model_switch(self, session_key: str, agent_model: str) -> bo
             override = self._session_model_overrides.get(session_key)
             return override is not None and override.get("model") == agent_model
     
    -    def _release_running_agent_state(self, session_key: str) -> None:
    +    def _release_running_agent_state(
    +        self,
    +        session_key: str,
    +        *,
    +        run_generation: Optional[int] = None,
    +    ) -> bool:
             """Pop ALL per-running-agent state entries for ``session_key``.
     
             Replaces ad-hoc ``del self._running_agents[key]`` calls scattered
    @@ -8681,13 +8686,25 @@ def _release_running_agent_state(self, session_key: str) -> None:
             across turns (``_session_model_overrides``, ``_voice_mode``,
             ``_pending_approvals``, ``_update_prompt_pending``) is NOT
             touched here — those have their own lifecycles.
    +
    +        When ``run_generation`` is provided, only clear the slot if that
    +        generation is still current for the session.  This prevents an
    +        older async run whose generation was bumped by /stop or /new from
    +        clobbering a newer run's state during its own unwind.  Returns
    +        True when the slot was cleared, False when an ownership guard
    +        blocked it.
             """
             if not session_key:
    -            return
    +            return False
    +        if run_generation is not None and not self._is_session_run_current(
    +            session_key, run_generation
    +        ):
    +            return False
             self._running_agents.pop(session_key, None)
             self._running_agents_ts.pop(session_key, None)
             if hasattr(self, "_busy_ack_ts"):
                 self._busy_ack_ts.pop(session_key, None)
    +        return True
     
         def _clear_session_boundary_security_state(self, session_key: str) -> None:
             """Clear approval state that must not survive a real conversation switch."""
    @@ -10249,10 +10266,24 @@ async def track_agent():
                 # Wait for agent to be created
                 while agent_holder[0] is None:
                     await asyncio.sleep(0.05)
    -            if session_key:
    -                self._running_agents[session_key] = agent_holder[0]
    -                if self._draining:
    -                    self._update_runtime_status("draining")
    +            if not session_key:
    +                return
    +            # Only promote the sentinel to the real agent if this run is still
    +            # current.  If /stop or /new bumped the generation while we were
    +            # spinning up, leave the newer run's slot alone — we'll be
    +            # discarded by the stale-result check in _handle_message_with_agent.
    +            if run_generation is not None and not self._is_session_run_current(
    +                session_key, run_generation
    +            ):
    +                logger.info(
    +                    "Skipping stale agent promotion for %s — generation %s is no longer current",
    +                    (session_key or "")[:20],
    +                    run_generation,
    +                )
    +                return
    +            self._running_agents[session_key] = agent_holder[0]
    +            if self._draining:
    +                self._update_runtime_status("draining")
             
             tracking_task = asyncio.create_task(track_agent())
             
    @@ -10758,7 +10789,14 @@ async def _notify_long_running():
                 # Clean up tracking
                 tracking_task.cancel()
                 if session_key:
    -                self._release_running_agent_state(session_key)
    +                # Only release the slot if this run's generation still owns
    +                # it.  A /stop or /new that bumped the generation while we
    +                # were unwinding has already installed its own state; this
    +                # guard prevents an old run from clobbering it on the way
    +                # out.
    +                self._release_running_agent_state(
    +                    session_key, run_generation=run_generation
    +                )
                 if self._draining:
                     self._update_runtime_status("draining")
                 

  • scripts/release.py

    @@ -111,6 +111,8 @@
         "[redacted:email]": "rnijhara",
         # contributors (manual mapping from git names)
         "[redacted:email]": "asheriif",
    +    "[redacted:email]": "dyxushuai",
    +    "[email protected]": "etcircle",
         "[redacted:email]": "liujinkun2025",
         "[redacted:email]": "dmahan93",
         "[redacted:email]": "ifrederico",

  • tests/gateway/test_session_split_brain_11016.py

    @@ -0,0 +1,399 @@
    +"""Regression tests for issue #11016 — Telegram sessions trapped in
    +repeated 'Interrupting current task...' while /stop reports no active task.
    +
    +Covers three layers of the fix:
    +
    +1. Adapter-side task ownership (_session_tasks map): /stop, /new, /reset
    +   actually cancel the in-flight adapter task and release the guard in
    +   order, so follow-up messages reach the new session.
    +
    +2. Adapter-side on-entry self-heal: if _active_sessions still has an
    +   entry but the recorded owner task is already done/cancelled, clear it
    +   on the next inbound message rather than trapping the user.
    +
    +3. Runner-side generation guard: a stale async run can't promote itself
    +   into _running_agents after /stop/ /new bumped the generation, and
    +   can't clear a newer run's slot on the way out.
    +"""
    +
    +import asyncio
    +from unittest.mock import AsyncMock, MagicMock, patch
    +
    +import pytest
    +
    +from gateway.config import GatewayConfig, Platform, PlatformConfig
    +from gateway.platforms.base import (
    +    BasePlatformAdapter,
    +    MessageEvent,
    +    MessageType,
    +)
    +from gateway.run import GatewayRunner, _AGENT_PENDING_SENTINEL
    +from gateway.session import SessionSource, build_session_key
    +
    +
    +# ---------------------------------------------------------------------------
    +# Adapter helpers
    +# ---------------------------------------------------------------------------
    +
    +
    +class _StubAdapter(BasePlatformAdapter):
    +    async def connect(self):
    +        pass
    +
    +    async def disconnect(self):
    +        pass
    +
    +    async def send(self, chat_id, text, **kwargs):
    +        pass
    +
    +    async def get_chat_info(self, chat_id):
    +        return {}
    +
    +
    +def _make_adapter():
    +    config = PlatformConfig(enabled=True, token="[redacted:secret-assignment]")
    +    adapter = _StubAdapter(config, Platform.TELEGRAM)
    +    adapter.sent_responses = []
    +
    +    async def _mock_send_retry(chat_id, content, **kwargs):
    +        adapter.sent_responses.append(content)
    +
    +    adapter._send_with_retry = _mock_send_retry
    +    return adapter
    +
    +
    +def _make_event(text="hello", chat_id="12345"):
    +    source = SessionSource(
    +        platform=Platform.TELEGRAM, chat_id=chat_id, chat_type="dm"
    +    )
    +    return MessageEvent(text=text, message_type=MessageType.TEXT, source=source)
    +
    +
    +def _session_key(chat_id="12345"):
    +    source = SessionSource(
    +        platform=Platform.TELEGRAM, chat_id=chat_id, chat_type="dm"
    +    )
    +    return build_session_key(source)
    +
    +
    +# ---------------------------------------------------------------------------
    +# Runner helpers
    +# ---------------------------------------------------------------------------
    +
    +
    +def _make_runner():
    +    runner = object.__new__(GatewayRunner)
    +    runner.config = GatewayConfig(
    +        platforms={Platform.TELEGRAM: PlatformConfig(enabled=True, token="***")}
    +    )
    +    runner.adapters = {}
    +    runner._running_agents = {}
    +    runner._running_agents_ts = {}
    +    runner._session_run_generation = {}
    +    runner._pending_messages = {}
    +    runner._draining = False
    +    runner._update_runtime_status = MagicMock()
    +    return runner
    +
    +
    +# ===========================================================================
    +# Layer 1: Adapter-side session cancellation on /stop /new /reset
    +# ===========================================================================
    +
    +
    +class TestAdapterSessionCancellation:
    +    @pytest.mark.asyncio
    +    @pytest.mark.parametrize("command_text", ["/stop", "/new", "/reset"])
    +    async def test_command_cancels_active_task_and_unblocks_follow_up(
    +        self, command_text
    +    ):
    +        """/stop /new /reset must cancel the adapter task and let follow-ups through."""
    +        adapter = _make_adapter()
    +        sk = _session_key()
    +        processing_started = asyncio.Event()
    +        processing_cancelled = asyncio.Event()
    +        blocked_first_message = True
    +
    +        async def _handler(event):
    +            nonlocal blocked_first_message
    +            cmd = event.get_command()
    +            if cmd in {"stop", "new", "reset", "model"}:
    +                return f"handled:{cmd}"
    +
    +            if blocked_first_message:
    +                blocked_first_message = False
    +                processing_started.set()
    +                try:
    +                    await asyncio.Event().wait()
    +                except asyncio.CancelledError:
    +                    processing_cancelled.set()
    +                    raise
    +            return f"handled:text:{event.text}"
    +
    +        adapter._message_handler = _handler
    +
    +        await adapter.handle_message(_make_event("hello world"))
    +        await processing_started.wait()
    +        await asyncio.sleep(0)
    +
    +        assert sk in adapter._active_sessions
    +        assert sk in adapter._session_tasks
    +
    +        await adapter.handle_message(_make_event(command_text))
    +
    +        assert processing_cancelled.is_set(), (
    +            f"{command_text} did not cancel the active processing task"
    +        )
    +        assert sk not in adapter._active_sessions
    +        assert sk not in adapter._pending_messages
    +        assert sk not in adapter._session_tasks
    +        expected = command_text.lstrip("/")
    +        assert any(f"handled:{expected}" in r for r in adapter.sent_responses)
    +
    +        # Follow-up must go through normally now that the session is clean.
    +        await adapter.handle_message(
    +            _make_event("/model xiaomi/mimo-v2-pro --provider nous")
    +        )
    +        await asyncio.sleep(0)
    +        await asyncio.sleep(0)
    +
    +        assert any("handled:model" in r for r in adapter.sent_responses), (
    +            f"follow-up /model stayed blocked after {command_text}"
    +        )
    +        assert sk not in adapter._pending_messages
    +
    +    @pytest.mark.asyncio
    +    async def test_new_keeps_guard_until_command_finishes_then_runs_follow_up(self):
    +        """/new must finish runner logic before cancelling old work or releasing the guard."""
    +        adapter = _make_adapter()
    +        sk = _session_key()
    +        processing_started = asyncio.Event()
    +        command_started = asyncio.Event()
    +        allow_command_finish = asyncio.Event()
    +        follow_up_processed = asyncio.Event()
    +        call_order = []
    +
    +        async def _handler(event):
    +            cmd = event.get_command()
    +            if cmd == "new":
    +                call_order.append("command:start")
    +                command_started.set()
    +                await allow_command_finish.wait()
    +                call_order.append("command:end")
    +                return "handled:new"
    +
    +            if event.text == "hello world":
    +                processing_started.set()
    +                try:
    +                    await asyncio.Event().wait()
    +                except asyncio.CancelledError:
    +                    call_order.append("original:cancelled")
    +                    raise
    +
    +            if event.text == "after reset":
    +                call_order.append("followup:processed")
    +                follow_up_processed.set()
    +            return f"handled:text:{event.text}"
    +
    +        adapter._message_handler = _handler
    +
    +        await adapter.handle_message(_make_event("hello world"))
    +        await processing_started.wait()
    +
    +        command_task = asyncio.create_task(adapter.handle_message(_make_event("/new")))
    +        await command_started.wait()
    +        await asyncio.sleep(0)
    +
    +        assert sk in adapter._active_sessions
    +
    +        await adapter.handle_message(_make_event("after reset"))
    +        await asyncio.sleep(0)
    +        await asyncio.sleep(0)
    +
    +        assert sk in adapter._active_sessions, "guard must stay active while /new is still running"
    +        assert sk in adapter._pending_messages, "follow-up should stay queued until /new finishes"
    +        assert not follow_up_processed.is_set(), "follow-up ran before /new completed"
    +        assert "original:cancelled" not in call_order, "old task was cancelled before runner completed /new"
    +
    +        allow_command_finish.set()
    +        await command_task
    +        await asyncio.wait_for(follow_up_processed.wait(), timeout=1.0)
    +
    +        assert any("handled:new" in r for r in adapter.sent_responses)
    +        assert call_order.index("command:end") < call_order.index("original:cancelled")
    +        assert call_order.index("original:cancelled") < call_order.index("followup:processed")
    +        assert sk not in adapter._pending_messages
    +
    +
    +# ===========================================================================
    +# Layer 2: Adapter-side on-entry self-heal for stale session locks
    +# ===========================================================================
    +
    +
    +class TestStaleSessionLockSelfHeal:
    +    @pytest.mark.asyncio
    +    async def test_stale_lock_with_done_task_is_healed_on_next_message(self):
    +        """A split-brain guard (owner task done but entry still live) heals on next inbound."""
    +        adapter = _make_adapter()
    +        sk = _session_key()
    +
    +        # Simulate the production split-brain: an _active_sessions entry
    +        # remains AND a recorded owner task, but that task is already done.
    +        async def _done():
    +            return None
    +
    +        done_task = asyncio.create_task(_done())
    +        await done_task
    +        assert done_task.done()
    +
    +        adapter._active_sessions[sk] = asyncio.Event()
    +        adapter._session_tasks[sk] = done_task
    +
    +        assert adapter._session_task_is_stale(sk)
    +
    +        async def _handler(event):
    +            return f"handled:{event.get_command() or 'text'}"
    +
    +        adapter._message_handler = _handler
    +
    +        # An ordinary message should heal the stale lock, then fall through
    +        # to normal dispatch.  User gets a reply instead of a busy ack.
    +        await adapter.handle_message(_make_event("hello"))
    +        # Drain any spawned background tasks.
    +        for _ in range(5):
    +            await asyncio.sleep(0)
    +
    +        assert any("handled:text" in r for r in adapter.sent_responses), (
    +            "stale lock trapped a normal message — split-brain not healed"
    +        )
    +
    +    def test_no_owner_task_is_not_treated_as_stale(self):
    +        """If _session_tasks has no entry at all, the guard isn't stale.
    +
    +        Tests and rare legitimate code paths install _active_sessions
    +        entries directly.  Auto-healing those would break real fixtures.
    +        """
    +        adapter = _make_adapter()
    +        sk = _session_key()
    +
    +        adapter._active_sessions[sk] = asyncio.Event()
    +        # No _session_tasks entry.
    +
    +        assert adapter._session_task_is_stale(sk) is False
    +        assert adapter._heal_stale_session_lock(sk) is False
    +
    +    def test_live_owner_task_is_not_stale(self):
    +        """When the owner task is alive, do NOT heal — agent is really busy."""
    +        adapter = _make_adapter()
    +        sk = _session_key()
    +
    +        fake_task = MagicMock()
    +        fake_task.done.return_value = False
    +        adapter._active_sessions[sk] = asyncio.Event()
    +        adapter._session_tasks[sk] = fake_task
    +
    +        assert adapter._session_task_is_stale(sk) is False
    +        assert adapter._heal_stale_session_lock(sk) is False
    +        # Lock still in place.
    +        assert sk in adapter._active_sessions
    +        assert sk in adapter._session_tasks
    +
    +
    +# ===========================================================================
    +# Layer 3: Runner-side generation guard on slot promotion + release
    +# ===========================================================================
    +
    +
    +class TestRunnerSessionGenerationGuard:
    +    def test_release_without_generation_behaves_as_before(self):
    +        runner = _make_runner()
    +        sk = "agent:main:telegram:dm:12345"
    +        runner._running_agents[sk] = "agent"
    +        runner._running_agents_ts[sk] = 1.0
    +        assert runner._release_running_agent_state(sk) is True
    +        assert sk not in runner._running_agents
    +        assert sk not in runner._running_agents_ts
    +
    +    def test_release_with_current_generation_clears_slot(self):
    +        runner = _make_runner()
    +        sk = "agent:main:telegram:dm:12345"
    +        gen = runner._begin_session_run_generation(sk)
    +        runner._running_agents[sk] = "agent"
    +        runner._running_agents_ts[sk] = 1.0
    +
    +        assert runner._release_running_agent_state(sk, run_generation=gen) is True
    +        assert sk not in runner._running_agents
    +
    +    def test_release_with_stale_generation_blocks(self):
    +        runner = _make_runner()
    +        sk = "agent:main:telegram:dm:12345"
    +        stale_gen = runner._begin_session_run_generation(sk)
    +        # /stop bumps the generation — stale run's generation is no longer current.
    +        runner._invalidate_session_run_generation(sk, reason="stop")
    +        # The fresh run lands next; imagine it has its own state installed.
    +        runner._running_agents[sk] = "fresh_agent"
    +        runner._running_agents_ts[sk] = 2.0
    +
    +        # Stale run's unwind MUST NOT clobber the fresh run's state.
    +        released = runner._release_running_agent_state(sk, run_generation=stale_gen)
    +
    +        assert released is False
    +        assert runner._running_agents[sk] == "fresh_agent"
    +        assert runner._running_agents_ts[sk] == 2.0
    +
    +    def test_is_session_run_current_tracks_bumps(self):
    +        runner = _make_runner()
    +        sk = "agent:main:telegram:dm:12345"
    +        gen1 = runner._begin_session_run_generation(sk)
    +        assert runner._is_session_run_current(sk, gen1) is True
    +
    +        runner._invalidate_session_run_generation(sk, reason="test")
    +        assert runner._is_session_run_current(sk, gen1) is False
    +
    +        gen2 = runner._begin_session_run_generation(sk)
    +        assert gen2 > gen1
    +        assert runner._is_session_run_current(sk, gen2) is True
    +
    +
    +# ===========================================================================
    +# Layer 1 (regression): old task's finally must NOT delete a newer guard
    +# ===========================================================================
    +
    +
    +class TestOldTaskCannotClobberNewerGuard:
    +    """Direct regression for the unconditional-delete bug.
    +
    +    Before the guard-match fix, a task in its finally would delete
    +    ``_active_sessions[session_key]`` unconditionally — even if a
    +    /stop/ /new command had already swapped in its own command_guard
    +    (which then gets clobbered, opening a race for follow-up messages).
    +    """
    +
    +    def test_release_session_guard_matches_on_event_identity(self):
    +        adapter = _make_adapter()
    +        sk = _session_key()
    +
    +        old_guard = asyncio.Event()
    +        new_guard = asyncio.Event()
    +        # Command swapped in a newer guard.
    +        adapter._active_sessions[sk] = new_guard
    +
    +        # Old task tries to release using its captured (stale) guard.
    +        adapter._release_session_guard(sk, guard=old_guard)
    +
    +        # The newer guard survives.
    +        assert adapter._active_sessions.get(sk) is new_guard
    +
    +        # Now the command itself releases using the matching guard.
    +        adapter._release_session_guard(sk, guard=new_guard)
    +        assert sk not in adapter._active_sessions
    +
    +    def test_release_session_guard_without_guard_releases_unconditionally(self):
    +        adapter = _make_adapter()
    +        sk = _session_key()
    +        adapter._active_sessions[sk] = asyncio.Event()
    +        # Callers that don't know the guard (e.g. cancel_session_processing's
    +        # default path) still work.
    +        adapter._release_session_guard(sk)
    +        assert sk not in adapter._active_sessions
    +