jinghui/hermes-webui
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(wakeup): pause credential-exhausted process wakeups

Browse Source
This commit is contained in:
Frank Song committed 2026-07-07 11:45:01 +08:00
1 parent 1d8208026c
commit bcb411db53
6 files changed
+584 -3
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

api/background_process.py
+6 -1
View File
@@ -1239,7 +1239,12 @@ def _start_server_side_wakeup_turn(
session_id, wakeup_prompt, source="process_wakeup"
)
status = int((resp or {}).get("_status", 200) or 200)
if status == 409:
if status == 409 and (resp or {}).get("error") == "process_wakeup_paused":
logger.info(
"server-side wakeup suppressed for session %s: provider credential state is paused",
session_id,
)
elif status == 409:
# Raced an active turn (e.g. a human /api/chat/start, or a
# sibling deferred-wakeup thread). Re-defer this prompt so it
# is delivered by the winning turn's teardown / next-turn drain
api/models.py
+156
View File
@@ -1100,6 +1100,7 @@ class Session:
enabled_toolsets=None,
composer_draft=None,
anchor_activity_scenes=None,
process_wakeup_pause=None,
**kwargs):
self.session_id = session_id or uuid.uuid4().hex[:12]
self.title = title
@@ -1171,6 +1172,7 @@ class Session:
self.enabled_toolsets = enabled_toolsets # List[str] or None — per-session toolset override
self.composer_draft = composer_draft if isinstance(composer_draft, dict) else {}
self.anchor_activity_scenes = anchor_activity_scenes if isinstance(anchor_activity_scenes, dict) else {}
self.process_wakeup_pause = process_wakeup_pause if isinstance(process_wakeup_pause, dict) else {}
raw_message_count = kwargs.get('message_count')
parsed_message_count = None
if raw_message_count is not None:
@@ -1231,6 +1233,7 @@ class Session:
'worktree_path', 'worktree_branch', 'worktree_repo_root', 'worktree_created_at',
'is_cli_session', 'source_tag', 'raw_source', 'session_source', 'source_label', 'read_only',
'enabled_toolsets', 'composer_draft', 'anchor_activity_scenes',
'process_wakeup_pause',
]
meta = {k: getattr(self, k, None) for k in METADATA_FIELDS}
meta['message_count'] = len(self.messages or [])
@@ -1537,11 +1540,164 @@ class Session:
'read_only': self.read_only,
'enabled_toolsets': self.enabled_toolsets,
'composer_draft': self.composer_draft if isinstance(self.composer_draft, dict) else {},
'process_wakeup_pause': self.process_wakeup_pause if isinstance(self.process_wakeup_pause, dict) else {},
'is_streaming': _is_streaming_session(
self.active_stream_id, active_stream_ids
) if include_runtime else False,
}
PROCESS_WAKEUP_PROVIDER_UNAVAILABLE_TYPES = frozenset({
'credential_pool_empty',
})
PROCESS_WAKEUP_PAUSE_ERROR = 'process_wakeup_paused'
_PROCESS_WAKEUP_PAUSE_VERSION = 1
def _process_wakeup_pause_part(value) -> str:
return str(value or '').strip().lower()
def _process_wakeup_pause_int(value, default: int = 0) -> int:
try:
parsed = int(value)
except (TypeError, ValueError):
return default
return parsed if parsed >= 0 else default
def _process_wakeup_pause_float(value, default: float) -> float:
try:
parsed = float(value)
except (TypeError, ValueError):
return default
return parsed if parsed > 0 else default
def _process_wakeup_pause_key(model=None, provider=None, classification=None) -> dict:
return {
'model': _process_wakeup_pause_part(model),
'provider': _process_wakeup_pause_part(provider),
'classification': _process_wakeup_pause_part(classification),
}
def process_wakeup_pause_matches(session, *, model=None, provider=None, classification=None) -> bool:
"""Return True when the session has an active pause for this wakeup lane."""
pause = getattr(session, 'process_wakeup_pause', None)
if not isinstance(pause, dict) or not pause.get('paused'):
return False
expected = _process_wakeup_pause_key(model, provider, classification)
for key, expected_value in expected.items():
if key == 'classification' and not expected_value:
continue
if _process_wakeup_pause_part(pause.get(key)) != expected_value:
return False
return True
def clear_process_wakeup_pause(session, *, reason: str = '') -> bool:
"""Clear any persisted process-wakeup pause metadata.
Returns True when a pause was present. Callers decide whether and how to
persist, because some paths are already inside a larger session writeback.
"""
pause = getattr(session, 'process_wakeup_pause', None)
if not isinstance(pause, dict) or not pause:
return False
session.process_wakeup_pause = {}
if reason:
session._last_process_wakeup_pause_clear_reason = str(reason)
return True
def clear_process_wakeup_pause_if_model_changed(session, *, model=None, provider=None) -> bool:
"""Reset a wakeup pause when the resolved model/provider lane changed."""
pause = getattr(session, 'process_wakeup_pause', None)
if not isinstance(pause, dict) or not pause.get('paused'):
return False
current = _process_wakeup_pause_key(model, provider, pause.get('classification'))
if (
_process_wakeup_pause_part(pause.get('model')) == current['model']
and _process_wakeup_pause_part(pause.get('provider')) == current['provider']
):
return False
return clear_process_wakeup_pause(session, reason='model_or_provider_changed')
def record_process_wakeup_provider_unavailable_pause(
session,
*,
classification: str,
model=None,
provider=None,
) -> dict | None:
"""Record the first visible provider-unavailable wakeup failure.
The persisted object is deliberately metadata-only: it records the lane and
counters, not the wakeup prompt or provider response body, so it remains
auditable without copying process output or credentials into diagnostics.
"""
classification = _process_wakeup_pause_part(classification)
if classification not in PROCESS_WAKEUP_PROVIDER_UNAVAILABLE_TYPES:
return None
now = time.time()
key = _process_wakeup_pause_key(model, provider, classification)
existing = getattr(session, 'process_wakeup_pause', None)
same_window = (
isinstance(existing, dict)
and existing.get('paused')
and _process_wakeup_pause_part(existing.get('model')) == key['model']
and _process_wakeup_pause_part(existing.get('provider')) == key['provider']
and _process_wakeup_pause_part(existing.get('classification')) == key['classification']
)
visible_error_count = 1
suppressed_count = 0
first_paused_at = now
if same_window:
first_paused_at = _process_wakeup_pause_float(existing.get('first_paused_at'), now)
visible_error_count = _process_wakeup_pause_int(
existing.get('visible_error_count'),
1,
) + 1
suppressed_count = _process_wakeup_pause_int(existing.get('suppressed_count'), 0)
session.process_wakeup_pause = {
'version': _PROCESS_WAKEUP_PAUSE_VERSION,
'paused': True,
'source': 'process_wakeup',
'classification': key['classification'],
'model': key['model'],
'provider': key['provider'],
'first_paused_at': first_paused_at,
'last_error_at': now,
'visible_error_count': visible_error_count,
'suppressed_count': suppressed_count,
}
return session.process_wakeup_pause
def suppress_process_wakeup_for_provider_pause(
session,
*,
model=None,
provider=None,
classification: str = 'credential_pool_empty',
) -> dict | None:
"""Increment suppression metadata if this automatic wakeup is paused."""
if not process_wakeup_pause_matches(
session,
model=model,
provider=provider,
classification=classification,
):
return None
pause = dict(getattr(session, 'process_wakeup_pause', {}) or {})
pause['suppressed_count'] = _process_wakeup_pause_int(pause.get('suppressed_count'), 0) + 1
pause['last_suppressed_at'] = time.time()
pause['last_suppressed_reason'] = 'provider_unavailable_pause'
session.process_wakeup_pause = pause
return pause
def _get_profile_home(profile) -> Path:
"""Resolve the hermes agent home directory for the given profile.
api/routes.py
+52 -1
View File
@@ -8671,6 +8671,9 @@ from api.models import (
_profile_has_user_projects,
is_cron_session,
is_safe_session_id,
PROCESS_WAKEUP_PAUSE_ERROR,
clear_process_wakeup_pause_if_model_changed,
suppress_process_wakeup_for_provider_pause,
)
@@ -19855,6 +19858,7 @@ def start_session_turn(
msg = str(message or "").strip()
if not msg:
return {"error": "message is required", "_status": 400}
turn_source = str(source or "process_wakeup").strip() or "process_wakeup"
try:
s = get_session(session_id)
except KeyError:
@@ -19883,6 +19887,53 @@ def start_session_turn(
profile_config=_pp_cfg,
prefer_cached_catalog=True,
)
if clear_process_wakeup_pause_if_model_changed(
s,
model=model,
provider=model_provider,
):
try:
s.save(touch_updated_at=False)
except Exception:
logger.debug(
"failed to persist process_wakeup pause reset for session %s",
session_id,
exc_info=True,
)
_paused_wakeup = None
if turn_source == "process_wakeup":
_paused_wakeup = suppress_process_wakeup_for_provider_pause(
s,
model=model,
provider=model_provider,
classification='credential_pool_empty',
)
if _paused_wakeup is not None:
try:
PENDING_BG_TASK_COMPLETIONS.discard(s.session_id)
except Exception:
logger.debug(
"failed to discard pending bg-task marker for paused wakeup %s",
session_id,
exc_info=True,
)
try:
s.save(touch_updated_at=False)
except Exception:
logger.debug(
"failed to persist process_wakeup suppression for session %s",
session_id,
exc_info=True,
)
return {
"error": PROCESS_WAKEUP_PAUSE_ERROR,
"message": (
"Automatic process wakeups are paused for this session because "
"the provider credential pool is unavailable."
),
"process_wakeup_pause": _paused_wakeup,
"_status": 409,
}
resp = _start_run(
s,
msg=msg,
@@ -19891,7 +19942,7 @@ def start_session_turn(
model=model,
model_provider=model_provider,
normalized_model=normalized_model,
source="process_wakeup",
source=turn_source,
route="start_session_turn",
)
api/streaming.py
+19
View File
@@ -53,7 +53,9 @@ from api.usage import prompt_cache_hit_percent
from api.models import (
_is_empty_partial_activity_message,
_evict_sessions_over_cap,
clear_process_wakeup_pause,
get_state_db_session_messages,
record_process_wakeup_provider_unavailable_pause,
reconciled_state_db_messages_for_session,
)
from api.session_ops import mark_session_title_generated, session_has_manual_title
@@ -8717,7 +8719,15 @@ def _run_agent_streaming(
_err_type,
_err_hint,
)
_pending_source = getattr(s, 'pending_user_source', None) or 'webui'
_materialize_pending_user_turn_before_error(s)
if _pending_source == 'process_wakeup':
record_process_wakeup_provider_unavailable_pause(
s,
classification=_err_type,
model=getattr(s, 'model', None) or resolved_model or model,
provider=getattr(s, 'model_provider', None) or resolved_provider,
)
s.active_stream_id = None
s.pending_user_message = None
s.pending_attachments = []
@@ -9188,6 +9198,7 @@ def _run_agent_streaming(
logger.debug("Failed to append cancelled turn journal event", exc_info=True)
put('cancel', _cancel_event_payload('Cancelled by user'))
return
clear_process_wakeup_pause(s, reason='run_completed')
with _stream_writeback_stage(_writeback_timings, "session_save"):
s.save()
if cancel_event.is_set():
@@ -9797,7 +9808,15 @@ def _run_agent_streaming(
)
return
_pending_source = getattr(s, 'pending_user_source', None) or 'webui'
_materialize_pending_user_turn_before_error(s)
if _pending_source == 'process_wakeup':
record_process_wakeup_provider_unavailable_pause(
s,
classification=_exc_type,
model=getattr(s, 'model', None) or resolved_model or model,
provider=getattr(s, 'model_provider', None) or resolved_provider,
)
s.active_stream_id = None
s.pending_user_message = None
s.pending_attachments = []
tests/test_issue3929_process_wakeup_pause.py
+291
View File
@@ -0,0 +1,291 @@
"""Regression coverage for #3929-D process-wakeup credential exhaustion pause."""
from __future__ import annotations
import queue
import sys
import types
from unittest import mock
import pytest
import api.config as config
import api.models as models
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 _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")
def _run_failing_process_wakeup(session: Session, tmp_path):
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=_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 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
context_before = list(saved.context_messages)
messages_before = list(saved.messages)
def _unexpected_start_run(*_args, **_kwargs):
raise AssertionError("paused process_wakeup must not start another provider call")
monkeypatch.setattr(routes, "_resolve_chat_workspace_with_recovery", lambda _s, _w: str(tmp_path))
monkeypatch.setattr(routes, "_read_profile_model_config", lambda _s, _p: (None, None, {}))
monkeypatch.setattr(
routes,
"_resolve_compatible_session_model_state",
lambda *_args, **_kwargs: ("test-model", "test-provider", False),
)
monkeypatch.setattr(routes, "_start_run", _unexpected_start_run)
response = routes.start_session_turn(
session.session_id,
"[IMPORTANT: Background process second completed.]",
source="process_wakeup",
)
assert response["_status"] == 409
assert response["error"] == PROCESS_WAKEUP_PAUSE_ERROR
saved_after = Session.load(session.session_id)
assert saved_after is not None
assert saved_after.messages == messages_before
assert saved_after.context_messages == context_before
assert saved_after.process_wakeup_pause["suppressed_count"] == 1
assert "last_suppressed_at" in saved_after.process_wakeup_pause
def test_process_wakeup_pause_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_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
tests/test_wakeup_defer_race.py
+60 -1
View File
@@ -482,7 +482,66 @@ def test_teardown_409_requeues_wakeup_so_it_is_not_lost(monkeypatch):
# --------------------------------------------------------------------------
# Test 7 — Greptile P1: multiple deferred wakeups in one teardown. Only the
# Test 7 — credential exhaustion pause: a paused wakeup is also a 409, but it
# must be treated as intentionally suppressed rather than an active-turn race.
# --------------------------------------------------------------------------
def test_paused_process_wakeup_409_does_not_requeue(monkeypatch):
"""A credential-exhaustion pause returns 409, but must not re-defer.
Ordinary 409s mean "another turn won the lock" and need redelivery. A
``process_wakeup_paused`` 409 means the wakeup was intentionally suppressed,
so re-queueing would recreate the same provider-unavailable loop.
"""
from api import background_process as bp, config as cfg
import api.routes as routes
_reset_cfg_state()
sid = "sess-paused-409"
holder = {"calls": [], "event": threading.Event(), "requeued": []}
def _paused_start_session_turn(session_id, message, *, source="process_wakeup"):
holder["calls"].append(
{"session_id": session_id, "message": message, "source": source}
)
holder["event"].set()
return {"_status": 409, "error": "process_wakeup_paused"}
def _unexpected_requeue(session_id, process_id, wakeup_prompt):
holder["requeued"].append(
{
"session_id": session_id,
"process_id": process_id,
"wakeup_prompt": wakeup_prompt,
}
)
try:
bp.record_deferred_wakeup(
sid,
"proc-paused-1",
"[IMPORTANT: Background process completed while credentials were unavailable.]",
)
monkeypatch.setattr(routes, "start_session_turn", _paused_start_session_turn)
monkeypatch.setattr(bp, "record_deferred_wakeup", _unexpected_requeue)
assert bp.drain_deferred_wakeups_for_session(sid) == 1
assert holder["event"].wait(timeout=1.0)
import time as _t
_t.sleep(0.1)
assert len(holder["calls"]) == 1
assert holder["requeued"] == []
with cfg.DEFERRED_PROCESS_WAKEUPS_LOCK:
assert sid not in cfg.DEFERRED_PROCESS_WAKEUPS
finally:
_reset_cfg_state()
# --------------------------------------------------------------------------
# Test 8 — Greptile P1: multiple deferred wakeups in one teardown. Only the
# FIRST starts a turn (the rest would 409 racing the per-session agent lock);
# entries 2..N must be re-deferred, not lost, and drain on later teardowns.
# --------------------------------------------------------------------------