From 160a397c48c30cf9630ad8a206274bde386b087d Mon Sep 17 00:00:00 2001 From: Rod Boev Date: Wed, 1 Jul 2026 15:35:08 -0400 Subject: [PATCH 1/4] fix(#4251): stop in-flight turns from reverting picker model choice --- api/streaming.py | 20 +- ...test_issue4251_wakeup_model_picker_race.py | 191 ++++++++++++++++++ 2 files changed, 206 insertions(+), 5 deletions(-) create mode 100644 tests/test_issue4251_wakeup_model_picker_race.py diff --git a/api/streaming.py b/api/streaming.py index e752bc8d..7c0c9f2e 100644 --- a/api/streaming.py +++ b/api/streaming.py @@ -6694,13 +6694,19 @@ def _run_agent_streaming( s = get_session(session_id) update_active_run(stream_id, phase="running", session_id=session_id) s.workspace = str(Path(workspace).expanduser().resolve()) - s.model = model + _last_persisted_model = getattr(s, "model", None) provider_context = ( str(model_provider).strip().lower() if model_provider is not None else getattr(s, "model_provider", None) ) - s.model_provider = provider_context or None + # #4251: the route layer already persisted this turn's model under the + # session lock before dispatch, so a mismatch here means a newer picker + # write won the race and must not be clobbered by the worker thread. + if _last_persisted_model in (None, "", model): + s.model = model + s.model_provider = provider_context or None + _last_persisted_model = model _agent_lock = _get_session_agent_lock(session_id) # TD1: set thread-local env context so concurrent sessions don't clobber globals @@ -6742,9 +6748,13 @@ def _run_agent_streaming( profile_home=_profile_home, has_profile=bool(getattr(s, "profile", None)), ) - s.model_provider = provider_context - if _repaired and model != (s.model or ""): - s.model = model + # #4251: only apply the profile-repair persistence if this turn still + # owns the session model value it last wrote. + if getattr(s, "model", None) == _last_persisted_model: + s.model_provider = provider_context + if _repaired and model != (s.model or ""): + s.model = model + _last_persisted_model = model # Capture the resolved profile name now, while profile context is # reliable. Used in the compression migration block to stamp s.profile diff --git a/tests/test_issue4251_wakeup_model_picker_race.py b/tests/test_issue4251_wakeup_model_picker_race.py new file mode 100644 index 00000000..2f2ba5e7 --- /dev/null +++ b/tests/test_issue4251_wakeup_model_picker_race.py @@ -0,0 +1,191 @@ +"""Regression tests for wakeup model picker races (#4251).""" + +import queue +import sys +import types +from unittest import mock + +import api.oauth +import api.streaming as streaming + + +class FakeSession: + def __init__(self, *, model, model_provider="anthropic"): + self.session_id = "sess-4251" + self.title = "Wakeup picker race" + self.workspace = "/tmp" + self.model = model + self.model_provider = model_provider + self.messages = [] + self.personality = None + self.profile = None + self.input_tokens = 0 + self.output_tokens = 0 + self.estimated_cost = None + self.tool_calls = [] + self.active_stream_id = None + self.pending_user_message = None + self.pending_attachments = [] + self.pending_started_at = None + + def save(self, touch_updated_at=True, **kwargs): + self._saved = touch_updated_at + + def compact(self): + return { + "session_id": self.session_id, + "title": self.title, + "workspace": self.workspace, + "model": self.model, + "created_at": 0, + "updated_at": 0, + "pinned": False, + "archived": False, + "project_id": None, + "profile": self.profile, + "input_tokens": self.input_tokens, + "output_tokens": self.output_tokens, + "estimated_cost": self.estimated_cost, + "personality": self.personality, + } + + +class CapturingAgent: + def __init__( + self, + model=None, + provider=None, + base_url=None, + api_key=None, + platform=None, + quiet_mode=False, + enabled_toolsets=None, + fallback_model=None, + session_id=None, + session_db=None, + stream_delta_callback=None, + reasoning_callback=None, + tool_progress_callback=None, + clarify_callback=None, + **kwargs, + ): + self.init_kwargs = { + "model": model, + "provider": provider, + "base_url": base_url, + "api_key": api_key, + "session_id": session_id, + "session_db": session_db, + } + self.session_id = session_id + self.context_compressor = None + self.session_prompt_tokens = 0 + self.session_completion_tokens = 0 + self.session_estimated_cost_usd = None + self.reasoning_config = None + self.ephemeral_system_prompt = None + self._last_error = None + + def run_conversation(self, **kwargs): + return { + "messages": [ + {"role": "user", "content": kwargs["persist_user_message"]}, + {"role": "assistant", "content": "ok"}, + ] + } + + def interrupt(self, _message): + return None + + +def _install_streaming_harness(monkeypatch, fake_session): + fake_runtime_module = types.ModuleType("hermes_cli.runtime_provider") + fake_runtime_module.resolve_runtime_provider = mock.Mock( + return_value={ + "provider": "anthropic", + "base_url": None, + "api_key": "rt-key", + } + ) + fake_hermes_cli = types.ModuleType("hermes_cli") + fake_hermes_cli.runtime_provider = fake_runtime_module + fake_hermes_state = types.ModuleType("hermes_state") + fake_hermes_state.SessionDB = mock.Mock(return_value=object()) + + def fake_runtime_lock(resolver, **kwargs): + return resolver(**kwargs) + + monkeypatch.setattr( + api.oauth, + "resolve_runtime_provider_with_anthropic_env_lock", + fake_runtime_lock, + ) + monkeypatch.setattr(streaming, "get_session", lambda _session_id: fake_session) + monkeypatch.setattr(streaming, "_get_ai_agent", lambda: CapturingAgent) + monkeypatch.setattr( + streaming, + "resolve_model_provider", + lambda *_args, **_kwargs: ("haiku-4-5", "anthropic", None), + ) + monkeypatch.setattr("api.config.get_config", lambda: {}) + monkeypatch.setattr("api.config._resolve_cli_toolsets", lambda *_args, **_kwargs: []) + monkeypatch.setitem(sys.modules, "hermes_cli", fake_hermes_cli) + monkeypatch.setitem(sys.modules, "hermes_cli.runtime_provider", fake_runtime_module) + monkeypatch.setitem(sys.modules, "hermes_state", fake_hermes_state) + + +def _run_streaming_turn(monkeypatch, fake_session, *, stream_id): + _install_streaming_harness(monkeypatch, fake_session) + fake_session.active_stream_id = stream_id + fake_queue = queue.Queue() + try: + streaming.STREAMS[stream_id] = fake_queue + streaming._run_agent_streaming( + session_id=fake_session.session_id, + msg_text="background wakeup turn", + model="haiku-4-5", + model_provider="anthropic", + workspace="/tmp", + stream_id=stream_id, + ) + finally: + streaming.STREAMS.pop(stream_id, None) + streaming.AGENT_INSTANCES.pop(stream_id, None) + + +def test_dispatch_stamp_does_not_clobber_newer_picker_model(monkeypatch): + fake_session = FakeSession(model="opus-4-8", model_provider="anthropic") + + _run_streaming_turn( + monkeypatch, + fake_session, + stream_id="stream-4251-race", + ) + + assert fake_session.model == "opus-4-8" + + +def test_dispatch_stamp_persists_resolved_model_when_no_race(monkeypatch): + fake_session = FakeSession(model="haiku-4-5", model_provider=None) + + _run_streaming_turn( + monkeypatch, + fake_session, + stream_id="stream-4251-steady", + ) + + assert fake_session.model == "haiku-4-5" + assert fake_session.model_provider == "anthropic" + + +def test_dispatch_stamp_persists_when_session_model_was_empty(monkeypatch): + fake_session = FakeSession(model="", model_provider=None) + + _run_streaming_turn( + monkeypatch, + fake_session, + stream_id="stream-4251-empty", + ) + + assert fake_session.model == "haiku-4-5" + assert fake_session.model_provider == "anthropic" From 5c7ff9c70e2c29f6f48264c422aedc5de5036f1a Mon Sep 17 00:00:00 2001 From: Rod Boev Date: Wed, 1 Jul 2026 15:42:57 -0400 Subject: [PATCH 2/4] fix(#4251): preserve raced picker ownership through profile repair --- api/streaming.py | 4 ++- ...test_issue4251_wakeup_model_picker_race.py | 35 ++++++++++++++++--- 2 files changed, 34 insertions(+), 5 deletions(-) diff --git a/api/streaming.py b/api/streaming.py index 7c0c9f2e..8a3d756a 100644 --- a/api/streaming.py +++ b/api/streaming.py @@ -6695,6 +6695,7 @@ def _run_agent_streaming( update_active_run(stream_id, phase="running", session_id=session_id) s.workspace = str(Path(workspace).expanduser().resolve()) _last_persisted_model = getattr(s, "model", None) + _turn_owns_persisted_model = False provider_context = ( str(model_provider).strip().lower() if model_provider is not None @@ -6707,6 +6708,7 @@ def _run_agent_streaming( s.model = model s.model_provider = provider_context or None _last_persisted_model = model + _turn_owns_persisted_model = True _agent_lock = _get_session_agent_lock(session_id) # TD1: set thread-local env context so concurrent sessions don't clobber globals @@ -6750,7 +6752,7 @@ def _run_agent_streaming( ) # #4251: only apply the profile-repair persistence if this turn still # owns the session model value it last wrote. - if getattr(s, "model", None) == _last_persisted_model: + if _turn_owns_persisted_model and getattr(s, "model", None) == _last_persisted_model: s.model_provider = provider_context if _repaired and model != (s.model or ""): s.model = model diff --git a/tests/test_issue4251_wakeup_model_picker_race.py b/tests/test_issue4251_wakeup_model_picker_race.py index 2f2ba5e7..dfb1385f 100644 --- a/tests/test_issue4251_wakeup_model_picker_race.py +++ b/tests/test_issue4251_wakeup_model_picker_race.py @@ -134,7 +134,14 @@ def _install_streaming_harness(monkeypatch, fake_session): monkeypatch.setitem(sys.modules, "hermes_state", fake_hermes_state) -def _run_streaming_turn(monkeypatch, fake_session, *, stream_id): +def _run_streaming_turn( + monkeypatch, + fake_session, + *, + stream_id, + dispatch_model="haiku-4-5", + dispatch_provider="anthropic", +): _install_streaming_harness(monkeypatch, fake_session) fake_session.active_stream_id = stream_id fake_queue = queue.Queue() @@ -143,8 +150,8 @@ def _run_streaming_turn(monkeypatch, fake_session, *, stream_id): streaming._run_agent_streaming( session_id=fake_session.session_id, msg_text="background wakeup turn", - model="haiku-4-5", - model_provider="anthropic", + model=dispatch_model, + model_provider=dispatch_provider, workspace="/tmp", stream_id=stream_id, ) @@ -154,7 +161,7 @@ def _run_streaming_turn(monkeypatch, fake_session, *, stream_id): def test_dispatch_stamp_does_not_clobber_newer_picker_model(monkeypatch): - fake_session = FakeSession(model="opus-4-8", model_provider="anthropic") + fake_session = FakeSession(model="opus-4-8", model_provider="openrouter") _run_streaming_turn( monkeypatch, @@ -163,6 +170,7 @@ def test_dispatch_stamp_does_not_clobber_newer_picker_model(monkeypatch): ) assert fake_session.model == "opus-4-8" + assert fake_session.model_provider == "openrouter" def test_dispatch_stamp_persists_resolved_model_when_no_race(monkeypatch): @@ -189,3 +197,22 @@ def test_dispatch_stamp_persists_when_session_model_was_empty(monkeypatch): assert fake_session.model == "haiku-4-5" assert fake_session.model_provider == "anthropic" + + +def test_profile_repair_skips_persistence_when_newer_picker_choice_already_won(monkeypatch): + fake_session = FakeSession(model="opus-4-8", model_provider="openrouter") + fake_session.profile = "worker-profile" + monkeypatch.setattr( + streaming, + "_apply_profile_home_context_to_streaming_model", + lambda **_kwargs: ("claude-profile-default", "anthropic", True), + ) + + _run_streaming_turn( + monkeypatch, + fake_session, + stream_id="stream-4251-profile-race", + ) + + assert fake_session.model == "opus-4-8" + assert fake_session.model_provider == "openrouter" From a609b99e7670e5e8132be1032139f986968334fc Mon Sep 17 00:00:00 2001 From: Rod Boev Date: Wed, 1 Jul 2026 16:19:53 -0400 Subject: [PATCH 3/4] fix(#4251): guard provider ownership under the session lock --- api/streaming.py | 48 ++++++++++++++----- ...test_issue4251_wakeup_model_picker_race.py | 41 ++++++++++++++++ 2 files changed, 76 insertions(+), 13 deletions(-) diff --git a/api/streaming.py b/api/streaming.py index 8a3d756a..161fe5da 100644 --- a/api/streaming.py +++ b/api/streaming.py @@ -6694,23 +6694,35 @@ def _run_agent_streaming( s = get_session(session_id) update_active_run(stream_id, phase="running", session_id=session_id) s.workspace = str(Path(workspace).expanduser().resolve()) - _last_persisted_model = getattr(s, "model", None) + _last_persisted_model = None + _last_persisted_provider = None _turn_owns_persisted_model = False provider_context = ( str(model_provider).strip().lower() if model_provider is not None else getattr(s, "model_provider", None) ) + provider_context = str(provider_context).strip().lower() if provider_context else None + _agent_lock = _get_session_agent_lock(session_id) # #4251: the route layer already persisted this turn's model under the # session lock before dispatch, so a mismatch here means a newer picker # write won the race and must not be clobbered by the worker thread. - if _last_persisted_model in (None, "", model): - s.model = model - s.model_provider = provider_context or None - _last_persisted_model = model - _turn_owns_persisted_model = True + with _agent_lock: + _last_persisted_model = getattr(s, "model", None) + _last_persisted_provider = getattr(s, "model_provider", None) + if _last_persisted_provider is not None: + _last_persisted_provider = str(_last_persisted_provider).strip().lower() or None + _persisted_model_is_empty = _last_persisted_model in (None, "") + _provider_matches = _last_persisted_provider in (None, provider_context) + if _persisted_model_is_empty or ( + _last_persisted_model == model and _provider_matches + ): + s.model = model + s.model_provider = provider_context + _last_persisted_model = model + _last_persisted_provider = provider_context + _turn_owns_persisted_model = True - _agent_lock = _get_session_agent_lock(session_id) # TD1: set thread-local env context so concurrent sessions don't clobber globals # Check for pre-flight cancel (user cancelled before agent even started) if cancel_event.is_set(): @@ -6751,12 +6763,22 @@ def _run_agent_streaming( has_profile=bool(getattr(s, "profile", None)), ) # #4251: only apply the profile-repair persistence if this turn still - # owns the session model value it last wrote. - if _turn_owns_persisted_model and getattr(s, "model", None) == _last_persisted_model: - s.model_provider = provider_context - if _repaired and model != (s.model or ""): - s.model = model - _last_persisted_model = model + # owns the session model/provider pair it last wrote. + provider_context = str(provider_context).strip().lower() if provider_context else None + with _agent_lock: + _current_provider = getattr(s, "model_provider", None) + if _current_provider is not None: + _current_provider = str(_current_provider).strip().lower() or None + if ( + _turn_owns_persisted_model + and getattr(s, "model", None) == _last_persisted_model + and _current_provider == _last_persisted_provider + ): + s.model_provider = provider_context + _last_persisted_provider = provider_context + if _repaired and model != (s.model or ""): + s.model = model + _last_persisted_model = model # Capture the resolved profile name now, while profile context is # reliable. Used in the compression migration block to stamp s.profile diff --git a/tests/test_issue4251_wakeup_model_picker_race.py b/tests/test_issue4251_wakeup_model_picker_race.py index dfb1385f..f5481de8 100644 --- a/tests/test_issue4251_wakeup_model_picker_race.py +++ b/tests/test_issue4251_wakeup_model_picker_race.py @@ -173,6 +173,19 @@ def test_dispatch_stamp_does_not_clobber_newer_picker_model(monkeypatch): assert fake_session.model_provider == "openrouter" +def test_dispatch_stamp_does_not_clobber_newer_picker_provider_only_choice(monkeypatch): + fake_session = FakeSession(model="haiku-4-5", model_provider="openrouter") + + _run_streaming_turn( + monkeypatch, + fake_session, + stream_id="stream-4251-provider-race", + ) + + assert fake_session.model == "haiku-4-5" + assert fake_session.model_provider == "openrouter" + + def test_dispatch_stamp_persists_resolved_model_when_no_race(monkeypatch): fake_session = FakeSession(model="haiku-4-5", model_provider=None) @@ -216,3 +229,31 @@ def test_profile_repair_skips_persistence_when_newer_picker_choice_already_won(m assert fake_session.model == "opus-4-8" assert fake_session.model_provider == "openrouter" + + +def test_dispatch_stamp_snapshots_provider_under_agent_lock(monkeypatch): + fake_session = FakeSession(model="haiku-4-5", model_provider="anthropic") + + class PickerUpdateLock: + def __init__(self): + self._entered = 0 + + def __enter__(self): + self._entered += 1 + if self._entered == 1: + fake_session.model_provider = "openrouter" + return self + + def __exit__(self, exc_type, exc, tb): + return False + + monkeypatch.setattr(streaming, "_get_session_agent_lock", lambda _session_id: PickerUpdateLock()) + + _run_streaming_turn( + monkeypatch, + fake_session, + stream_id="stream-4251-lock-race", + ) + + assert fake_session.model == "haiku-4-5" + assert fake_session.model_provider == "openrouter" From f3c01812b73ccbb3d48e7a15943258a0f698dc4d Mon Sep 17 00:00:00 2001 From: Rod Boev Date: Wed, 1 Jul 2026 16:20:31 -0400 Subject: [PATCH 4/4] fix(#4251): drop dead post-repair ownership writes --- api/streaming.py | 2 -- 1 file changed, 2 deletions(-) diff --git a/api/streaming.py b/api/streaming.py index 161fe5da..3425f1cb 100644 --- a/api/streaming.py +++ b/api/streaming.py @@ -6775,10 +6775,8 @@ def _run_agent_streaming( and _current_provider == _last_persisted_provider ): s.model_provider = provider_context - _last_persisted_provider = provider_context if _repaired and model != (s.model or ""): s.model = model - _last_persisted_model = model # Capture the resolved profile name now, while profile context is # reliable. Used in the compression migration block to stamp s.profile