diff --git a/api/models.py b/api/models.py index bfb23cdc..8859eab2 100644 --- a/api/models.py +++ b/api/models.py @@ -1574,6 +1574,12 @@ def _process_wakeup_pause_lane(model=None, provider=None) -> tuple[str, str]: try: resolved_model, resolved_provider = _cfg.canonical_model_provider_lane(model, provider) except Exception: + logger.debug( + "failed to canonicalize process_wakeup pause lane for model=%r provider=%r", + model, + provider, + exc_info=True, + ) resolved_model, resolved_provider = None, None if resolved_model: model_part = _process_wakeup_pause_part(resolved_model) diff --git a/api/streaming.py b/api/streaming.py index 4f1993f0..edd233b4 100644 --- a/api/streaming.py +++ b/api/streaming.py @@ -6393,6 +6393,8 @@ def _run_agent_streaming( When ephemeral=True, session mutations are skipped — used by /btw to get a streaming answer without persisting to the parent session. """ + _turn_route_model = model + _turn_route_provider = model_provider q = STREAMS.get(stream_id) if q is None: # The stream was cancelled before the worker started; the route layer @@ -8733,13 +8735,12 @@ def _run_agent_streaming( _err_type, _err_hint, ) - _pending_source = getattr(s, 'pending_user_source', None) or 'webui' - if _pending_source == 'process_wakeup': + if _turn_pending_source == 'process_wakeup': record_process_wakeup_provider_unavailable_pause( s, classification=_err_type, - model=resolved_model or model, - provider=resolved_provider or provider_context or model_provider, + model=_turn_route_model, + provider=_turn_route_provider, ) _materialize_pending_user_turn_before_error(s) s.active_stream_id = None @@ -9839,8 +9840,8 @@ def _run_agent_streaming( _pause = record_process_wakeup_provider_unavailable_pause( s, classification=_exc_type, - model=resolved_model or model, - provider=resolved_provider or provider_context or model_provider, + model=_turn_route_model, + provider=_turn_route_provider, ) if _pause is not None: try: @@ -9859,13 +9860,12 @@ def _run_agent_streaming( ) return - _pending_source = getattr(s, 'pending_user_source', None) or 'webui' - if _pending_source == 'process_wakeup': + if _turn_pending_source == 'process_wakeup': record_process_wakeup_provider_unavailable_pause( s, classification=_exc_type, - model=resolved_model or model, - provider=resolved_provider or provider_context or model_provider, + model=_turn_route_model, + provider=_turn_route_provider, ) _materialize_pending_user_turn_before_error(s) s.active_stream_id = None diff --git a/tests/test_issue3929_process_wakeup_pause.py b/tests/test_issue3929_process_wakeup_pause.py index 7e1bdd92..560c7de5 100644 --- a/tests/test_issue3929_process_wakeup_pause.py +++ b/tests/test_issue3929_process_wakeup_pause.py @@ -143,6 +143,56 @@ def _run_failing_process_wakeup(session: Session, tmp_path, *, stream_id=None): return [(item[0], item[1]) for item in list(fake_queue.queue)] +def _run_failing_process_wakeup_route( + session: Session, + tmp_path, + *, + route_model, + route_provider=None, + resolved_model, + resolved_provider, + resolved_base_url=None, + custom_connection=(None, None), + stream_id=None, +): + stream_id = str(stream_id or session.active_stream_id) + fake_queue = queue.Queue() + streaming.STREAMS[stream_id] = fake_queue + config.STREAM_PARTIAL_TEXT[stream_id] = "" + + with mock.patch.object(streaming, "_get_ai_agent", return_value=_CredentialPoolEmptyAgent), \ + mock.patch.object( + streaming, + "resolve_model_provider", + return_value=(resolved_model, resolved_provider, resolved_base_url), + ), \ + mock.patch.object( + streaming, + "resolve_custom_provider_connection", + return_value=custom_connection, + ), \ + mock.patch("api.config._resolve_cli_toolsets", return_value=[]): + streaming._run_agent_streaming( + session_id=session.session_id, + msg_text=session.pending_user_message, + model=route_model, + model_provider=route_provider, + workspace=str(tmp_path), + stream_id=stream_id, + ) + return [(item[0], item[1]) for item in list(fake_queue.queue)] + + +def _patch_process_wakeup_route(monkeypatch, tmp_path, *, model, provider): + monkeypatch.setattr(routes, "_resolve_chat_workspace_with_recovery", lambda _s, _w: str(tmp_path)) + monkeypatch.setattr(routes, "_read_profile_model_config", lambda _s, _p: (None, None, {})) + monkeypatch.setattr( + routes, + "_resolve_compatible_session_model_state", + lambda *_args, **_kwargs: (model, provider, False), + ) + + def test_credential_empty_process_wakeup_pauses_repeated_automatic_turns(tmp_path, monkeypatch): session = Session( session_id="wakeup_pause", @@ -620,6 +670,116 @@ def test_process_wakeup_pause_suppresses_custom_provider_single_segment_session( assert saved.process_wakeup_pause["suppressed_count"] == 1 +def test_process_wakeup_pause_records_route_lane_after_custom_runtime_rewrite(tmp_path, monkeypatch): + session = Session( + session_id="wakeup_pause_custom_runtime_rewrite", + workspace=str(tmp_path), + model="@custom:proxy:model:tag", + model_provider=None, + messages=[{"role": "user", "content": "Earlier prompt", "timestamp": 1}], + context_messages=[{"role": "user", "content": "Earlier prompt"}], + active_stream_id="stream-custom-runtime-rewrite", + pending_user_message="[IMPORTANT: Background process first completed.]", + pending_started_at=1234.0, + pending_user_source="process_wakeup", + ) + session.save() + models.SESSIONS[session.session_id] = session + + events = _run_failing_process_wakeup_route( + session, + tmp_path, + route_model="@custom:proxy:model:tag", + route_provider=None, + resolved_model="model:tag", + resolved_provider="custom:proxy", + custom_connection=(None, "http://proxy.example/v1"), + ) + saved = Session.load(session.session_id) + assert saved is not None + assert any(event == "apperror" and data["type"] == "credential_pool_empty" for event, data in events) + assert saved.process_wakeup_pause["model"] == "model:tag" + assert saved.process_wakeup_pause["provider"] == "custom:proxy" + + def _unexpected_start_run(*_args, **_kwargs): + raise AssertionError("post-rewrite custom-provider wakeup must be suppressed") + + _patch_process_wakeup_route( + monkeypatch, + tmp_path, + model="@custom:proxy:model:tag", + provider=None, + ) + monkeypatch.setattr(routes, "_start_run", _unexpected_start_run) + + response = routes.start_session_turn( + session.session_id, + "[IMPORTANT: Background process second completed.]", + source="process_wakeup", + ) + + assert response["_status"] == 409 + assert response["error"] == PROCESS_WAKEUP_PAUSE_ERROR + saved_after = Session.load(session.session_id) + assert saved_after is not None + assert saved_after.process_wakeup_pause["suppressed_count"] == 1 + + +def test_process_wakeup_pause_records_unset_route_provider_before_runtime_backfill(tmp_path, monkeypatch): + monkeypatch.setitem(config.cfg, "model", {"default": "claude-sonnet-test"}) + session = Session( + session_id="wakeup_pause_unset_route_provider", + workspace=str(tmp_path), + model="claude-sonnet-test", + model_provider=None, + messages=[{"role": "user", "content": "Earlier prompt", "timestamp": 1}], + context_messages=[{"role": "user", "content": "Earlier prompt"}], + active_stream_id="stream-unset-route-provider", + pending_user_message="[IMPORTANT: Background process first completed.]", + pending_started_at=1234.0, + pending_user_source="process_wakeup", + ) + session.save() + models.SESSIONS[session.session_id] = session + + events = _run_failing_process_wakeup_route( + session, + tmp_path, + route_model="claude-sonnet-test", + route_provider=None, + resolved_model="claude-sonnet-test", + resolved_provider=None, + ) + saved = Session.load(session.session_id) + assert saved is not None + assert any(event == "apperror" and data["type"] == "credential_pool_empty" for event, data in events) + assert saved.process_wakeup_pause["model"] == "claude-sonnet-test" + assert saved.process_wakeup_pause["provider"] == "" + + def _unexpected_start_run(*_args, **_kwargs): + raise AssertionError("runtime-provider backfill must not clear the paused route lane") + + _patch_process_wakeup_route( + monkeypatch, + tmp_path, + model="claude-sonnet-test", + provider=None, + ) + monkeypatch.setattr(routes, "_start_run", _unexpected_start_run) + + response = routes.start_session_turn( + session.session_id, + "[IMPORTANT: Background process second completed.]", + source="process_wakeup", + ) + + assert response["_status"] == 409 + assert response["error"] == PROCESS_WAKEUP_PAUSE_ERROR + saved_after = Session.load(session.session_id) + assert saved_after is not None + assert saved_after.process_wakeup_pause["suppressed_count"] == 1 + + def test_stale_credential_empty_process_wakeup_still_records_pause(tmp_path): session = Session( session_id="wakeup_pause_stale",