"""Call-site wiring guard for the internal-event session-context pin. ``_pinned_session_context_prompt(..., internal=True)`` reuses the existing pin verbatim so an internal event (kanban wake, delegation completion, watch notification) cannot re-key it. A helper-level unit test would stay green if the call site in ``_handle_message_with_agent`` stopped forwarding ``event.internal``, so this test drives the REAL handler through human -> internal -> human on one session and asserts the context prompt that reaches ``_run_agent`` is byte-identical on all three turns. """ from __future__ import annotations import sys import types from datetime import datetime from unittest.mock import AsyncMock, MagicMock import pytest from gateway.config import ChannelOverride, GatewayConfig, Platform, PlatformConfig from gateway.platforms.base import MessageEvent from gateway.run_turn_runner import TurnRunner from gateway.session import SessionEntry, SessionSource from gateway.turn_context import TurnContext KEY = "agent:main:discord:group:1513247605675790346:117431298246705156" _ORIGIN = dict( platform=Platform.DISCORD, chat_id="1513247605675790346", chat_type="group", user_id="117431298246705156", scope_id="1480524732964278294", ) def _human_source() -> SessionSource: return SessionSource( **_ORIGIN, chat_name="Guild / #general", user_name="Ace", message_id="1552671843494666330", ) def _wake_source() -> SessionSource: # kanban_watchers._push_wake shape: rebuilt from the persisted origin, so # no chat_name / user_name / message_id. return SessionSource(**_ORIGIN) def _make_runner( monkeypatch, config: GatewayConfig | None = None, *, durable_prompt_pin: dict | None = None, ): import agent.model_metadata as mm import gateway.run as gr import gateway.session as gs monkeypatch.setattr(gs, "_discord_tools_loaded", lambda: True) monkeypatch.setattr(gr, "_resolve_runtime_agent_kwargs", lambda: {"api_key": "fake"}) monkeypatch.setattr(mm, "get_model_context_length", lambda *a, **k: 100_000) r = gr.GatewayRunner(config or GatewayConfig()) r.adapters = {} r._running_agents = {} r._running_agents_ts = {} r._pending_messages = {} r._pending_approvals = {} r._is_user_authorized = lambda s: True r._set_session_env = lambda c: None r._handle_active_session_busy_message = AsyncMock(return_value=False) r._session_db = MagicMock() r._recover_telegram_topic_thread_id = lambda s: None r._cache_session_source = lambda k, s: None r._is_session_run_current = lambda k, g: True r._begin_session_run_generation = lambda k: 1 r._reply_anchor_for_event = lambda e: None r._get_guild_id = lambda e: None r._should_send_voice_reply = lambda *a, **k: False r.hooks = MagicMock() r.hooks.emit = AsyncMock() # The turn lease is released by the real turn tail, which _run_agent's # stub skips; disable leasing so three sequential turns can run. r._turn_leases = None store = MagicMock() store.get_or_create_session.return_value = SessionEntry( session_key=KEY, session_id="sess-wiring", created_at=datetime(2026, 1, 1), updated_at=datetime(2026, 1, 2), platform=Platform.DISCORD, chat_type="group", ) store.load_transcript.return_value = [] store.has_platform_message_id.return_value = False if durable_prompt_pin is not None: def _get_prompt_pin(_key, *, expected_session_id=None): if expected_session_id is not None: assert expected_session_id == "sess-wiring" value = durable_prompt_pin.get("value") return dict(value) if isinstance(value, dict) else None def _set_prompt_pin(_key, value, *, expected_session_id=None): assert expected_session_id == "sess-wiring" durable_prompt_pin["value"] = dict(value) if isinstance(value, dict) else None return True store.get_prompt_pin.side_effect = _get_prompt_pin store.set_prompt_pin.side_effect = _set_prompt_pin r.session_store = store return r def _capture(runner, sink: list): async def fake_run_agent(**kw): sink.append(kw) return { "final_response": "ok", "messages": [], "tools": [], "history_offset": 0, "last_prompt_tokens": 0, } runner._run_agent = fake_run_agent async def _drive(runner, turns, *, channel_prompt=None): for internal, src in turns: event = MessageEvent( text="[kanban] wake" if internal else "hi", source=src, message_id=None if internal else src.message_id, internal=internal, # Adapters resolve channel_prompts onto human events; the kanban # wake is built without one. channel_prompt=None if internal else channel_prompt, ) await runner._handle_message_with_agent(event, src, KEY, 1) @pytest.mark.asyncio async def test_internal_event_reuses_pin_through_real_handler(monkeypatch): runner = _make_runner(monkeypatch) calls: list[dict] = [] _capture(runner, calls) # Internal-first on a fresh session (kanban/API-server wake, startup resume): # with no pin to reuse it must still render AND pin, or every internal-only # turn re-renders and loses verbatim reuse. await _drive(runner, ((True, _wake_source()),)) assert runner._peek_session_state(KEY).conversation.ephemeral_pin is not None await _drive(runner, ((False, _human_source()), (True, _wake_source()), (False, _human_source()))) seen = [kw["context_prompt"] for kw in calls[1:]] assert len(seen) == 3, f"_run_agent reached {len(seen)}/3 turns" # The human render names the chat; the wake-shaped source cannot. If the # internal turn re-rendered, its bytes would differ and the next human # turn would re-key back (A->B->A). assert "Guild / #general" in seen[0] assert seen[0] == seen[1] == seen[2], "internal event re-keyed the session-context pin" # --------------------------------------------------------------------------- # The other ephemeral prompt components must not toggle across the sequence either. # --------------------------------------------------------------------------- THREAD_ID = "1552000000000000001" PARENT_ID = "1513247605675790346" def _effective_ephemeral(runner, kw) -> str: """The real TurnRunner combiner applied to what reached ``_run_agent``.""" ctx = TurnContext(source=kw["source"], context_prompt=kw["context_prompt"], channel_prompt=kw.get("channel_prompt")) return TurnRunner(runner, ctx)._combined_ephemeral_prompt() def _thread_origin() -> dict: return dict( platform=Platform.DISCORD, chat_id=THREAD_ID, chat_type="thread", thread_id=THREAD_ID, user_id="117431298246705156", scope_id="1480524732964278294", ) def _human_thread_source() -> SessionSource: return SessionSource( **_thread_origin(), parent_chat_id=PARENT_ID, chat_name="Guild / #dev / build thread", user_name="Ace", message_id="1552671843494666331", ) def _wake_thread_source() -> SessionSource: # _push_wake rebuilds chat_id/thread_id from the subscription; it has no # parent_chat_id, chat_name, user_name or message_id. return SessionSource(**_thread_origin()) @pytest.mark.asyncio async def test_internal_event_keeps_channel_prompt_and_parent_override(monkeypatch): config = GatewayConfig() config.platforms[Platform.DISCORD] = PlatformConfig( enabled=True, channel_overrides={PARENT_ID: ChannelOverride(system_prompt="Parent persona.")}, ) runner = _make_runner(monkeypatch, config) calls: list[dict] = [] _capture(runner, calls) await _drive( runner, ((False, _human_thread_source()), (True, _wake_thread_source()), (False, _human_thread_source())), channel_prompt="Channel hint.", ) assert len(calls) == 3 eph = [_effective_ephemeral(runner, kw) for kw in calls] assert "Channel hint." in eph[0] and "Parent persona." in eph[0] assert eph[0] == eph[1] == eph[2], "internal event toggled the channel ephemeral components" @pytest.mark.asyncio @pytest.mark.parametrize( ("result_key", "interrupted"), (("pending_steer", False), ("interrupt_message", True)), ids=("leftover-steer", "interrupt-text"), ) @pytest.mark.parametrize("channel_prompt", ["Channel hint.", "", None], ids=("hint", "empty", "none")) async def test_eventless_followup_keeps_effective_prompt_through_next_human( monkeypatch, result_key, interrupted, channel_prompt ): config = GatewayConfig() config.platforms[Platform.DISCORD] = PlatformConfig( enabled=True, channel_overrides={PARENT_ID: ChannelOverride(system_prompt="Parent persona.")}, ) runner = _make_runner(monkeypatch, config) calls: list[dict] = [] _capture(runner, calls) runner._run_agent_deliver_first_response = AsyncMock() runner._refresh_agent_cache_message_count = AsyncMock() adapter = MagicMock() adapter.get_pending_message.return_value = None adapter._active_sessions = {} source = _human_thread_source() await _drive(runner, ((False, source),), channel_prompt=channel_prompt) first = calls[0] turn_ctx = TurnContext( source=first["source"], context_prompt=first["context_prompt"], channel_prompt=first["channel_prompt"], session_key=first["session_key"], session_id=first["session_id"], run_generation=1, history=[], ) result = { "final_response": "done", "messages": [], result_key: "follow up", "interrupted": interrupted, } pending_event, pending = await runner._run_agent_drain_pending(result, adapter, source, KEY) assert pending_event is None assert pending == "follow up" await runner._run_agent_queued_followup( turn_ctx, adapter, pending, pending_event, "done", result, None ) await _drive(runner, ((False, source),), channel_prompt=channel_prompt) assert [call["channel_prompt"] for call in calls] == [channel_prompt] * 3 ephemeral = [_effective_ephemeral(runner, call) for call in calls] assert "Parent persona." in ephemeral[0] assert ephemeral[0] == ephemeral[1] == ephemeral[2] @pytest.mark.asyncio async def test_event_backed_followup_overrides_inherited_channel_prompt(monkeypatch): runner = _make_runner(monkeypatch) calls: list[dict] = [] _capture(runner, calls) runner._run_agent_deliver_first_response = AsyncMock() runner._refresh_agent_cache_message_count = AsyncMock() runner._prepare_profile_scoped_inbound_message_text = AsyncMock(return_value="queued") runner._session_key_for_source = lambda source: KEY source = _human_source() adapter = MagicMock() adapter._active_sessions = {} pending_event = MessageEvent( text="queued", source=source, message_id="queued-message-1", channel_prompt="Queued event prompt.", ) turn_ctx = TurnContext( source=source, context_prompt="ctx", channel_prompt="Inherited prompt.", session_key=KEY, session_id="sess-wiring", run_generation=1, history=[], ) result = {"final_response": "done", "messages": []} await runner._run_agent_queued_followup( turn_ctx, adapter, "queued", pending_event, "done", result, None ) assert len(calls) == 1 assert calls[0]["channel_prompt"] == "Queued event prompt.", "event prompt must win over the inherited one" @pytest.mark.asyncio async def test_first_internal_event_after_restart_rehydrates_durable_prompt_pins(monkeypatch): config = GatewayConfig() config.platforms[Platform.DISCORD] = PlatformConfig( enabled=True, channel_overrides={PARENT_ID: ChannelOverride(system_prompt="Parent persona.")}, ) durable: dict = {} before = _make_runner(monkeypatch, config, durable_prompt_pin=durable) calls_before: list[dict] = [] _capture(before, calls_before) await _drive(before, ((False, _human_thread_source()),), channel_prompt="Channel hint.") persisted = durable.get("value") assert isinstance(persisted, dict) assert persisted["channel_prompt"] == "Channel hint." assert persisted["parent_chat_id"] == PARENT_ID # A brand-new runner has no ConversationState. The first post-restart event is internal. after = _make_runner(monkeypatch, config, durable_prompt_pin=durable) calls_after: list[dict] = [] _capture(after, calls_after) await _drive( after, ((True, _wake_thread_source()), (False, _human_thread_source())), channel_prompt="Channel hint.", ) human, first_internal, next_human = calls_before[0], calls_after[0], calls_after[1] assert after._peek_session_state(KEY).conversation.ephemeral_pin is not None assert after._peek_session_state(KEY).conversation.channel_pin is not None assert first_internal["context_prompt"] == human["context_prompt"] == next_human["context_prompt"] assert first_internal["channel_prompt"] == human["channel_prompt"] == next_human["channel_prompt"] assert ( first_internal["source"].parent_chat_id == human["source"].parent_chat_id == next_human["source"].parent_chat_id ) assert _effective_ephemeral(after, first_internal) == _effective_ephemeral(before, human) @pytest.mark.asyncio async def test_first_internal_event_after_agent_eviction_keeps_effective_prompt(monkeypatch): """/stop, /undo and /model evict the cached agent mid-conversation and clear the context pin while the channel pin survives; a wake arriving next must still send the human turn's bytes.""" config = GatewayConfig() config.platforms[Platform.DISCORD] = PlatformConfig( enabled=True, channel_overrides={PARENT_ID: ChannelOverride(system_prompt="Parent persona.")}, ) runner = _make_runner(monkeypatch, config, durable_prompt_pin={}) calls: list[dict] = [] _capture(runner, calls) await _drive(runner, ((False, _human_thread_source()),), channel_prompt="Channel hint.") runner._evict_cached_agent(KEY) await _drive( runner, ((True, _wake_thread_source()), (False, _human_thread_source())), channel_prompt="Channel hint.", ) assert len(calls) == 3 eph = [_effective_ephemeral(runner, kw) for kw in calls] assert "Guild / #dev / build thread" in eph[0] assert eph[0] == eph[1] == eph[2], "wake after eviction re-rendered the session context (A->B->A)" def _synthetic_turn(runner, source, kind: str) -> MessageEvent: """The event each real producer builds: goal continuation / heartbeat via ``_synthetic_prompt_event``, ``/goal resume`` via ``_enqueue_goal_turn``.""" if kind == "goal-resume": enqueued: list[MessageEvent] = [] runner._adapter_and_key_for = lambda _event: (object(), KEY) runner._enqueue_fifo = lambda _key, turn, _adapter: enqueued.append(turn) command = MessageEvent(text="/goal resume", source=source, channel_prompt="Channel hint.") runner._enqueue_goal_turn(command, "[goal] resume", label="resume", kickoff=False) return enqueued[0] if kind == "heartbeat-rebuilt-route": # A watch route rebuilt from a stored origin can lack the chat/user names the human turn # rendered; the context pin must still be reused, not re-rendered from it. source = _wake_thread_source() return runner._synthetic_prompt_event(source, "[goal] continue") @pytest.mark.asyncio @pytest.mark.parametrize("kind", ["goal-continuation", "goal-resume", "heartbeat-rebuilt-route"]) async def test_synthetic_goal_turn_keeps_pins_and_effective_prompt(monkeypatch, kind): """Goal continuation, heartbeat due-prompts and ``/goal resume`` are non-internal (they keep authorization and the emergency stop) but carry no channel prompt of their own; they must reuse the human turn's pins instead of overwriting the channel pin with None (A->B->A).""" config = GatewayConfig() config.platforms[Platform.DISCORD] = PlatformConfig( enabled=True, channel_overrides={PARENT_ID: ChannelOverride(system_prompt="Parent persona.")}, ) runner = _make_runner(monkeypatch, config) calls: list[dict] = [] _capture(runner, calls) human = _human_thread_source() await _drive(runner, ((False, human),), channel_prompt="Channel hint.") synthetic = _synthetic_turn(runner, human, kind) assert not synthetic.internal and synthetic.channel_prompt is None if kind == "goal-continuation": # A continuation queued behind the running turn drains through the in-band follow-up. runner._run_agent_deliver_first_response = AsyncMock() runner._refresh_agent_cache_message_count = AsyncMock() runner._session_key_for_source = lambda _source: KEY adapter = MagicMock(_active_sessions={}, _pending_messages={KEY: synthetic}) adapter.get_pending_message = lambda key: adapter._pending_messages.pop(key, None) result = {"final_response": "done", "messages": []} pending_event, pending = await runner._run_agent_drain_pending(result, adapter, human, KEY) assert pending_event is synthetic first = calls[0] turn_ctx = TurnContext(source=first["source"], context_prompt=first["context_prompt"], channel_prompt=first["channel_prompt"], session_key=KEY, session_id=first["session_id"], run_generation=1, history=[]) await runner._run_agent_queued_followup(turn_ctx, adapter, pending, pending_event, "done", result, None) else: await runner._handle_message_with_agent(synthetic, synthetic.source, KEY, 1) assert runner._peek_session_state(KEY).conversation.channel_pin[0] == "Channel hint." await _drive(runner, ((False, human),), channel_prompt="Channel hint.") assert [kw["channel_prompt"] for kw in calls] == ["Channel hint."] * 3 eph = [_effective_ephemeral(runner, kw) for kw in calls] assert "Channel hint." in eph[0] and "Parent persona." in eph[0] assert eph[0] == eph[1] == eph[2], "synthetic turn toggled the effective ephemeral prompt" @pytest.mark.asyncio @pytest.mark.parametrize("origin_message_id", [None, "m-origin"], ids=("no-msgid", "msgid")) async def test_goal_continuation_keeps_channel_pin_through_session_authority(tmp_path, monkeypatch, origin_message_id): """The default runtime admits adapter turns through the native session envelope; a goal continuation must arrive at the handler still asking to reuse the pins (#126109). The continuation is enqueued mid-turn like the post-turn hook does and drains from the adapter FIFO; a source carrying ``message_id`` (Discord) must keep its adapter ref through the producer's copy or native admission refuses it as ``not_found``.""" import gateway.run as gr from gateway.platforms.base import SendResult from gateway.session_authority import initialize_session_authority from plugins.platforms.whatsapp.adapter import WhatsAppAdapter from tests.waiting import async_wait_until chat = "15551234567@s.whatsapp.net" monkeypatch.setenv("WHATSAPP_ALLOWED_USERS", "15551234567") monkeypatch.setattr(gr, "_resolve_runtime_agent_kwargs", lambda: {"api_key": "fake"}) runner = gr.GatewayRunner(GatewayConfig(sessions_dir=tmp_path / "sessions")) adapter = WhatsAppAdapter(PlatformConfig(enabled=True, extra={ "text_batch_delay_seconds": 0, "session_path": str(tmp_path / "wa")})) adapter.gateway_runner, runner.adapters = runner, {Platform.WHATSAPP: adapter} authority = await initialize_session_authority(runner, profile_id="default", instance_id="pins") runner._wire_adapter_handlers(adapter) calls: list[dict] = [] _capture(runner, calls) monkeypatch.setattr(adapter, "send", AsyncMock(return_value=SendResult(success=True, message_id="s"))) source = adapter.build_source(chat_id=chat, user_id=chat, chat_type="dm", message_id=origin_message_id) run_agent = runner._run_agent async def run_and_continue(**kw): if not calls: # the goal judge said continue: enqueue while this turn still runs runner._enqueue_fifo(runner._session_key_for_source(source), runner._synthetic_prompt_event( source, "[Continuing toward your standing goal]\nGoal: x"), adapter) return await run_agent(**kw) runner._run_agent = run_and_continue turns = ( (2, MessageEvent(text="hi", source=source, message_id="h1", channel_prompt="Channel hint.")), (3, MessageEvent(text="hi again", source=source, message_id="h2", channel_prompt="Channel hint.")), ) try: for want, event in turns: await adapter.handle_message(event) await async_wait_until( lambda: len(calls) >= want and not adapter._background_tasks and not adapter._session_tasks, timeout=60, what=f"turn {want}") finally: authority.db.close() assert [kw["channel_prompt"] for kw in calls] == ["Channel hint."] * 3 pin = runner._peek_session_state(runner._session_key_for_source(source)).conversation.channel_pin assert pin[0] == "Channel hint." @pytest.mark.asyncio async def test_human_turn_without_channel_prompt_still_replaces_the_pin(monkeypatch): """Negative control: only synthetic turns inherit. A human turn whose channel prompt was removed must still clear it, or a deleted channel prompt would stick forever.""" runner = _make_runner(monkeypatch) calls: list[dict] = [] _capture(runner, calls) await _drive(runner, ((False, _human_thread_source()),), channel_prompt="Channel hint.") await _drive(runner, ((False, _human_thread_source()),), channel_prompt=None) assert [kw["channel_prompt"] for kw in calls] == ["Channel hint.", None] assert runner._peek_session_state(KEY).conversation.channel_pin[0] is None @pytest.mark.asyncio async def test_human_first_after_restart_ignores_stale_durable_prompt_pin(monkeypatch): durable = { "value": { "version": 1, "context_key": "stale-key", "context_prompt": "STALE CONTEXT", "redact_pii": False, "channel_prompt": "Stale channel prompt.", "parent_chat_id": "stale-parent", } } config = GatewayConfig() config.platforms[Platform.DISCORD] = PlatformConfig( enabled=True, channel_overrides={PARENT_ID: ChannelOverride(system_prompt="Parent persona.")}, ) runner = _make_runner(monkeypatch, config, durable_prompt_pin=durable) calls: list[dict] = [] _capture(runner, calls) await _drive(runner, ((False, _human_thread_source()),), channel_prompt="Fresh channel prompt.") assert len(calls) == 1 assert calls[0]["context_prompt"] != "STALE CONTEXT" assert calls[0]["channel_prompt"] == "Fresh channel prompt." assert calls[0]["source"].parent_chat_id == PARENT_ID assert durable["value"]["context_prompt"] == calls[0]["context_prompt"] assert durable["value"]["channel_prompt"] == "Fresh channel prompt." assert durable["value"]["parent_chat_id"] == PARENT_ID @pytest.mark.asyncio async def test_internal_event_never_reuses_prompt_pin_from_another_privacy_policy(monkeypatch): """A pin rendered with redact_pii off must not reach the model once redaction is on, not even through the internal-event reuse path after a restart.""" import gateway.run as gr monkeypatch.setattr(gr, "_load_gateway_config", lambda: {"privacy": {"redact_pii": True}}) durable = { "value": { "version": 1, "context_key": "unredacted-key", "context_prompt": "UNREDACTED CONTEXT", "redact_pii": False, "channel_prompt": "Channel hint.", "parent_chat_id": None, } } runner = _make_runner(monkeypatch, durable_prompt_pin=durable) calls: list[dict] = [] _capture(runner, calls) await _drive(runner, ((True, _wake_source()),)) assert calls[0]["context_prompt"] != "UNREDACTED CONTEXT" assert calls[0]["channel_prompt"] == "Channel hint." # --------------------------------------------------------------------------- # #131294 — a channel_overrides model is the session's CONFIGURED model, not a # fallback. The post-turn ``_run_agent_evict_on_fallback`` check baselined # against the global model only, so every successful turn in an overridden chat # evicted the cached agent (and cleared its ephemeral pin). These drive the # REAL ``_run_agent`` / ``TurnRunner.run_sync`` path (only the AIAgent class and # config reads are substituted), not the ``_capture`` stub above. # --------------------------------------------------------------------------- _WELCOME = "https://welcome-api.nousresearch.com/v1" class _TurnAgent: """AIAgent stand-in: keeps the resolved route, answers without an API call.""" instances = 0 def __init__(self, *args, **kwargs): type(self).instances += 1 self.model = kwargs.get("model") self.provider = kwargs.get("provider") self.base_url = kwargs.get("base_url") self.session_id = kwargs.get("session_id") self.tools = [] self.request_overrides = dict(kwargs.get("request_overrides") or {}) # Mirror AIAgent.__init__: the welcome host pins its one model, and the primary # route is snapshotted for the post-turn fallback classifier. from hermes_cli.anon_auth import pin_model_for_route self.model = pin_model_for_route(self.provider, self.base_url, self.model) self._primary_runtime = {"model": self.model, "provider": self.provider, "base_url": self.base_url} def run_conversation(self, user_message, conversation_history=None, task_id=None, **kwargs): return { "final_response": "ok", "messages": (conversation_history or []) + [ {"role": "user", "content": user_message}, {"role": "assistant", "content": "ok"}, ], "api_calls": 1, } @pytest.mark.parametrize( "runtime, global_model, expected_model", [ ({"api_key": "fake", "provider": "openrouter"}, "default/model", "chan/model"), ({"api_key": "fake", "provider": "nous", "base_url": _WELCOME}, "nous/welcome", "nous/welcome"), ], ids=["named-channel-model", "welcome-host-pins-over-channel-model"], ) @pytest.mark.asyncio async def test_channel_override_turns_keep_one_cached_agent_and_the_pin( monkeypatch, runtime, global_model, expected_model, ): import gateway.run as gr config = GatewayConfig() config.platforms[Platform.DISCORD] = PlatformConfig( enabled=True, channel_overrides={_ORIGIN["chat_id"]: ChannelOverride(model="chan/model")}, ) monkeypatch.setattr(gr, "_load_gateway_config", lambda: {"model": {"default": global_model}}) monkeypatch.setattr(gr, "_resolve_gateway_model", lambda config=None: global_model) fake_run_agent = types.ModuleType("run_agent") fake_run_agent.AIAgent = _TurnAgent monkeypatch.setitem(sys.modules, "run_agent", fake_run_agent) _TurnAgent.instances = 0 runner = _make_runner(monkeypatch, config) monkeypatch.setattr(gr, "_resolve_runtime_agent_kwargs", lambda: dict(runtime)) runner.session_store._entries = {} await _drive(runner, ((False, _human_source()), (True, _wake_source()), (False, _human_source()))) entry = runner._agent_cache.get(KEY) assert entry is not None, "channel-override turn evicted the session's cached agent" assert entry[0].model == expected_model assert _TurnAgent.instances == 1, f"agent rebuilt {_TurnAgent.instances - 1}x across 3 turns" assert runner._peek_session_state(KEY).conversation.ephemeral_pin is not None, "context pin cleared"