1081 lines
41 KiB
Python
1081 lines
41 KiB
Python
"""Regression coverage for #3929-D process-wakeup credential exhaustion pause."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
import queue
|
|
import sys
|
|
import types
|
|
from unittest import mock
|
|
|
|
import pytest
|
|
|
|
import api.config as config
|
|
import api.models as models
|
|
import api.profiles as profiles
|
|
import api.providers as providers
|
|
import api.routes as routes
|
|
import api.streaming as streaming
|
|
from api.models import PROCESS_WAKEUP_PAUSE_ERROR, Session
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _isolate_session_dir(tmp_path, monkeypatch):
|
|
session_dir = tmp_path / "sessions"
|
|
session_dir.mkdir()
|
|
monkeypatch.setattr(models, "SESSION_DIR", session_dir)
|
|
monkeypatch.setattr(models, "SESSION_INDEX_FILE", session_dir / "_index.json")
|
|
models.SESSIONS.clear()
|
|
yield
|
|
models.SESSIONS.clear()
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _isolate_stream_state():
|
|
config.STREAMS.clear()
|
|
config.CANCEL_FLAGS.clear()
|
|
config.AGENT_INSTANCES.clear()
|
|
config.STREAM_PARTIAL_TEXT.clear()
|
|
config.STREAM_REASONING_TEXT.clear()
|
|
config.STREAM_LIVE_TOOL_CALLS.clear()
|
|
config.STREAM_GOAL_RELATED.clear()
|
|
config.PENDING_BG_TASK_COMPLETIONS.clear()
|
|
with config.ACTIVE_RUNS_LOCK:
|
|
config.ACTIVE_RUNS.clear()
|
|
yield
|
|
config.STREAMS.clear()
|
|
config.CANCEL_FLAGS.clear()
|
|
config.AGENT_INSTANCES.clear()
|
|
config.STREAM_PARTIAL_TEXT.clear()
|
|
config.STREAM_REASONING_TEXT.clear()
|
|
config.STREAM_LIVE_TOOL_CALLS.clear()
|
|
config.STREAM_GOAL_RELATED.clear()
|
|
config.PENDING_BG_TASK_COMPLETIONS.clear()
|
|
with config.ACTIVE_RUNS_LOCK:
|
|
config.ACTIVE_RUNS.clear()
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _isolate_agent_locks():
|
|
config.SESSION_AGENT_LOCKS.clear()
|
|
yield
|
|
config.SESSION_AGENT_LOCKS.clear()
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _default_live_credential_revalidation(monkeypatch):
|
|
monkeypatch.setattr(routes, "provider_has_usable_credential", lambda *_args, **_kwargs: False)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _mock_hermes_modules(monkeypatch):
|
|
fake_runtime_module = types.ModuleType("hermes_cli.runtime_provider")
|
|
fake_runtime_module.resolve_runtime_provider = lambda requested=None, **_kw: {
|
|
"provider": requested or "test-provider",
|
|
"api_key": "synthetic-key",
|
|
"base_url": None,
|
|
}
|
|
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=None)
|
|
|
|
injected = {
|
|
"hermes_cli": fake_hermes_cli,
|
|
"hermes_cli.runtime_provider": fake_runtime_module,
|
|
"hermes_state": fake_hermes_state,
|
|
}
|
|
missing = object()
|
|
saved = {name: sys.modules.get(name, missing) for name in injected}
|
|
sys.modules.update(injected)
|
|
yield
|
|
for name, previous in saved.items():
|
|
if previous is missing:
|
|
sys.modules.pop(name, None)
|
|
else:
|
|
sys.modules[name] = previous
|
|
|
|
|
|
class _MockAgent:
|
|
def __init__(self, **kwargs):
|
|
self.session_id = kwargs.get("session_id")
|
|
self.stream_delta_callback = kwargs.get("stream_delta_callback")
|
|
self.reasoning_callback = kwargs.get("reasoning_callback")
|
|
self.tool_progress_callback = kwargs.get("tool_progress_callback")
|
|
self.session_prompt_tokens = 0
|
|
self.session_completion_tokens = 0
|
|
self.session_estimated_cost_usd = 0.0
|
|
self.context_compressor = None
|
|
self._last_error = None
|
|
self.ephemeral_system_prompt = None
|
|
|
|
def interrupt(self, _message):
|
|
pass
|
|
|
|
|
|
class _CredentialPoolEmptyAgent(_MockAgent):
|
|
def run_conversation(self, **_kwargs):
|
|
raise RuntimeError("All 0 credential(s) exhausted for test-provider")
|
|
|
|
|
|
class _StaleCredentialPoolEmptyAgent(_MockAgent):
|
|
def run_conversation(self, **_kwargs):
|
|
session = models.SESSIONS[self.session_id]
|
|
session.active_stream_id = "stream-newer-run"
|
|
session.model = "newer-model"
|
|
session.model_provider = "newer-provider"
|
|
session.pending_user_source = "webui"
|
|
session.save(touch_updated_at=False)
|
|
raise RuntimeError("All 0 credential(s) exhausted for test-provider")
|
|
|
|
|
|
class _FakeCredentialPoolEntry:
|
|
def __init__(self, payload):
|
|
self._payload = dict(payload)
|
|
for key, value in self._payload.items():
|
|
setattr(self, key, value)
|
|
self.source = self._payload.get("source", "manual")
|
|
self.label = self._payload.get("label", "")
|
|
self.key_source = self._payload.get("key_source", "")
|
|
|
|
def to_dict(self):
|
|
return dict(self._payload)
|
|
|
|
|
|
class _FakeCredentialPool:
|
|
def __init__(self, entries):
|
|
self._entries = list(entries)
|
|
|
|
def entries(self):
|
|
return list(self._entries)
|
|
|
|
|
|
def _install_fake_agent_credential_pool(monkeypatch, pool_data):
|
|
fake_agent = types.ModuleType("agent")
|
|
fake_agent.__path__ = []
|
|
fake_pool_module = types.ModuleType("agent.credential_pool")
|
|
|
|
def _load_pool(provider_id):
|
|
entries = pool_data.get(provider_id, [])
|
|
return _FakeCredentialPool(_FakeCredentialPoolEntry(entry) for entry in entries)
|
|
|
|
fake_pool_module.load_pool = _load_pool
|
|
monkeypatch.setitem(sys.modules, "agent", fake_agent)
|
|
monkeypatch.setitem(sys.modules, "agent.credential_pool", fake_pool_module)
|
|
|
|
|
|
def _run_failing_process_wakeup(session: Session, tmp_path, *, 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=("test-model", "test-provider", None)), \
|
|
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="test-model",
|
|
model_provider="test-provider",
|
|
workspace=str(tmp_path),
|
|
stream_id=stream_id,
|
|
)
|
|
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",
|
|
title="Wakeup pause",
|
|
workspace=str(tmp_path),
|
|
model="test-model",
|
|
model_provider="test-provider",
|
|
messages=[{"role": "user", "content": "Earlier prompt", "timestamp": 1}],
|
|
context_messages=[{"role": "user", "content": "Earlier prompt"}],
|
|
active_stream_id="stream-wakeup-pause-1",
|
|
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(session, tmp_path)
|
|
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 sum(1 for message in saved.messages if message.get("_error")) == 1
|
|
assert saved.process_wakeup_pause["paused"] is True
|
|
assert saved.process_wakeup_pause["classification"] == "credential_pool_empty"
|
|
assert saved.process_wakeup_pause["suppressed_count"] == 0
|
|
assert saved.process_wakeup_pause["credential_state_fingerprint"]
|
|
context_before = list(saved.context_messages)
|
|
messages_before = list(saved.messages)
|
|
|
|
def _unexpected_start_run(*_args, **_kwargs):
|
|
raise AssertionError("paused process_wakeup must not start another provider call")
|
|
|
|
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: ("test-model", "test-provider", False),
|
|
)
|
|
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.messages == messages_before
|
|
assert saved_after.context_messages == context_before
|
|
assert saved_after.process_wakeup_pause["suppressed_count"] == 1
|
|
assert "last_suppressed_at" in saved_after.process_wakeup_pause
|
|
|
|
|
|
def test_process_wakeup_pause_revalidates_when_credential_state_changes(tmp_path, monkeypatch):
|
|
hermes_home = tmp_path / "hermes-home"
|
|
hermes_home.mkdir()
|
|
auth_json = hermes_home / "auth.json"
|
|
auth_json.write_text('{"credential_pool": {}}\n', encoding="utf-8")
|
|
monkeypatch.setattr(models, "_get_profile_home", lambda _profile: hermes_home)
|
|
session = Session(
|
|
session_id="wakeup_pause_credential_refresh",
|
|
workspace=str(tmp_path),
|
|
model="test-model",
|
|
model_provider="test-provider",
|
|
)
|
|
pause = models.record_process_wakeup_provider_unavailable_pause(
|
|
session,
|
|
classification="credential_pool_empty",
|
|
model="test-model",
|
|
provider="test-provider",
|
|
)
|
|
assert pause is not None
|
|
paused_fingerprint = pause["credential_state_fingerprint"]
|
|
session.save()
|
|
models.SESSIONS[session.session_id] = session
|
|
|
|
auth_json.write_text(
|
|
'{"credential_pool": {"test-provider": [{"id": "refilled-token"}]}}\n',
|
|
encoding="utf-8",
|
|
)
|
|
assert models.process_wakeup_credential_state_fingerprint(session) != paused_fingerprint
|
|
|
|
captured = {}
|
|
|
|
def _fake_start_run(s, **kwargs):
|
|
captured["source"] = kwargs.get("source")
|
|
captured["model"] = kwargs.get("model")
|
|
captured["model_provider"] = kwargs.get("model_provider")
|
|
return {"stream_id": "stream-credential-refresh", "session_id": s.session_id, "_status": 200}
|
|
|
|
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: ("test-model", "test-provider", False),
|
|
)
|
|
monkeypatch.setattr(routes, "_start_run", _fake_start_run)
|
|
|
|
response = routes.start_session_turn(
|
|
session.session_id,
|
|
"[IMPORTANT: Background process completed after credential refill.]",
|
|
source="process_wakeup",
|
|
)
|
|
|
|
assert response["_status"] == 200
|
|
assert response["stream_id"] == "stream-credential-refresh"
|
|
assert captured == {
|
|
"source": "process_wakeup",
|
|
"model": "test-model",
|
|
"model_provider": "test-provider",
|
|
}
|
|
saved = Session.load(session.session_id)
|
|
assert saved is not None
|
|
assert saved.process_wakeup_pause == {}
|
|
|
|
|
|
def test_process_wakeup_pause_survives_rotation_style_auth_rewrite(tmp_path, monkeypatch):
|
|
hermes_home = tmp_path / "hermes-home"
|
|
hermes_home.mkdir()
|
|
auth_json = hermes_home / "auth.json"
|
|
auth_json.write_text(
|
|
json.dumps(
|
|
{
|
|
"credential_pool": {
|
|
"other-provider": [
|
|
{
|
|
"id": "other-token",
|
|
"source": "oauth",
|
|
"auth_type": "oauth",
|
|
"access_token": "old-access",
|
|
"refresh_token": "old-refresh",
|
|
"expires_at": 1000,
|
|
"last_status": "ok",
|
|
"request_count": 1,
|
|
"updated_at": "old",
|
|
}
|
|
]
|
|
}
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setattr(models, "_get_profile_home", lambda _profile: hermes_home)
|
|
session = Session(
|
|
session_id="wakeup_pause_auth_rotation",
|
|
workspace=str(tmp_path),
|
|
model="test-model",
|
|
model_provider="test-provider",
|
|
)
|
|
pause = models.record_process_wakeup_provider_unavailable_pause(
|
|
session,
|
|
classification="credential_pool_empty",
|
|
model="test-model",
|
|
provider="test-provider",
|
|
)
|
|
assert pause is not None
|
|
paused_fingerprint = pause["credential_state_fingerprint"]
|
|
session.save()
|
|
models.SESSIONS[session.session_id] = session
|
|
|
|
auth_json.write_text(
|
|
json.dumps(
|
|
{
|
|
"credential_pool": {
|
|
"other-provider": [
|
|
{
|
|
"id": "other-token",
|
|
"source": "oauth",
|
|
"auth_type": "oauth",
|
|
"access_token": "new-access",
|
|
"refresh_token": "new-refresh",
|
|
"expires_at": 2000,
|
|
"last_status": "ok",
|
|
"request_count": 2,
|
|
"updated_at": "new",
|
|
}
|
|
]
|
|
}
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
assert models.process_wakeup_credential_state_fingerprint(session) == paused_fingerprint
|
|
|
|
def _unexpected_start_run(*_args, **_kwargs):
|
|
raise AssertionError("token rotation must not clear a paused wakeup")
|
|
|
|
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: ("test-model", "test-provider", False),
|
|
)
|
|
monkeypatch.setattr(routes, "_start_run", _unexpected_start_run)
|
|
|
|
response = routes.start_session_turn(
|
|
session.session_id,
|
|
"[IMPORTANT: Background process completed during token rotation.]",
|
|
source="process_wakeup",
|
|
)
|
|
|
|
assert response["_status"] == 409
|
|
assert response["error"] == PROCESS_WAKEUP_PAUSE_ERROR
|
|
saved = Session.load(session.session_id)
|
|
assert saved is not None
|
|
assert saved.process_wakeup_pause["suppressed_count"] == 1
|
|
assert saved.process_wakeup_pause["credential_state_fingerprint"] == paused_fingerprint
|
|
|
|
|
|
def test_process_wakeup_pause_revalidates_status_recovery_without_fingerprint_change(tmp_path, monkeypatch):
|
|
hermes_home = tmp_path / "hermes-home"
|
|
hermes_home.mkdir()
|
|
auth_json = hermes_home / "auth.json"
|
|
exhausted_entry = {
|
|
"id": "pooled-token",
|
|
"label": "Test provider credential",
|
|
"source": "manual",
|
|
"auth_type": "api_key",
|
|
"runtime_api_key": "sk-test-provider",
|
|
"last_status": "exhausted",
|
|
"last_status_at": "2026-07-07T00:00:00Z",
|
|
"last_error_code": "429",
|
|
"last_error_reset_at": "2999-01-01T00:00:00Z",
|
|
}
|
|
pool_data = {"test-provider": [dict(exhausted_entry)]}
|
|
auth_json.write_text(json.dumps({"credential_pool": pool_data}), encoding="utf-8")
|
|
_install_fake_agent_credential_pool(monkeypatch, pool_data)
|
|
monkeypatch.setattr(models, "_get_profile_home", lambda _profile: hermes_home)
|
|
monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: hermes_home)
|
|
monkeypatch.setattr(routes, "provider_has_usable_credential", providers.provider_has_usable_credential)
|
|
session = Session(
|
|
session_id="wakeup_pause_status_recovery",
|
|
workspace=str(tmp_path),
|
|
model="test-model",
|
|
model_provider="test-provider",
|
|
)
|
|
pause = models.record_process_wakeup_provider_unavailable_pause(
|
|
session,
|
|
classification="credential_pool_empty",
|
|
model="test-model",
|
|
provider="test-provider",
|
|
)
|
|
assert pause is not None
|
|
paused_fingerprint = pause["credential_state_fingerprint"]
|
|
session.save()
|
|
models.SESSIONS[session.session_id] = session
|
|
|
|
recovered_entry = dict(exhausted_entry)
|
|
recovered_entry["last_status"] = "ok"
|
|
recovered_entry["last_status_at"] = "2026-07-07T00:01:00Z"
|
|
recovered_entry["last_error_reset_at"] = None
|
|
pool_data["test-provider"] = [recovered_entry]
|
|
auth_json.write_text(json.dumps({"credential_pool": pool_data}), encoding="utf-8")
|
|
assert models.process_wakeup_credential_state_fingerprint(session) == paused_fingerprint
|
|
|
|
captured = {}
|
|
|
|
def _fake_start_run(s, **kwargs):
|
|
captured["source"] = kwargs.get("source")
|
|
captured["model"] = kwargs.get("model")
|
|
captured["model_provider"] = kwargs.get("model_provider")
|
|
return {"stream_id": "stream-status-recovered", "session_id": s.session_id, "_status": 200}
|
|
|
|
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: ("test-model", "test-provider", False),
|
|
)
|
|
monkeypatch.setattr(routes, "_start_run", _fake_start_run)
|
|
|
|
response = routes.start_session_turn(
|
|
session.session_id,
|
|
"[IMPORTANT: Background process completed after status recovery.]",
|
|
source="process_wakeup",
|
|
)
|
|
|
|
assert response["_status"] == 200
|
|
assert response["stream_id"] == "stream-status-recovered"
|
|
assert captured == {
|
|
"source": "process_wakeup",
|
|
"model": "test-model",
|
|
"model_provider": "test-provider",
|
|
}
|
|
saved = Session.load(session.session_id)
|
|
assert saved is not None
|
|
assert saved.process_wakeup_pause == {}
|
|
|
|
|
|
def test_process_wakeup_pause_suppresses_at_provider_model_session(tmp_path, monkeypatch):
|
|
session = Session(
|
|
session_id="wakeup_pause_at_model",
|
|
workspace=str(tmp_path),
|
|
model="@test-provider:test-model",
|
|
model_provider=None,
|
|
)
|
|
pause = models.record_process_wakeup_provider_unavailable_pause(
|
|
session,
|
|
classification="credential_pool_empty",
|
|
model="test-model",
|
|
provider="test-provider",
|
|
)
|
|
assert pause is not None
|
|
session.save()
|
|
models.SESSIONS[session.session_id] = session
|
|
|
|
def _unexpected_start_run(*_args, **_kwargs):
|
|
raise AssertionError("@provider:model wakeup must be suppressed on the paused lane")
|
|
|
|
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: ("@test-provider:test-model", None, False),
|
|
)
|
|
monkeypatch.setattr(routes, "_start_run", _unexpected_start_run)
|
|
|
|
response = routes.start_session_turn(
|
|
session.session_id,
|
|
"[IMPORTANT: Background process completed for @provider:model lane.]",
|
|
source="process_wakeup",
|
|
)
|
|
|
|
assert response["_status"] == 409
|
|
assert response["error"] == PROCESS_WAKEUP_PAUSE_ERROR
|
|
saved = Session.load(session.session_id)
|
|
assert saved is not None
|
|
assert saved.process_wakeup_pause["model"] == "test-model"
|
|
assert saved.process_wakeup_pause["provider"] == "test-provider"
|
|
assert saved.process_wakeup_pause["suppressed_count"] == 1
|
|
|
|
|
|
def test_process_wakeup_pause_suppresses_openrouter_tagged_model_session(tmp_path, monkeypatch):
|
|
monkeypatch.setitem(
|
|
config.cfg,
|
|
"model",
|
|
{"provider": "anthropic", "default": "claude-sonnet-4.6"},
|
|
)
|
|
assert models._process_wakeup_pause_key(
|
|
"deepseek/deepseek-r1:free",
|
|
"openrouter",
|
|
"credential_pool_empty",
|
|
) == models._process_wakeup_pause_key(
|
|
"deepseek/deepseek-r1:free",
|
|
None,
|
|
"credential_pool_empty",
|
|
)
|
|
session = Session(
|
|
session_id="wakeup_pause_openrouter_tagged_model",
|
|
workspace=str(tmp_path),
|
|
model="deepseek/deepseek-r1:free",
|
|
model_provider=None,
|
|
)
|
|
pause = models.record_process_wakeup_provider_unavailable_pause(
|
|
session,
|
|
classification="credential_pool_empty",
|
|
model="deepseek/deepseek-r1:free",
|
|
provider="openrouter",
|
|
)
|
|
assert pause is not None
|
|
session.save()
|
|
models.SESSIONS[session.session_id] = session
|
|
|
|
def _unexpected_start_run(*_args, **_kwargs):
|
|
raise AssertionError("OpenRouter tagged wakeup must be suppressed on the paused lane")
|
|
|
|
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: ("deepseek/deepseek-r1:free", None, False),
|
|
)
|
|
monkeypatch.setattr(routes, "_start_run", _unexpected_start_run)
|
|
|
|
response = routes.start_session_turn(
|
|
session.session_id,
|
|
"[IMPORTANT: Background process completed for OpenRouter tagged lane.]",
|
|
source="process_wakeup",
|
|
)
|
|
|
|
assert response["_status"] == 409
|
|
assert response["error"] == PROCESS_WAKEUP_PAUSE_ERROR
|
|
saved = Session.load(session.session_id)
|
|
assert saved is not None
|
|
assert saved.process_wakeup_pause["model"] == "deepseek/deepseek-r1:free"
|
|
assert saved.process_wakeup_pause["provider"] == "openrouter"
|
|
assert saved.process_wakeup_pause["suppressed_count"] == 1
|
|
|
|
|
|
def test_process_wakeup_pause_suppresses_local_endpoint_custom_slug_session(tmp_path, monkeypatch):
|
|
monkeypatch.setitem(
|
|
config.cfg,
|
|
"model",
|
|
{
|
|
"provider": "ollama",
|
|
"default": "llama3.2",
|
|
"base_url": "http://ollama.internal:11434/v1",
|
|
},
|
|
)
|
|
endpoint_lane = models._process_wakeup_pause_key(
|
|
"@custom:ollama.internal:11434:llama3.2",
|
|
None,
|
|
"credential_pool_empty",
|
|
)
|
|
assert models._process_wakeup_pause_key(
|
|
"llama3.2",
|
|
"ollama",
|
|
"credential_pool_empty",
|
|
) == endpoint_lane
|
|
session = Session(
|
|
session_id="wakeup_pause_local_endpoint_slug",
|
|
workspace=str(tmp_path),
|
|
model="@custom:ollama.internal:11434:llama3.2",
|
|
model_provider=None,
|
|
)
|
|
pause = models.record_process_wakeup_provider_unavailable_pause(
|
|
session,
|
|
classification="credential_pool_empty",
|
|
model="llama3.2",
|
|
provider="ollama",
|
|
)
|
|
assert pause is not None
|
|
session.save()
|
|
models.SESSIONS[session.session_id] = session
|
|
|
|
def _unexpected_start_run(*_args, **_kwargs):
|
|
raise AssertionError("local-endpoint wakeup must be suppressed on the resolved provider lane")
|
|
|
|
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: ("@custom:ollama.internal:11434:llama3.2", None, False),
|
|
)
|
|
monkeypatch.setattr(routes, "_start_run", _unexpected_start_run)
|
|
|
|
response = routes.start_session_turn(
|
|
session.session_id,
|
|
"[IMPORTANT: Background process completed for local endpoint lane.]",
|
|
source="process_wakeup",
|
|
)
|
|
|
|
assert response["_status"] == 409
|
|
assert response["error"] == PROCESS_WAKEUP_PAUSE_ERROR
|
|
saved = Session.load(session.session_id)
|
|
assert saved is not None
|
|
assert saved.process_wakeup_pause["model"] == "llama3.2"
|
|
assert saved.process_wakeup_pause["provider"] == endpoint_lane["provider"]
|
|
assert saved.process_wakeup_pause["suppressed_count"] == 1
|
|
|
|
|
|
def test_process_wakeup_pause_suppresses_custom_provider_tagged_model_session(tmp_path, monkeypatch):
|
|
session = Session(
|
|
session_id="wakeup_pause_custom_tagged_model",
|
|
workspace=str(tmp_path),
|
|
model="@custom:proxy:model:tag",
|
|
model_provider=None,
|
|
)
|
|
pause = models.record_process_wakeup_provider_unavailable_pause(
|
|
session,
|
|
classification="credential_pool_empty",
|
|
model="model:tag",
|
|
provider="custom:proxy",
|
|
)
|
|
assert pause is not None
|
|
session.save()
|
|
models.SESSIONS[session.session_id] = session
|
|
|
|
def _unexpected_start_run(*_args, **_kwargs):
|
|
raise AssertionError("custom-provider tagged wakeup must be suppressed on the paused lane")
|
|
|
|
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: ("@custom:proxy:model:tag", None, False),
|
|
)
|
|
monkeypatch.setattr(routes, "_start_run", _unexpected_start_run)
|
|
|
|
response = routes.start_session_turn(
|
|
session.session_id,
|
|
"[IMPORTANT: Background process completed for custom tagged lane.]",
|
|
source="process_wakeup",
|
|
)
|
|
|
|
assert response["_status"] == 409
|
|
assert response["error"] == PROCESS_WAKEUP_PAUSE_ERROR
|
|
saved = Session.load(session.session_id)
|
|
assert saved is not None
|
|
assert saved.process_wakeup_pause["model"] == "model:tag"
|
|
assert saved.process_wakeup_pause["provider"] == "custom:proxy"
|
|
assert saved.process_wakeup_pause["suppressed_count"] == 1
|
|
|
|
|
|
def test_process_wakeup_pause_suppresses_custom_provider_single_segment_session(tmp_path, monkeypatch):
|
|
assert models._process_wakeup_pause_key(
|
|
"tag",
|
|
"custom:proxy",
|
|
"credential_pool_empty",
|
|
) == models._process_wakeup_pause_key(
|
|
"@custom:proxy:tag",
|
|
None,
|
|
"credential_pool_empty",
|
|
)
|
|
session = Session(
|
|
session_id="wakeup_pause_custom_single_segment",
|
|
workspace=str(tmp_path),
|
|
model="@custom:proxy:tag",
|
|
model_provider=None,
|
|
)
|
|
pause = models.record_process_wakeup_provider_unavailable_pause(
|
|
session,
|
|
classification="credential_pool_empty",
|
|
model="tag",
|
|
provider="custom:proxy",
|
|
)
|
|
assert pause is not None
|
|
session.save()
|
|
models.SESSIONS[session.session_id] = session
|
|
|
|
def _unexpected_start_run(*_args, **_kwargs):
|
|
raise AssertionError("custom-provider shorthand wakeup must be suppressed on the paused lane")
|
|
|
|
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: ("@custom:proxy:tag", None, False),
|
|
)
|
|
monkeypatch.setattr(routes, "_start_run", _unexpected_start_run)
|
|
|
|
response = routes.start_session_turn(
|
|
session.session_id,
|
|
"[IMPORTANT: Background process completed for custom shorthand lane.]",
|
|
source="process_wakeup",
|
|
)
|
|
|
|
assert response["_status"] == 409
|
|
assert response["error"] == PROCESS_WAKEUP_PAUSE_ERROR
|
|
saved = Session.load(session.session_id)
|
|
assert saved is not None
|
|
assert saved.process_wakeup_pause["model"] == "tag"
|
|
assert saved.process_wakeup_pause["provider"] == "custom:proxy"
|
|
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",
|
|
title="Wakeup pause stale",
|
|
workspace=str(tmp_path),
|
|
model="test-model",
|
|
model_provider="test-provider",
|
|
messages=[{"role": "user", "content": "Earlier prompt", "timestamp": 1}],
|
|
context_messages=[{"role": "user", "content": "Earlier prompt"}],
|
|
active_stream_id="stream-wakeup-pause-stale",
|
|
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
|
|
stream_id = str(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=_StaleCredentialPoolEmptyAgent), \
|
|
mock.patch.object(streaming, "resolve_model_provider", return_value=("test-model", "test-provider", None)), \
|
|
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="test-model",
|
|
model_provider="test-provider",
|
|
workspace=str(tmp_path),
|
|
stream_id=stream_id,
|
|
)
|
|
|
|
saved = Session.load(session.session_id)
|
|
assert saved is not None
|
|
assert saved.active_stream_id == "stream-newer-run"
|
|
assert saved.model == "newer-model"
|
|
assert saved.model_provider == "newer-provider"
|
|
assert saved.pending_user_source == "webui"
|
|
assert saved.process_wakeup_pause["paused"] is True
|
|
assert saved.process_wakeup_pause["classification"] == "credential_pool_empty"
|
|
assert saved.process_wakeup_pause["model"] == "test-model"
|
|
assert saved.process_wakeup_pause["provider"] == "test-provider"
|
|
assert saved.process_wakeup_pause["suppressed_count"] == 0
|
|
assert not any(message.get("_error") for message in saved.messages)
|
|
|
|
|
|
def test_process_wakeup_pause_resets_when_model_provider_lane_changes(tmp_path, monkeypatch):
|
|
session = Session(
|
|
session_id="wakeup_pause_reset",
|
|
workspace=str(tmp_path),
|
|
model="old-model",
|
|
model_provider="old-provider",
|
|
process_wakeup_pause={
|
|
"version": 1,
|
|
"paused": True,
|
|
"source": "process_wakeup",
|
|
"classification": "credential_pool_empty",
|
|
"model": "old-model",
|
|
"provider": "old-provider",
|
|
"first_paused_at": 1.0,
|
|
"last_error_at": 1.0,
|
|
"visible_error_count": 1,
|
|
"suppressed_count": 2,
|
|
},
|
|
)
|
|
session.save()
|
|
models.SESSIONS[session.session_id] = session
|
|
|
|
captured = {}
|
|
|
|
def _fake_start_run(s, **kwargs):
|
|
captured["model"] = kwargs.get("model")
|
|
captured["model_provider"] = kwargs.get("model_provider")
|
|
return {"stream_id": "stream-reset", "session_id": s.session_id, "_status": 200}
|
|
|
|
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: ("new-model", "new-provider", True),
|
|
)
|
|
monkeypatch.setattr(routes, "_start_run", _fake_start_run)
|
|
|
|
response = routes.start_session_turn(
|
|
session.session_id,
|
|
"[IMPORTANT: Background process completed after provider change.]",
|
|
source="process_wakeup",
|
|
)
|
|
|
|
assert response["_status"] == 200
|
|
assert response["stream_id"] == "stream-reset"
|
|
assert captured == {"model": "new-model", "model_provider": "new-provider"}
|
|
saved = Session.load(session.session_id)
|
|
assert saved is not None
|
|
assert saved.process_wakeup_pause == {}
|
|
|
|
|
|
def test_success_path_clears_process_wakeup_pause_after_late_cancel_checks():
|
|
src = Path(__file__).parent.parent.joinpath("api", "streaming.py").read_text(encoding="utf-8")
|
|
session_save_idx = src.index('with _stream_writeback_stage(_writeback_timings, "session_save")')
|
|
post_save_cancel_idx = src.index("if cancel_event.is_set():", session_save_idx)
|
|
state_sync_idx = src.index('with _stream_writeback_stage(_writeback_timings, "state_sync")')
|
|
final_cancel_idx = src.index("if cancel_event.is_set():", state_sync_idx)
|
|
pause_clear_idx = src.index("clear_process_wakeup_pause(s, reason='run_completed')")
|
|
done_payload_idx = src.index('with _stream_writeback_stage(_writeback_timings, "done_payload")')
|
|
|
|
assert session_save_idx < post_save_cancel_idx < state_sync_idx
|
|
assert state_sync_idx < final_cancel_idx < pause_clear_idx < done_payload_idx
|
|
|
|
|
|
def test_gateway_success_path_checks_cancel_before_clearing_process_wakeup_pause():
|
|
src = Path(__file__).parent.parent.joinpath("api", "gateway_chat.py").read_text(encoding="utf-8")
|
|
current_idx = src.index("if not _stream_writeback_is_current(s, stream_id):")
|
|
early_cancel_idx = src.index("if cancel_event.is_set():", current_idx)
|
|
pending_clear_idx = src.index("s.pending_user_source = None")
|
|
pause_clear_idx = src.index('clear_process_wakeup_pause(s, reason="run_completed")')
|
|
final_cancel_idx = src.rindex("if cancel_event.is_set():", current_idx, pause_clear_idx)
|
|
save_idx = src.index("s.save()", pause_clear_idx)
|
|
|
|
assert current_idx < early_cancel_idx < pending_clear_idx < final_cancel_idx < pause_clear_idx < save_idx
|
|
|
|
|
|
def test_process_wakeup_pause_does_not_suppress_explicit_non_wakeup_turn(tmp_path, monkeypatch):
|
|
session = Session(
|
|
session_id="wakeup_pause_manual_recover",
|
|
workspace=str(tmp_path),
|
|
model="test-model",
|
|
model_provider="test-provider",
|
|
process_wakeup_pause={
|
|
"version": 1,
|
|
"paused": True,
|
|
"source": "process_wakeup",
|
|
"classification": "credential_pool_empty",
|
|
"model": "test-model",
|
|
"provider": "test-provider",
|
|
"first_paused_at": 1.0,
|
|
"last_error_at": 1.0,
|
|
"visible_error_count": 1,
|
|
"suppressed_count": 2,
|
|
},
|
|
)
|
|
session.save()
|
|
models.SESSIONS[session.session_id] = session
|
|
|
|
captured = {}
|
|
|
|
def _fake_start_run(s, **kwargs):
|
|
captured["source"] = kwargs.get("source")
|
|
captured["message"] = kwargs.get("msg")
|
|
return {"stream_id": "stream-manual-recover", "session_id": s.session_id, "_status": 200}
|
|
|
|
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: ("test-model", "test-provider", False),
|
|
)
|
|
monkeypatch.setattr(routes, "_start_run", _fake_start_run)
|
|
|
|
response = routes.start_session_turn(
|
|
session.session_id,
|
|
"Explicit recovery attempt",
|
|
source="manual_recover",
|
|
)
|
|
|
|
assert response["_status"] == 200
|
|
assert response["stream_id"] == "stream-manual-recover"
|
|
assert captured == {"source": "manual_recover", "message": "Explicit recovery attempt"}
|
|
saved = Session.load(session.session_id)
|
|
assert saved is not None
|
|
assert saved.process_wakeup_pause["suppressed_count"] == 2
|
|
assert "last_suppressed_at" not in saved.process_wakeup_pause
|