2654 lines
99 KiB
Python
2654 lines
99 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 hashlib
|
|
import sys
|
|
import threading
|
|
import types
|
|
from unittest import mock
|
|
|
|
import pytest
|
|
|
|
import api.config as config
|
|
import api.gateway_chat as gateway_chat
|
|
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_process_wakeup_recovery_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 _SuccessfulAgent(_MockAgent):
|
|
def run_conversation(self, **kwargs):
|
|
history = list(kwargs.get("conversation_history") or [])
|
|
return {
|
|
"messages": history
|
|
+ [
|
|
{"role": "user", "content": kwargs.get("persist_user_message", "")},
|
|
{"role": "assistant", "content": "Stream reply"},
|
|
]
|
|
}
|
|
|
|
|
|
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_cancelled_stale_process_wakeup_credential_failure_records_pause(tmp_path, monkeypatch):
|
|
stream_id = "stream-cancelled-stale-wakeup"
|
|
session = Session(
|
|
session_id="cancelled_stale_wakeup_pause",
|
|
title="Cancelled stale 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_id,
|
|
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
|
|
fake_queue = queue.Queue()
|
|
streaming.STREAMS[stream_id] = fake_queue
|
|
config.STREAM_PARTIAL_TEXT[stream_id] = ""
|
|
|
|
class _CancelledStaleCredentialPoolEmptyAgent(_MockAgent):
|
|
def run_conversation(self, **_kwargs):
|
|
cancel_flag = config.CANCEL_FLAGS.get(stream_id)
|
|
assert cancel_flag is not None
|
|
cancel_flag.set()
|
|
stale_session = models.SESSIONS[self.session_id]
|
|
stale_session.active_stream_id = "stream-newer-run"
|
|
stale_session.save(touch_updated_at=False)
|
|
raise RuntimeError("All 0 credential(s) exhausted for test-provider")
|
|
|
|
with mock.patch.object(streaming, "_get_ai_agent", return_value=_CancelledStaleCredentialPoolEmptyAgent), \
|
|
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,
|
|
)
|
|
|
|
events = [(item[0], item[1]) for item in list(fake_queue.queue)]
|
|
assert any(event == "cancel" for event, _data in events)
|
|
assert not any(event == "apperror" and data["type"] == "credential_pool_empty" for event, data in events)
|
|
saved = Session.load(session.session_id)
|
|
assert saved is not None
|
|
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
|
|
|
|
def _unexpected_start_run(*_args, **_kwargs):
|
|
raise AssertionError("cancelled stale process_wakeup pause must suppress the next automatic wakeup")
|
|
|
|
_patch_process_wakeup_route(
|
|
monkeypatch,
|
|
tmp_path,
|
|
model="test-model",
|
|
provider="test-provider",
|
|
)
|
|
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_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)
|
|
monkeypatch.setattr(
|
|
routes,
|
|
"provider_has_process_wakeup_recovery_credential",
|
|
lambda provider_id, *, refresh=False: provider_id == "test-provider",
|
|
)
|
|
|
|
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_keeps_changed_credential_state_until_provider_is_usable(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_changed_still_unusable",
|
|
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",
|
|
)
|
|
changed_fingerprint = models.process_wakeup_credential_state_fingerprint(session)
|
|
assert changed_fingerprint != paused_fingerprint
|
|
|
|
def _unexpected_start_run(*_args, **_kwargs):
|
|
raise AssertionError("changed credential metadata must not clear until the provider is usable")
|
|
|
|
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)
|
|
monkeypatch.setattr(
|
|
routes,
|
|
"provider_has_process_wakeup_recovery_credential",
|
|
lambda *_args, **_kwargs: False,
|
|
)
|
|
|
|
response = routes.start_session_turn(
|
|
session.session_id,
|
|
"[IMPORTANT: Background process completed after unusable credential metadata changed.]",
|
|
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"] == changed_fingerprint
|
|
|
|
|
|
def test_process_wakeup_pause_keeps_pause_when_config_key_matches_exhausted_pool_secret(tmp_path, monkeypatch):
|
|
session = Session(
|
|
session_id="wakeup_pause_config_key_matches_exhausted_pool",
|
|
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
|
|
session.save()
|
|
models.SESSIONS[session.session_id] = session
|
|
exhausted_fingerprint = hashlib.sha256(b"same-key").hexdigest()[:16]
|
|
monkeypatch.setattr(providers, "_get_provider_api_key", lambda _provider: "same-key")
|
|
monkeypatch.setattr(
|
|
providers,
|
|
"_pool_entry_payloads",
|
|
lambda _provider: [{
|
|
"last_status": "dead",
|
|
"secret_fingerprint": exhausted_fingerprint,
|
|
}],
|
|
)
|
|
monkeypatch.setattr(
|
|
routes,
|
|
"provider_has_process_wakeup_recovery_credential",
|
|
providers.provider_has_process_wakeup_recovery_credential,
|
|
)
|
|
|
|
def _unexpected_start_run(*_args, **_kwargs):
|
|
raise AssertionError("same exhausted pool secret must not prove recovery")
|
|
|
|
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 with same exhausted config key.]",
|
|
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
|
|
|
|
|
|
def test_process_wakeup_pause_keeps_pause_when_config_key_has_no_pool_evidence(tmp_path, monkeypatch):
|
|
session = Session(
|
|
session_id="wakeup_pause_config_key_no_pool_evidence",
|
|
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
|
|
session.save()
|
|
models.SESSIONS[session.session_id] = session
|
|
monkeypatch.setattr(providers, "_get_provider_api_key", lambda _provider: "generic-config-key")
|
|
monkeypatch.setattr(providers, "_pool_entry_payloads", lambda _provider: [])
|
|
monkeypatch.setattr(
|
|
routes,
|
|
"provider_has_process_wakeup_recovery_credential",
|
|
providers.provider_has_process_wakeup_recovery_credential,
|
|
)
|
|
|
|
def _unexpected_start_run(*_args, **_kwargs):
|
|
raise AssertionError("a generic configured key is not pool-lane recovery evidence")
|
|
|
|
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 with generic config key.]",
|
|
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
|
|
|
|
|
|
def test_process_wakeup_pause_keeps_pause_when_unusable_pool_entry_lacks_fingerprint(tmp_path, monkeypatch):
|
|
session = Session(
|
|
session_id="wakeup_pause_unusable_pool_entry_lacks_fingerprint",
|
|
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
|
|
session.save()
|
|
models.SESSIONS[session.session_id] = session
|
|
monkeypatch.setattr(providers, "_get_provider_api_key", lambda _provider: "generic-config-key")
|
|
monkeypatch.setattr(
|
|
providers,
|
|
"_pool_entry_payloads",
|
|
lambda _provider: [{"last_status": "dead"}],
|
|
)
|
|
monkeypatch.setattr(
|
|
routes,
|
|
"provider_has_process_wakeup_recovery_credential",
|
|
providers.provider_has_process_wakeup_recovery_credential,
|
|
)
|
|
|
|
def _unexpected_start_run(*_args, **_kwargs):
|
|
raise AssertionError("unknown exhausted pool secret must not prove recovery")
|
|
|
|
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 with unknown exhausted pool key.]",
|
|
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
|
|
|
|
|
|
def test_process_wakeup_pause_clears_when_config_key_differs_from_exhausted_pool_secret(tmp_path, monkeypatch):
|
|
session = Session(
|
|
session_id="wakeup_pause_config_key_differs_from_exhausted_pool",
|
|
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
|
|
session.save()
|
|
models.SESSIONS[session.session_id] = session
|
|
exhausted_fingerprint = hashlib.sha256(b"old-pool-key").hexdigest()
|
|
monkeypatch.setattr(providers, "_get_provider_api_key", lambda _provider: "new-config-key")
|
|
monkeypatch.setattr(
|
|
providers,
|
|
"_pool_entry_payloads",
|
|
lambda _provider: [{
|
|
"last_status": "dead",
|
|
"secret_fingerprint": exhausted_fingerprint,
|
|
}],
|
|
)
|
|
monkeypatch.setattr(
|
|
routes,
|
|
"provider_has_process_wakeup_recovery_credential",
|
|
providers.provider_has_process_wakeup_recovery_credential,
|
|
)
|
|
|
|
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-config-key-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 with new config key.]",
|
|
source="process_wakeup",
|
|
)
|
|
|
|
assert response["_status"] == 200
|
|
assert response["stream_id"] == "stream-config-key-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_successful_clear_serializes_against_concurrent_suppression(tmp_path, monkeypatch):
|
|
session = Session(
|
|
session_id="wakeup_pause_clear_vs_suppress_race",
|
|
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
|
|
session.save()
|
|
models.SESSIONS[session.session_id] = session
|
|
|
|
recovery_revalidating = threading.Event()
|
|
release_recovery = threading.Event()
|
|
responses = {}
|
|
starts = []
|
|
starts_lock = threading.Lock()
|
|
|
|
def _provider_has_recovery(_session, **_kwargs):
|
|
if threading.current_thread().name == "pause-clear":
|
|
recovery_revalidating.set()
|
|
assert release_recovery.wait(timeout=3)
|
|
return True
|
|
return False
|
|
|
|
def _fake_start_run(s, **kwargs):
|
|
with starts_lock:
|
|
starts.append((threading.current_thread().name, kwargs.get("source")))
|
|
return {"stream_id": f"stream-{threading.current_thread().name}", "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, "_process_wakeup_provider_has_recovery_credential", _provider_has_recovery)
|
|
monkeypatch.setattr(routes, "_start_run", _fake_start_run)
|
|
|
|
def _run(name):
|
|
responses[name] = routes.start_session_turn(
|
|
session.session_id,
|
|
f"[IMPORTANT: {name}]",
|
|
source="process_wakeup",
|
|
)
|
|
|
|
clear_thread = threading.Thread(target=_run, args=("clear",), name="pause-clear")
|
|
clear_thread.start()
|
|
assert recovery_revalidating.wait(timeout=3)
|
|
|
|
suppress_thread = threading.Thread(target=_run, args=("suppress",), name="pause-suppress")
|
|
suppress_thread.start()
|
|
release_recovery.set()
|
|
|
|
clear_thread.join(timeout=3)
|
|
suppress_thread.join(timeout=3)
|
|
assert not clear_thread.is_alive()
|
|
assert not suppress_thread.is_alive()
|
|
|
|
assert responses["clear"]["_status"] == 200
|
|
assert responses["suppress"]["_status"] == 200
|
|
assert all(resp.get("error") != PROCESS_WAKEUP_PAUSE_ERROR for resp in responses.values())
|
|
assert starts
|
|
saved = Session.load(session.session_id)
|
|
assert saved is not None
|
|
assert saved.process_wakeup_pause == {}
|
|
|
|
|
|
def test_streaming_success_pause_clear_serializes_against_concurrent_suppression(tmp_path, monkeypatch):
|
|
stream_id = "streaming-pause-clear-vs-suppress"
|
|
session_id = "streaming_pause_clear_vs_suppress"
|
|
stream_queue = queue.Queue()
|
|
config.STREAMS[stream_id] = stream_queue
|
|
previous_pause = {
|
|
"paused": True,
|
|
"model": "test-model",
|
|
"provider": "test-provider",
|
|
"classification": "credential_pool_empty",
|
|
"first_paused_at": 1.0,
|
|
"last_visible_error_at": 1.0,
|
|
"visible_error_count": 1,
|
|
"suppressed_count": 2,
|
|
"credential_state_fingerprint": "fingerprint-before",
|
|
}
|
|
session = Session(
|
|
session_id=session_id,
|
|
workspace=str(tmp_path),
|
|
model="test-model",
|
|
model_provider="test-provider",
|
|
messages=[{"role": "user", "content": "before", "timestamp": 1.0}],
|
|
context_messages=[{"role": "user", "content": "before"}],
|
|
active_stream_id=stream_id,
|
|
pending_user_message="recover",
|
|
pending_user_source="process_wakeup",
|
|
process_wakeup_pause=dict(previous_pause),
|
|
)
|
|
session.save()
|
|
models.SESSIONS[session_id] = session
|
|
|
|
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),
|
|
)
|
|
responses = {}
|
|
starts = []
|
|
|
|
def _fake_start_run(s, **kwargs):
|
|
starts.append((threading.current_thread().name, kwargs.get("source")))
|
|
return {
|
|
"stream_id": f"stream-{threading.current_thread().name}",
|
|
"session_id": s.session_id,
|
|
"_status": 200,
|
|
}
|
|
|
|
monkeypatch.setattr(routes, "_start_run", _fake_start_run)
|
|
|
|
suppress_started = threading.Event()
|
|
suppress_thread_holder = {}
|
|
|
|
def _run_suppress():
|
|
suppress_started.set()
|
|
responses["suppress"] = routes.start_session_turn(
|
|
session_id,
|
|
"[IMPORTANT: Background process completed while streaming success settles.]",
|
|
source="process_wakeup",
|
|
)
|
|
|
|
original_save = Session.save
|
|
save_calls = {"clear_save": 0}
|
|
|
|
def _save_and_race_suppression(self, *args, **kwargs):
|
|
if (
|
|
getattr(self, "session_id", None) == session_id
|
|
and save_calls["clear_save"] == 0
|
|
and not (getattr(self, "process_wakeup_pause", {}) or {})
|
|
):
|
|
lock = config.SESSION_AGENT_LOCKS.get(session_id)
|
|
assert lock is not None
|
|
assert lock.locked()
|
|
save_calls["clear_save"] += 1
|
|
suppress_thread = threading.Thread(target=_run_suppress, name="pause-suppress")
|
|
suppress_thread_holder["thread"] = suppress_thread
|
|
suppress_thread.start()
|
|
assert suppress_started.wait(timeout=3)
|
|
return original_save(self, *args, **kwargs)
|
|
|
|
monkeypatch.setattr(Session, "save", _save_and_race_suppression)
|
|
|
|
with mock.patch.object(streaming, "_get_ai_agent", return_value=_SuccessfulAgent), \
|
|
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,
|
|
)
|
|
|
|
suppress_thread = suppress_thread_holder.get("thread")
|
|
assert suppress_thread is not None
|
|
suppress_thread.join(timeout=3)
|
|
assert not suppress_thread.is_alive()
|
|
assert save_calls["clear_save"] == 1
|
|
assert responses["suppress"]["_status"] == 200
|
|
assert responses["suppress"]["stream_id"] == "stream-pause-suppress"
|
|
assert starts == [("pause-suppress", "process_wakeup")]
|
|
saved = Session.load(session_id)
|
|
assert saved is not None
|
|
assert saved.process_wakeup_pause == {}
|
|
|
|
|
|
def test_streaming_success_pause_clear_preserves_concurrent_session_update(tmp_path, monkeypatch):
|
|
stream_id = "streaming-pause-clear-preserves-rename"
|
|
session_id = "streaming_pause_clear_preserves_rename"
|
|
stream_queue = queue.Queue()
|
|
config.STREAMS[stream_id] = stream_queue
|
|
previous_pause = {
|
|
"paused": True,
|
|
"model": "test-model",
|
|
"provider": "test-provider",
|
|
"classification": "credential_pool_empty",
|
|
"first_paused_at": 1.0,
|
|
"last_visible_error_at": 1.0,
|
|
"visible_error_count": 1,
|
|
"suppressed_count": 2,
|
|
"credential_state_fingerprint": "fingerprint-before",
|
|
}
|
|
session = Session(
|
|
session_id=session_id,
|
|
title="Original title",
|
|
workspace=str(tmp_path),
|
|
model="test-model",
|
|
model_provider="test-provider",
|
|
messages=[{"role": "user", "content": "before", "timestamp": 1.0}],
|
|
context_messages=[{"role": "user", "content": "before"}],
|
|
active_stream_id=stream_id,
|
|
pending_user_message="recover",
|
|
pending_user_source="process_wakeup",
|
|
process_wakeup_pause=dict(previous_pause),
|
|
)
|
|
session.save()
|
|
models.SESSIONS[session_id] = session
|
|
|
|
original_save = Session.save
|
|
original_get_session = streaming.get_session
|
|
state = {"success_saved": False, "renamed": False}
|
|
|
|
def _save_and_mark_success_snapshot(self, *args, **kwargs):
|
|
result = original_save(self, *args, **kwargs)
|
|
if (
|
|
getattr(self, "session_id", None) == session_id
|
|
and getattr(self, "active_stream_id", None) is None
|
|
and any(
|
|
msg.get("role") == "assistant" and msg.get("content") == "Stream reply"
|
|
for msg in (getattr(self, "messages", None) or [])
|
|
)
|
|
):
|
|
state["success_saved"] = True
|
|
return result
|
|
|
|
def _get_session_after_concurrent_rename(sid, *args, **kwargs):
|
|
if sid == session_id and state["success_saved"] and not state["renamed"]:
|
|
latest = Session.load(session_id)
|
|
assert latest is not None
|
|
latest.title = "Renamed during settle"
|
|
original_save(latest, touch_updated_at=False)
|
|
with models.LOCK:
|
|
models.SESSIONS.pop(session_id, None)
|
|
state["renamed"] = True
|
|
return original_get_session(sid, *args, **kwargs)
|
|
|
|
monkeypatch.setattr(Session, "save", _save_and_mark_success_snapshot)
|
|
monkeypatch.setattr(streaming, "get_session", _get_session_after_concurrent_rename)
|
|
|
|
with mock.patch.object(streaming, "_get_ai_agent", return_value=_SuccessfulAgent), \
|
|
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,
|
|
)
|
|
|
|
assert state == {"success_saved": True, "renamed": True}
|
|
saved = Session.load(session_id)
|
|
assert saved is not None
|
|
assert saved.title == "Renamed during settle"
|
|
assert saved.process_wakeup_pause == {}
|
|
done_payloads = [item[1] for item in list(stream_queue.queue) if item[0] == "done"]
|
|
assert done_payloads
|
|
assert done_payloads[-1]["session"]["title"] == "Renamed during settle"
|
|
|
|
|
|
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_process_wakeup_recovery_credential",
|
|
providers.provider_has_process_wakeup_recovery_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_revalidates_at_provider_model_with_canonical_provider(tmp_path, monkeypatch):
|
|
session = Session(
|
|
session_id="wakeup_pause_at_model_recovered",
|
|
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
|
|
|
|
calls = []
|
|
|
|
def _provider_has_process_wakeup_recovery_credential(provider_id, *, refresh=False):
|
|
calls.append((provider_id, refresh))
|
|
return provider_id == "test-provider"
|
|
|
|
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-at-provider-recovered", "session_id": s.session_id, "_status": 200}
|
|
|
|
_patch_process_wakeup_route(
|
|
monkeypatch,
|
|
tmp_path,
|
|
model="@test-provider:test-model",
|
|
provider=None,
|
|
)
|
|
monkeypatch.setattr(
|
|
routes,
|
|
"provider_has_process_wakeup_recovery_credential",
|
|
_provider_has_process_wakeup_recovery_credential,
|
|
)
|
|
monkeypatch.setattr(routes, "_start_run", _fake_start_run)
|
|
|
|
response = routes.start_session_turn(
|
|
session.session_id,
|
|
"[IMPORTANT: Background process completed after @provider recovery.]",
|
|
source="process_wakeup",
|
|
)
|
|
|
|
assert response["_status"] == 200
|
|
assert response["stream_id"] == "stream-at-provider-recovered"
|
|
assert calls == [("test-provider", True)]
|
|
assert captured == {
|
|
"source": "process_wakeup",
|
|
"model": "@test-provider:test-model",
|
|
"model_provider": None,
|
|
}
|
|
saved = Session.load(session.session_id)
|
|
assert saved is not None
|
|
assert saved.process_wakeup_pause == {}
|
|
|
|
|
|
def test_process_wakeup_pause_revalidation_uses_session_profile_not_default(tmp_path, monkeypatch):
|
|
base_home = tmp_path / "hermes-home"
|
|
(base_home / "profiles" / "work").mkdir(parents=True)
|
|
monkeypatch.setattr(profiles, "_DEFAULT_HERMES_HOME", base_home)
|
|
monkeypatch.setattr(profiles, "_is_isolated_profile_mode", lambda: False)
|
|
profiles.clear_request_profile()
|
|
|
|
session = Session(
|
|
session_id="wakeup_pause_named_profile_revalidation",
|
|
workspace=str(tmp_path),
|
|
model="test-model",
|
|
model_provider="test-provider",
|
|
profile="work",
|
|
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": 0,
|
|
},
|
|
)
|
|
session.process_wakeup_pause["credential_state_fingerprint"] = (
|
|
models.process_wakeup_credential_state_fingerprint(session)
|
|
)
|
|
session.save()
|
|
models.SESSIONS[session.session_id] = session
|
|
|
|
calls = []
|
|
|
|
def _provider_has_process_wakeup_recovery_credential(provider_id, *, refresh=False):
|
|
active_profile = profiles.get_active_profile_name()
|
|
calls.append((provider_id, refresh, active_profile))
|
|
return active_profile == "default"
|
|
|
|
def _unexpected_start_run(*_args, **_kwargs):
|
|
raise AssertionError("named-profile wakeup must not clear from default-profile credentials")
|
|
|
|
_patch_process_wakeup_route(
|
|
monkeypatch,
|
|
tmp_path,
|
|
model="test-model",
|
|
provider="test-provider",
|
|
)
|
|
monkeypatch.setattr(
|
|
routes,
|
|
"provider_has_process_wakeup_recovery_credential",
|
|
_provider_has_process_wakeup_recovery_credential,
|
|
)
|
|
monkeypatch.setattr(routes, "_start_run", _unexpected_start_run)
|
|
|
|
try:
|
|
response = routes.start_session_turn(
|
|
session.session_id,
|
|
"[IMPORTANT: Background process completed while named profile is still exhausted.]",
|
|
source="process_wakeup",
|
|
)
|
|
finally:
|
|
profiles.clear_request_profile()
|
|
|
|
assert response["_status"] == 409
|
|
assert response["error"] == PROCESS_WAKEUP_PAUSE_ERROR
|
|
assert calls == [("test-provider", True, "work")]
|
|
saved = Session.load(session.session_id)
|
|
assert saved is not None
|
|
assert saved.process_wakeup_pause["suppressed_count"] == 1
|
|
assert profiles.get_active_profile_name() == "default"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("recovered_status", "recovered_reset_at"),
|
|
[
|
|
("ok", None),
|
|
("exhausted", "2000-01-01T00:00:00Z"),
|
|
],
|
|
)
|
|
def test_process_wakeup_pause_revalidates_named_profile_sanitized_env_pool_recovery(
|
|
tmp_path,
|
|
monkeypatch,
|
|
recovered_status,
|
|
recovered_reset_at,
|
|
):
|
|
base_home = tmp_path / "hermes-home"
|
|
profile_home = base_home / "profiles" / "work"
|
|
profile_home.mkdir(parents=True)
|
|
api_key = "sk-or-profile-recovered"
|
|
(profile_home / ".env").write_text(f"OPENROUTER_API_KEY={api_key}\n", encoding="utf-8")
|
|
auth_json = profile_home / "auth.json"
|
|
secret_fingerprint = hashlib.sha256(api_key.encode("utf-8")).hexdigest()[:16]
|
|
|
|
def _write_pool_entry(status, reset_at):
|
|
entry = {
|
|
"id": "openrouter-env",
|
|
"label": "OpenRouter env key",
|
|
"source": "env",
|
|
"auth_type": "api_key",
|
|
"secret_fingerprint": f"sha256:{secret_fingerprint}",
|
|
"last_status": status,
|
|
"last_status_at": "2026-07-10T00:00:00Z",
|
|
"last_error_code": "429",
|
|
}
|
|
if reset_at is not None:
|
|
entry["last_error_reset_at"] = reset_at
|
|
auth_json.write_text(
|
|
json.dumps({"credential_pool": {"openrouter": [entry]}}),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
fake_auth_module = types.ModuleType("hermes_cli.auth")
|
|
|
|
def _read_credential_pool(provider_id):
|
|
raw = json.loads(config._get_auth_store_path().read_text(encoding="utf-8"))
|
|
return list((raw.get("credential_pool") or {}).get(provider_id, []))
|
|
|
|
fake_auth_module.read_credential_pool = _read_credential_pool
|
|
monkeypatch.setitem(sys.modules, "hermes_cli.auth", fake_auth_module)
|
|
fake_hermes_cli = sys.modules.get("hermes_cli")
|
|
if fake_hermes_cli is not None:
|
|
fake_hermes_cli.auth = fake_auth_module
|
|
|
|
monkeypatch.setattr(profiles, "_DEFAULT_HERMES_HOME", base_home)
|
|
monkeypatch.setattr(profiles, "_is_isolated_profile_mode", lambda: False)
|
|
profiles.clear_request_profile()
|
|
|
|
_write_pool_entry("exhausted", "2999-01-01T00:00:00Z")
|
|
session = Session(
|
|
session_id=f"wakeup_pause_named_profile_sanitized_env_{recovered_status}",
|
|
workspace=str(tmp_path),
|
|
model="test-model",
|
|
model_provider="openrouter",
|
|
profile="work",
|
|
)
|
|
pause = models.record_process_wakeup_provider_unavailable_pause(
|
|
session,
|
|
classification="credential_pool_empty",
|
|
model="test-model",
|
|
provider="openrouter",
|
|
)
|
|
assert pause is not None
|
|
paused_fingerprint = pause["credential_state_fingerprint"]
|
|
session.save()
|
|
models.SESSIONS[session.session_id] = session
|
|
|
|
_write_pool_entry(recovered_status, recovered_reset_at)
|
|
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": f"stream-named-profile-recovered-{recovered_status}",
|
|
"session_id": s.session_id,
|
|
"_status": 200,
|
|
}
|
|
|
|
_patch_process_wakeup_route(
|
|
monkeypatch,
|
|
tmp_path,
|
|
model="test-model",
|
|
provider="openrouter",
|
|
)
|
|
monkeypatch.setattr(
|
|
routes,
|
|
"provider_has_process_wakeup_recovery_credential",
|
|
providers.provider_has_process_wakeup_recovery_credential,
|
|
)
|
|
monkeypatch.setattr(routes, "_start_run", _fake_start_run)
|
|
|
|
try:
|
|
response = routes.start_session_turn(
|
|
session.session_id,
|
|
"[IMPORTANT: Background process completed after named profile credential recovery.]",
|
|
source="process_wakeup",
|
|
)
|
|
finally:
|
|
profiles.clear_request_profile()
|
|
|
|
assert response["_status"] == 200
|
|
assert response["stream_id"] == f"stream-named-profile-recovered-{recovered_status}"
|
|
assert captured == {
|
|
"source": "process_wakeup",
|
|
"model": "test-model",
|
|
"model_provider": "openrouter",
|
|
}
|
|
saved = Session.load(session.session_id)
|
|
assert saved is not None
|
|
assert saved.process_wakeup_pause == {}
|
|
assert profiles.get_active_profile_name() == "default"
|
|
|
|
|
|
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_process_wakeup_pause_keeps_empty_provider_lane_after_fingerprint_change(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)
|
|
monkeypatch.setitem(config.cfg, "model", {"default": "claude-sonnet-test"})
|
|
session = Session(
|
|
session_id="wakeup_pause_empty_provider_probe",
|
|
workspace=str(tmp_path),
|
|
model="claude-sonnet-test",
|
|
model_provider=None,
|
|
)
|
|
pause = models.record_process_wakeup_provider_unavailable_pause(
|
|
session,
|
|
classification="credential_pool_empty",
|
|
model="claude-sonnet-test",
|
|
provider=None,
|
|
)
|
|
assert pause is not None
|
|
assert pause["provider"] == ""
|
|
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",
|
|
)
|
|
changed_fingerprint = models.process_wakeup_credential_state_fingerprint(session)
|
|
assert changed_fingerprint != paused_fingerprint
|
|
|
|
def _unexpected_start_run(*_args, **_kwargs):
|
|
raise AssertionError("empty-provider lane must prove recovery before restarting")
|
|
|
|
_patch_process_wakeup_route(
|
|
monkeypatch,
|
|
tmp_path,
|
|
model="claude-sonnet-test",
|
|
provider=None,
|
|
)
|
|
monkeypatch.setattr(
|
|
routes,
|
|
"provider_has_process_wakeup_recovery_credential",
|
|
lambda *_args, **_kwargs: False,
|
|
)
|
|
monkeypatch.setattr(routes, "_start_run", _unexpected_start_run)
|
|
|
|
response = routes.start_session_turn(
|
|
session.session_id,
|
|
"[IMPORTANT: Background process completed after empty-provider credential change.]",
|
|
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"] == changed_fingerprint
|
|
|
|
|
|
def test_gateway_cancel_during_completion_save_restores_process_wakeup_pause(tmp_path, monkeypatch):
|
|
stream_id = "gateway-pause-save-race-stream"
|
|
session_id = "gateway_pause_save_race"
|
|
stream_queue = queue.Queue()
|
|
config.STREAMS[stream_id] = stream_queue
|
|
monkeypatch.setattr(gateway_chat, "RunJournalWriter", lambda *_args, **_kwargs: None)
|
|
monkeypatch.setattr(gateway_chat, "gateway_approval_unavailable_reason", lambda *_args, **_kwargs: None)
|
|
monkeypatch.setattr(config, "get_config", lambda: {"webui_gateway_base_url": "http://gateway.test"})
|
|
|
|
class _GatewayResponse:
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *_exc):
|
|
return False
|
|
|
|
def __iter__(self):
|
|
payload = {"choices": [{"delta": {"content": "Gateway reply"}}]}
|
|
return iter([
|
|
("data: " + json.dumps(payload) + "\n").encode("utf-8"),
|
|
b"data: [DONE]\n",
|
|
])
|
|
|
|
monkeypatch.setattr(
|
|
gateway_chat.urllib.request,
|
|
"urlopen",
|
|
lambda *_args, **_kwargs: _GatewayResponse(),
|
|
)
|
|
|
|
previous_pause = {
|
|
"paused": True,
|
|
"model": "claude-sonnet-test",
|
|
"provider": "test-provider",
|
|
"classification": "credential_pool_empty",
|
|
"first_paused_at": 1.0,
|
|
"last_visible_error_at": 1.0,
|
|
"visible_error_count": 1,
|
|
"suppressed_count": 0,
|
|
"credential_state_fingerprint": "fingerprint-before",
|
|
}
|
|
previous_messages = [{"role": "user", "content": "before", "timestamp": 1.0}]
|
|
session = Session(
|
|
session_id=session_id,
|
|
workspace=str(tmp_path),
|
|
model="claude-sonnet-test",
|
|
model_provider="test-provider",
|
|
messages=list(previous_messages),
|
|
context_messages=list(previous_messages),
|
|
active_stream_id=stream_id,
|
|
pending_user_message="wake up",
|
|
pending_user_source="process_wakeup",
|
|
process_wakeup_pause=dict(previous_pause),
|
|
)
|
|
session.save()
|
|
models.SESSIONS[session_id] = session
|
|
|
|
original_save = Session.save
|
|
save_calls = {"count": 0, "cleared_pause": 0}
|
|
|
|
def _save_and_cancel_after_success_clear(self, *args, **kwargs):
|
|
save_calls["count"] += 1
|
|
result = original_save(self, *args, **kwargs)
|
|
if (
|
|
getattr(self, "session_id", None) == session_id
|
|
and not getattr(self, "process_wakeup_pause", None)
|
|
and save_calls["cleared_pause"] == 0
|
|
):
|
|
save_calls["cleared_pause"] += 1
|
|
config.CANCEL_FLAGS[stream_id].set()
|
|
return result
|
|
|
|
monkeypatch.setattr(Session, "save", _save_and_cancel_after_success_clear)
|
|
|
|
gateway_chat._run_gateway_chat_streaming(
|
|
session_id,
|
|
"wake up",
|
|
"claude-sonnet-test",
|
|
str(tmp_path),
|
|
stream_id,
|
|
model_provider="test-provider",
|
|
)
|
|
|
|
assert save_calls["cleared_pause"] == 1
|
|
saved = Session.load(session_id)
|
|
assert saved is not None
|
|
assert saved.process_wakeup_pause == previous_pause
|
|
assert saved.messages == previous_messages
|
|
assert saved.context_messages == previous_messages
|
|
queued_events = [item[0] for item in list(stream_queue.queue)]
|
|
assert "cancel" in queued_events
|
|
assert "done" not in queued_events
|
|
|
|
|
|
def test_gateway_late_cancel_preserves_completed_webui_turn(tmp_path, monkeypatch):
|
|
stream_id = "gateway-webui-late-cancel-stream"
|
|
session_id = "gateway_webui_late_cancel"
|
|
stream_queue = queue.Queue()
|
|
config.STREAMS[stream_id] = stream_queue
|
|
monkeypatch.setattr(gateway_chat, "RunJournalWriter", lambda *_args, **_kwargs: None)
|
|
monkeypatch.setattr(gateway_chat, "gateway_approval_unavailable_reason", lambda *_args, **_kwargs: None)
|
|
monkeypatch.setattr(config, "get_config", lambda: {"webui_gateway_base_url": "http://gateway.test"})
|
|
|
|
class _GatewayResponse:
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *_exc):
|
|
return False
|
|
|
|
def __iter__(self):
|
|
payload = {"choices": [{"delta": {"content": "Gateway reply"}}]}
|
|
return iter([
|
|
("data: " + json.dumps(payload) + "\n").encode("utf-8"),
|
|
b"data: [DONE]\n",
|
|
])
|
|
|
|
monkeypatch.setattr(
|
|
gateway_chat.urllib.request,
|
|
"urlopen",
|
|
lambda *_args, **_kwargs: _GatewayResponse(),
|
|
)
|
|
|
|
previous_messages = [{"role": "user", "content": "before", "timestamp": 1.0}]
|
|
session = Session(
|
|
session_id=session_id,
|
|
workspace=str(tmp_path),
|
|
model="claude-sonnet-test",
|
|
model_provider="test-provider",
|
|
messages=list(previous_messages),
|
|
context_messages=list(previous_messages),
|
|
active_stream_id=stream_id,
|
|
pending_user_message="hello",
|
|
pending_user_source="webui",
|
|
)
|
|
session.save()
|
|
models.SESSIONS[session_id] = session
|
|
|
|
original_save = Session.save
|
|
save_calls = {"completed": 0}
|
|
|
|
def _save_and_cancel_after_completed_webui_turn(self, *args, **kwargs):
|
|
result = original_save(self, *args, **kwargs)
|
|
if (
|
|
getattr(self, "session_id", None) == session_id
|
|
and save_calls["completed"] == 0
|
|
and getattr(self, "active_stream_id", None) is None
|
|
and getattr(self, "pending_user_message", None) is None
|
|
and any(
|
|
msg.get("role") == "assistant" and msg.get("content") == "Gateway reply"
|
|
for msg in (getattr(self, "messages", None) or [])
|
|
)
|
|
):
|
|
save_calls["completed"] += 1
|
|
config.CANCEL_FLAGS[stream_id].set()
|
|
return result
|
|
|
|
monkeypatch.setattr(Session, "save", _save_and_cancel_after_completed_webui_turn)
|
|
|
|
gateway_chat._run_gateway_chat_streaming(
|
|
session_id,
|
|
"hello",
|
|
"claude-sonnet-test",
|
|
str(tmp_path),
|
|
stream_id,
|
|
model_provider="test-provider",
|
|
)
|
|
|
|
assert save_calls["completed"] == 1
|
|
saved = Session.load(session_id)
|
|
assert saved is not None
|
|
assert saved.active_stream_id is None
|
|
assert saved.pending_user_message is None
|
|
assert [msg.get("content") for msg in saved.messages] == [
|
|
"before",
|
|
"hello",
|
|
"Gateway reply",
|
|
]
|
|
assert [msg.get("content") for msg in saved.context_messages] == [
|
|
"before",
|
|
"hello",
|
|
"Gateway reply",
|
|
]
|
|
queued_events = [item[0] for item in list(stream_queue.queue)]
|
|
assert "cancel" in queued_events
|
|
assert "done" not in queued_events
|
|
|
|
|
|
def test_gateway_late_cancel_preserves_existing_pause_for_webui_recovery(tmp_path, monkeypatch):
|
|
stream_id = "gateway-webui-recovery-late-cancel-stream"
|
|
session_id = "gateway_webui_recovery_late_cancel"
|
|
stream_queue = queue.Queue()
|
|
config.STREAMS[stream_id] = stream_queue
|
|
monkeypatch.setattr(gateway_chat, "RunJournalWriter", lambda *_args, **_kwargs: None)
|
|
monkeypatch.setattr(gateway_chat, "gateway_approval_unavailable_reason", lambda *_args, **_kwargs: None)
|
|
monkeypatch.setattr(config, "get_config", lambda: {"webui_gateway_base_url": "http://gateway.test"})
|
|
|
|
class _GatewayResponse:
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *_exc):
|
|
return False
|
|
|
|
def __iter__(self):
|
|
payload = {"choices": [{"delta": {"content": "Gateway recovery reply"}}]}
|
|
return iter([
|
|
("data: " + json.dumps(payload) + "\n").encode("utf-8"),
|
|
b"data: [DONE]\n",
|
|
])
|
|
|
|
monkeypatch.setattr(
|
|
gateway_chat.urllib.request,
|
|
"urlopen",
|
|
lambda *_args, **_kwargs: _GatewayResponse(),
|
|
)
|
|
|
|
previous_pause = {
|
|
"paused": True,
|
|
"model": "claude-sonnet-test",
|
|
"provider": "test-provider",
|
|
"classification": "credential_pool_empty",
|
|
"first_paused_at": 1.0,
|
|
"last_visible_error_at": 1.0,
|
|
"visible_error_count": 1,
|
|
"suppressed_count": 2,
|
|
"credential_state_fingerprint": "fingerprint-before",
|
|
}
|
|
previous_messages = [{"role": "user", "content": "before", "timestamp": 1.0}]
|
|
session = Session(
|
|
session_id=session_id,
|
|
workspace=str(tmp_path),
|
|
model="claude-sonnet-test",
|
|
model_provider="test-provider",
|
|
messages=list(previous_messages),
|
|
context_messages=list(previous_messages),
|
|
active_stream_id=stream_id,
|
|
pending_user_message="try recovery",
|
|
pending_user_source="webui",
|
|
process_wakeup_pause=dict(previous_pause),
|
|
)
|
|
session.save()
|
|
models.SESSIONS[session_id] = session
|
|
|
|
original_save = Session.save
|
|
save_calls = {"completed": 0}
|
|
|
|
def _save_and_cancel_after_completed_recovery_turn(self, *args, **kwargs):
|
|
result = original_save(self, *args, **kwargs)
|
|
if (
|
|
getattr(self, "session_id", None) == session_id
|
|
and save_calls["completed"] == 0
|
|
and getattr(self, "active_stream_id", None) is None
|
|
and getattr(self, "pending_user_message", None) is None
|
|
and any(
|
|
msg.get("role") == "assistant" and msg.get("content") == "Gateway recovery reply"
|
|
for msg in (getattr(self, "messages", None) or [])
|
|
)
|
|
):
|
|
save_calls["completed"] += 1
|
|
config.CANCEL_FLAGS[stream_id].set()
|
|
return result
|
|
|
|
monkeypatch.setattr(Session, "save", _save_and_cancel_after_completed_recovery_turn)
|
|
|
|
gateway_chat._run_gateway_chat_streaming(
|
|
session_id,
|
|
"try recovery",
|
|
"claude-sonnet-test",
|
|
str(tmp_path),
|
|
stream_id,
|
|
model_provider="test-provider",
|
|
)
|
|
|
|
assert save_calls["completed"] == 1
|
|
saved = Session.load(session_id)
|
|
assert saved is not None
|
|
assert saved.process_wakeup_pause == previous_pause
|
|
assert [msg.get("content") for msg in saved.messages] == [
|
|
"before",
|
|
"try recovery",
|
|
"Gateway recovery reply",
|
|
]
|
|
assert [msg.get("content") for msg in saved.context_messages] == [
|
|
"before",
|
|
"try recovery",
|
|
"Gateway recovery reply",
|
|
]
|
|
queued_events = [item[0] for item in list(stream_queue.queue)]
|
|
assert "cancel" in queued_events
|
|
assert "done" not in queued_events
|
|
|
|
|
|
def test_gateway_post_save_cancel_after_success_commit_emits_done(tmp_path, monkeypatch):
|
|
stream_id = "gateway-post-save-success-cancel-stream"
|
|
session_id = "gateway_post_save_success_cancel"
|
|
stream_queue = queue.Queue()
|
|
config.STREAMS[stream_id] = stream_queue
|
|
monkeypatch.setattr(gateway_chat, "RunJournalWriter", lambda *_args, **_kwargs: None)
|
|
monkeypatch.setattr(gateway_chat, "gateway_approval_unavailable_reason", lambda *_args, **_kwargs: None)
|
|
monkeypatch.setattr(config, "get_config", lambda: {"webui_gateway_base_url": "http://gateway.test"})
|
|
|
|
class _GatewayResponse:
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *_exc):
|
|
return False
|
|
|
|
def __iter__(self):
|
|
payload = {"choices": [{"delta": {"content": "Gateway success reply"}}]}
|
|
return iter([
|
|
("data: " + json.dumps(payload) + "\n").encode("utf-8"),
|
|
b"data: [DONE]\n",
|
|
])
|
|
|
|
monkeypatch.setattr(
|
|
gateway_chat.urllib.request,
|
|
"urlopen",
|
|
lambda *_args, **_kwargs: _GatewayResponse(),
|
|
)
|
|
|
|
previous_pause = {
|
|
"paused": True,
|
|
"model": "claude-sonnet-test",
|
|
"provider": "test-provider",
|
|
"classification": "credential_pool_empty",
|
|
"first_paused_at": 1.0,
|
|
"last_visible_error_at": 1.0,
|
|
"visible_error_count": 1,
|
|
"suppressed_count": 2,
|
|
"credential_state_fingerprint": "fingerprint-before",
|
|
}
|
|
previous_messages = [{"role": "user", "content": "before", "timestamp": 1.0}]
|
|
session = Session(
|
|
session_id=session_id,
|
|
workspace=str(tmp_path),
|
|
model="claude-sonnet-test",
|
|
model_provider="test-provider",
|
|
messages=list(previous_messages),
|
|
context_messages=list(previous_messages),
|
|
active_stream_id=stream_id,
|
|
pending_user_message="try recovery",
|
|
pending_user_source="webui",
|
|
process_wakeup_pause=dict(previous_pause),
|
|
)
|
|
session.save()
|
|
models.SESSIONS[session_id] = session
|
|
|
|
original_payload = streaming._session_payload_with_full_messages
|
|
payload_calls = {"count": 0}
|
|
|
|
def _payload_and_cancel_after_success_commit(*args, **kwargs):
|
|
payload_calls["count"] += 1
|
|
config.CANCEL_FLAGS[stream_id].set()
|
|
return original_payload(*args, **kwargs)
|
|
|
|
monkeypatch.setattr(streaming, "_session_payload_with_full_messages", _payload_and_cancel_after_success_commit)
|
|
|
|
gateway_chat._run_gateway_chat_streaming(
|
|
session_id,
|
|
"try recovery",
|
|
"claude-sonnet-test",
|
|
str(tmp_path),
|
|
stream_id,
|
|
model_provider="test-provider",
|
|
)
|
|
|
|
assert payload_calls["count"] >= 1
|
|
saved = Session.load(session_id)
|
|
assert saved is not None
|
|
assert saved.process_wakeup_pause == {}
|
|
queued_events = [item[0] for item in list(stream_queue.queue)]
|
|
assert "done" in queued_events
|
|
assert "stream_end" in queued_events
|
|
assert "cancel" not in queued_events
|
|
|
|
|
|
def test_streaming_late_cancel_after_pause_clear_save_persists_restored_pause(tmp_path, monkeypatch):
|
|
stream_id = "streaming-pause-clear-save-cancel"
|
|
session_id = "streaming_pause_clear_save_cancel"
|
|
stream_queue = queue.Queue()
|
|
config.STREAMS[stream_id] = stream_queue
|
|
|
|
previous_pause = {
|
|
"paused": True,
|
|
"model": "test-model",
|
|
"provider": "test-provider",
|
|
"classification": "credential_pool_empty",
|
|
"first_paused_at": 1.0,
|
|
"last_visible_error_at": 1.0,
|
|
"visible_error_count": 1,
|
|
"suppressed_count": 3,
|
|
"credential_state_fingerprint": "fingerprint-before",
|
|
}
|
|
previous_messages = [{"role": "user", "content": "before", "timestamp": 1.0}]
|
|
session = Session(
|
|
session_id=session_id,
|
|
workspace=str(tmp_path),
|
|
model="test-model",
|
|
model_provider="test-provider",
|
|
messages=list(previous_messages),
|
|
context_messages=list(previous_messages),
|
|
active_stream_id=stream_id,
|
|
pending_user_message="wake up",
|
|
pending_user_source="process_wakeup",
|
|
process_wakeup_pause=dict(previous_pause),
|
|
)
|
|
session.save()
|
|
models.SESSIONS[session_id] = session
|
|
|
|
original_save = Session.save
|
|
save_calls = {"cleared_pause": 0, "restored_pause": 0}
|
|
|
|
def _save_and_cancel_after_pause_clear(self, *args, **kwargs):
|
|
result = original_save(self, *args, **kwargs)
|
|
if getattr(self, "session_id", None) != session_id:
|
|
return result
|
|
pause = dict(getattr(self, "process_wakeup_pause", {}) or {})
|
|
if not pause and save_calls["cleared_pause"] == 0:
|
|
save_calls["cleared_pause"] += 1
|
|
config.CANCEL_FLAGS[stream_id].set()
|
|
elif pause == previous_pause and config.CANCEL_FLAGS[stream_id].is_set():
|
|
save_calls["restored_pause"] += 1
|
|
return result
|
|
|
|
monkeypatch.setattr(Session, "save", _save_and_cancel_after_pause_clear)
|
|
|
|
with mock.patch.object(streaming, "_get_ai_agent", return_value=_SuccessfulAgent), \
|
|
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,
|
|
)
|
|
|
|
assert save_calls["cleared_pause"] == 1
|
|
assert save_calls["restored_pause"] >= 1
|
|
saved = Session.load(session_id)
|
|
assert saved is not None
|
|
assert saved.process_wakeup_pause == previous_pause
|
|
queued_events = [item[0] for item in list(stream_queue.queue)]
|
|
assert "cancel" in queued_events
|
|
assert "done" not in queued_events
|
|
|
|
|
|
def test_streaming_post_save_cancel_after_success_commit_emits_done(tmp_path, monkeypatch):
|
|
stream_id = "streaming-pause-clear-post-save-cancel"
|
|
session_id = "streaming_pause_clear_post_save_cancel"
|
|
stream_queue = queue.Queue()
|
|
config.STREAMS[stream_id] = stream_queue
|
|
|
|
previous_pause = {
|
|
"paused": True,
|
|
"model": "test-model",
|
|
"provider": "test-provider",
|
|
"classification": "credential_pool_empty",
|
|
"first_paused_at": 1.0,
|
|
"last_visible_error_at": 1.0,
|
|
"visible_error_count": 1,
|
|
"suppressed_count": 3,
|
|
"credential_state_fingerprint": "fingerprint-before",
|
|
}
|
|
previous_messages = [{"role": "user", "content": "before", "timestamp": 1.0}]
|
|
session = Session(
|
|
session_id=session_id,
|
|
workspace=str(tmp_path),
|
|
model="test-model",
|
|
model_provider="test-provider",
|
|
messages=list(previous_messages),
|
|
context_messages=list(previous_messages),
|
|
active_stream_id=stream_id,
|
|
pending_user_message="wake up",
|
|
pending_user_source="process_wakeup",
|
|
process_wakeup_pause=dict(previous_pause),
|
|
)
|
|
session.save()
|
|
models.SESSIONS[session_id] = session
|
|
|
|
original_payload = streaming._session_payload_with_full_messages
|
|
payload_calls = {"count": 0}
|
|
|
|
def _payload_and_cancel_after_success_commit(*args, **kwargs):
|
|
payload_calls["count"] += 1
|
|
config.CANCEL_FLAGS[stream_id].set()
|
|
return original_payload(*args, **kwargs)
|
|
|
|
monkeypatch.setattr(streaming, "_session_payload_with_full_messages", _payload_and_cancel_after_success_commit)
|
|
|
|
with mock.patch.object(streaming, "_get_ai_agent", return_value=_SuccessfulAgent), \
|
|
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,
|
|
)
|
|
|
|
assert payload_calls["count"] >= 1
|
|
saved = Session.load(session_id)
|
|
assert saved is not None
|
|
assert saved.process_wakeup_pause == {}
|
|
queued_events = [item[0] for item in list(stream_queue.queue)]
|
|
assert "done" in queued_events
|
|
assert "cancel" not in queued_events
|
|
|
|
|
|
def test_streaming_no_pause_post_save_cancel_after_success_commit_emits_done(tmp_path, monkeypatch):
|
|
stream_id = "streaming-no-pause-post-save-cancel"
|
|
session_id = "streaming_no_pause_post_save_cancel"
|
|
stream_queue = queue.Queue()
|
|
config.STREAMS[stream_id] = stream_queue
|
|
|
|
previous_messages = [{"role": "user", "content": "before", "timestamp": 1.0}]
|
|
session = Session(
|
|
session_id=session_id,
|
|
workspace=str(tmp_path),
|
|
model="test-model",
|
|
model_provider="test-provider",
|
|
messages=list(previous_messages),
|
|
context_messages=list(previous_messages),
|
|
active_stream_id=stream_id,
|
|
pending_user_message="hello",
|
|
pending_user_source="webui",
|
|
)
|
|
session.save()
|
|
models.SESSIONS[session_id] = session
|
|
|
|
original_payload = streaming._session_payload_with_full_messages
|
|
payload_calls = {"count": 0}
|
|
|
|
def _payload_and_cancel_after_success_commit(*args, **kwargs):
|
|
payload_calls["count"] += 1
|
|
config.CANCEL_FLAGS[stream_id].set()
|
|
return original_payload(*args, **kwargs)
|
|
|
|
monkeypatch.setattr(streaming, "_session_payload_with_full_messages", _payload_and_cancel_after_success_commit)
|
|
|
|
with mock.patch.object(streaming, "_get_ai_agent", return_value=_SuccessfulAgent), \
|
|
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,
|
|
)
|
|
|
|
assert payload_calls["count"] >= 1
|
|
saved = Session.load(session_id)
|
|
assert saved is not None
|
|
assert saved.process_wakeup_pause == {}
|
|
assert saved.active_stream_id is None
|
|
assert saved.pending_user_message is None
|
|
assert [msg.get("content") for msg in saved.messages] == [
|
|
"before",
|
|
"hello",
|
|
"Stream reply",
|
|
]
|
|
queued_events = [item[0] for item in list(stream_queue.queue)]
|
|
assert "done" in queued_events
|
|
assert "stream_end" in queued_events
|
|
assert "cancel" not in queued_events
|
|
|
|
|
|
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")')
|
|
session_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_snapshot_idx = src.index("_process_wakeup_pause_before_clear =", final_cancel_idx)
|
|
pause_clear_idx = src.index("clear_process_wakeup_pause(s, reason='run_completed')")
|
|
post_clear_cancel_idx = src.index("if cancel_event.is_set():", pause_clear_idx)
|
|
post_clear_restore_idx = src.index(
|
|
"s.process_wakeup_pause = dict(_process_wakeup_pause_before_clear)",
|
|
post_clear_cancel_idx,
|
|
)
|
|
pause_save_idx = src.index('"process_wakeup_pause_clear_save"', post_clear_cancel_idx)
|
|
post_save_cancel_idx = src.index("if cancel_event.is_set():", pause_save_idx)
|
|
post_save_restore_idx = src.index(
|
|
"s.process_wakeup_pause = dict(_process_wakeup_pause_before_clear)",
|
|
post_save_cancel_idx,
|
|
)
|
|
done_payload_idx = src.index('with _stream_writeback_stage(_writeback_timings, "done_payload")')
|
|
|
|
assert session_save_idx < session_save_cancel_idx < state_sync_idx
|
|
assert state_sync_idx < final_cancel_idx < pause_snapshot_idx < pause_clear_idx
|
|
assert pause_clear_idx < post_clear_cancel_idx < post_clear_restore_idx < pause_save_idx
|
|
assert pause_save_idx < post_save_cancel_idx < post_save_restore_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")
|
|
pre_clear_comment_idx = src.index("# Recheck immediately before clearing", pending_clear_idx)
|
|
final_cancel_idx = src.index("if cancel_event.is_set():", pre_clear_comment_idx)
|
|
pause_clear_idx = src.index(
|
|
'clear_process_wakeup_pause(s, reason="run_completed")',
|
|
final_cancel_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
|