From bcb411db53c1283cc1b1a196d2ffacf281436e02 Mon Sep 17 00:00:00 2001 From: Frank Song Date: Tue, 7 Jul 2026 11:45:01 +0800 Subject: [PATCH] fix(wakeup): pause credential-exhausted process wakeups --- api/background_process.py | 7 +- api/models.py | 156 ++++++++++ api/routes.py | 53 +++- api/streaming.py | 19 ++ tests/test_issue3929_process_wakeup_pause.py | 291 +++++++++++++++++++ tests/test_wakeup_defer_race.py | 61 +++- 6 files changed, 584 insertions(+), 3 deletions(-) create mode 100644 tests/test_issue3929_process_wakeup_pause.py diff --git a/api/background_process.py b/api/background_process.py index cffaa224..8692737f 100644 --- a/api/background_process.py +++ b/api/background_process.py @@ -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 diff --git a/api/models.py b/api/models.py index 9f945693..ffa3c1a8 100644 --- a/api/models.py +++ b/api/models.py @@ -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. diff --git a/api/routes.py b/api/routes.py index 3e5ae1bc..450b202e 100644 --- a/api/routes.py +++ b/api/routes.py @@ -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", ) diff --git a/api/streaming.py b/api/streaming.py index 4bc49fe0..4e4ff711 100644 --- a/api/streaming.py +++ b/api/streaming.py @@ -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 = [] diff --git a/tests/test_issue3929_process_wakeup_pause.py b/tests/test_issue3929_process_wakeup_pause.py new file mode 100644 index 00000000..1a7e953c --- /dev/null +++ b/tests/test_issue3929_process_wakeup_pause.py @@ -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 diff --git a/tests/test_wakeup_defer_race.py b/tests/test_wakeup_defer_race.py index 848833a2..7d3ce27d 100644 --- a/tests/test_wakeup_defer_race.py +++ b/tests/test_wakeup_defer_race.py @@ -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. # --------------------------------------------------------------------------