From 60c8fc6290a080b530488d63eb118f406e4dff49 Mon Sep 17 00:00:00 2001 From: teknium1 <127238744+teknium1@users.noreply.github.com> Date: Fri, 24 Jul 2026 19:05:29 -0700 Subject: [PATCH] fix(gateway): deliver the first message when the deferred agent build outlives 30s (#63078) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Leg 2 of #63078: prompt.submit returns {"status":"streaming"} immediately and runs _start_agent_build + _wait_agent(timeout=30s) behind it. The deferred build (MCP discovery with per-server retry backoff, synchronous model-metadata HTTP, skills scanning) routinely outlives 30s on cold starts; on timeout run_after_agent_ready emitted an error EVENT and returned without ever calling _run_prompt_submit — the user's first message was permanently discarded while the build finished successfully in the background. The desktop's optimistic row eventually cleared with no visible error: the blank first session. New _wait_agent_for_prompt replaces the flat cliff for the deferred prompt path only (_sess()'s RPC-blocking _wait_agent keeps its 30s contract): - The pending prompt stays attached to the (already off-RPC) run thread and is delivered the moment the still-running build completes — a slow build is no longer message loss. - The wait runs in 5s slices so a cancel (session.interrupt / churn) is honored promptly; the cancelled path returns None and defers to the caller's cancel branch (the #65567 emit) for user-visible messaging. - Past 30s the client gets ONE keyed notification.show ('Still starting the agent…', key=agent-build-slow, desktop toast / TUI status bar), cleared on delivery — patient, never silent. - Permanent failure only when the build itself fails: agent_error set at ready, the build thread died without signalling ready (fail fast via the new _agent_build_thread handle instead of sitting out the cap on a corpse), or the bounded cap expired on a genuinely hung build. The cap defaults to 600s and is tunable via agent.build_wait_timeout in config.yaml (no new env vars); the error message states the message was not sent. Tests: slow-build delivery with zero error events; the keyed progress notice shown once and cleared; build-failure surfacing exactly one error event with the real reason; dead-thread fail-fast; cancel honored mid-wait; config override + fallback semantics; cap expiry message. The compute-host fallback test stubs the new waiter alongside _wait_agent. --- cli-config.yaml.example | 8 + hermes_cli/config.py | 8 + tests/test_tui_gateway_server.py | 325 +++++++++++++++++++++++++++++++ tui_gateway/server.py | 132 ++++++++++++- 4 files changed, 471 insertions(+), 2 deletions(-) diff --git a/cli-config.yaml.example b/cli-config.yaml.example index d52e3ec009c1..3beb568cab7a 100644 --- a/cli-config.yaml.example +++ b/cli-config.yaml.example @@ -799,6 +799,14 @@ agent: # window on /restart, and keep it well under systemd's TimeoutStopSec. # restart_drain_timeout: 0 + # Upper bound (seconds) a submitted prompt waits for the deferred agent + # build (MCP discovery, model metadata, skills scan) before failing with a + # visible error. The wait is patient — the message is delivered as soon as + # the build completes, and a progress notice is shown past 30s — so this cap + # only fires on a genuinely hung build. Raise it for deployments with many + # slow or unreachable MCP servers. Default 600. + # build_wait_timeout: 600 + # Max app-level retry attempts for API errors (connection drops, provider # timeouts, 5xx, etc.) before the agent surfaces the failure. Lower this # to 1 if you use fallback providers and want fast failover on flaky diff --git a/hermes_cli/config.py b/hermes_cli/config.py index cc03c5a57b5c..9874a37ec395 100644 --- a/hermes_cli/config.py +++ b/hermes_cli/config.py @@ -966,6 +966,14 @@ DEFAULT_CONFIG = { # Set a positive value in config.yaml only if you explicitly want a # grace window on /restart (and keep it well under TimeoutStopSec). "restart_drain_timeout": 0, + # Upper bound (seconds) a submitted prompt waits for the deferred + # agent build (MCP discovery, model metadata, skills scan) before + # failing with a visible error (#63078). The gateway's wait is + # patient — the prompt is delivered the moment the build completes + # and a progress notice is emitted past 30s — so this cap only fires + # on a genuinely hung build. Raise it for deployments with many slow + # or unreachable MCP servers. + "build_wait_timeout": 600, # Max app-level retry attempts for API errors (connection drops, # provider timeouts, 5xx, etc.) before the agent surfaces the # failure. The OpenAI SDK already does its own low-level retries diff --git a/tests/test_tui_gateway_server.py b/tests/test_tui_gateway_server.py index 8cc002c8c5fb..f851704e8100 100644 --- a/tests/test_tui_gateway_server.py +++ b/tests/test_tui_gateway_server.py @@ -293,6 +293,8 @@ def test_prompt_submit_fails_open_inline_when_compute_host_dispatch_breaks(monke monkeypatch.setattr(server, "_persist_branch_seed", lambda _session: None) monkeypatch.setattr(server, "_start_agent_build", lambda _sid, _session: None) monkeypatch.setattr(server, "_wait_agent", lambda _session, _rid: None) + # The deferred inline-fallback thread now waits via the patient variant. + monkeypatch.setattr(server, "_wait_agent_for_prompt", lambda _session, _rid, _sid: None) monkeypatch.setattr( server, "_run_prompt_submit", @@ -8101,6 +8103,329 @@ def test_session_not_running_before_agent_ready_emits_error_event(monkeypatch): server._sessions.pop("sid", None) +def test_slow_agent_build_delivers_prompt_instead_of_timing_out(monkeypatch): + """#63078 server-side half: a deferred build slower than the old 30s + ``_wait_agent`` cliff must NOT eat the first message. The patient wait + keeps the pending prompt attached and delivers it as soon as the + still-running build completes.""" + threads = [] + emitted = [] + calls = {"run_prompt": 0} + + class _FakeThread: + def __init__(self, target=None, daemon=None): + self.target = target + threads.append(self) + + def start(self): + return None + + def is_alive(self): + return True + + ready = threading.Event() + session = _session(agent_ready=ready) + session["agent"] = None + server._sessions["sid"] = session + + # The build "completes" only after the wait loop has already gone through + # several empty slices — i.e. well past what a single fixed-timeout wait + # slice would tolerate. + slices = {"n": 0} + + class _SlowReady: + def wait(self, timeout=None): + slices["n"] += 1 + if slices["n"] >= 3: + ready.set() + session["agent"] = types.SimpleNamespace() + return True + return False + + def is_set(self): + return ready.is_set() + + session["agent_ready"] = _SlowReady() + + try: + monkeypatch.setattr(server.threading, "Thread", _FakeThread) + monkeypatch.setattr(server, "_emit", lambda *args, **kwargs: emitted.append(args)) + monkeypatch.setattr(server, "_ensure_session_db_row", lambda session: None) + monkeypatch.setattr(server, "_persist_branch_seed", lambda session: None) + monkeypatch.setattr(server, "_start_agent_build", lambda sid, session: None) + monkeypatch.setattr( + server, + "_run_prompt_submit", + lambda *args, **kwargs: calls.__setitem__( + "run_prompt", calls["run_prompt"] + 1 + ), + ) + + submit = server.handle_request( + { + "id": "1", + "method": "prompt.submit", + "params": {"session_id": "sid", "text": "first message"}, + } + ) + assert submit.get("result"), f"got error: {submit.get('error')}" + + threads[0].target() + + # The message was DELIVERED, not dropped, and no error event fired. + assert calls["run_prompt"] == 1 + error_events = [e for e in emitted if e and e[0] == "error"] + assert not error_events, f"unexpected error events: {error_events}" + finally: + server._sessions.pop("sid", None) + + +def test_slow_agent_build_emits_keyed_progress_notice(monkeypatch): + """Past the slow threshold the patient wait must tell the user once + (keyed notification.show) and clear the notice when the build lands — + a long wait is acceptable, a silent one is not.""" + threads = [] + emitted = [] + calls = {"run_prompt": 0} + + class _FakeThread: + def __init__(self, target=None, daemon=None): + self.target = target + threads.append(self) + + def start(self): + return None + + def is_alive(self): + return True + + ready = threading.Event() + session = _session(agent_ready=ready) + session["agent"] = None + server._sessions["sid"] = session + + slices = {"n": 0} + + class _SlowReady: + def wait(self, timeout=None): + slices["n"] += 1 + if slices["n"] >= 3: + ready.set() + session["agent"] = types.SimpleNamespace() + return True + return False + + def is_set(self): + return ready.is_set() + + session["agent_ready"] = _SlowReady() + + try: + monkeypatch.setattr(server.threading, "Thread", _FakeThread) + # Every wait slice lands past the slow threshold. + monkeypatch.setattr(server, "_AGENT_BUILD_SLOW_NOTICE_AFTER", 0.0) + monkeypatch.setattr(server, "_emit", lambda *args, **kwargs: emitted.append(args)) + monkeypatch.setattr(server, "_ensure_session_db_row", lambda session: None) + monkeypatch.setattr(server, "_persist_branch_seed", lambda session: None) + monkeypatch.setattr(server, "_start_agent_build", lambda sid, session: None) + monkeypatch.setattr( + server, + "_run_prompt_submit", + lambda *args, **kwargs: calls.__setitem__( + "run_prompt", calls["run_prompt"] + 1 + ), + ) + + submit = server.handle_request( + { + "id": "1", + "method": "prompt.submit", + "params": {"session_id": "sid", "text": "first message"}, + } + ) + assert submit.get("result"), f"got error: {submit.get('error')}" + + threads[0].target() + + assert calls["run_prompt"] == 1 + shows = [e for e in emitted if e and e[0] == "notification.show" and e[1] == "sid"] + clears = [e for e in emitted if e and e[0] == "notification.clear" and e[1] == "sid"] + # Exactly one keyed notice, replaced-in-place semantics, then cleared. + assert len(shows) == 1, f"expected one slow-build notice, got: {shows}" + assert shows[0][2].get("key") == server._AGENT_BUILD_SLOW_NOTICE_KEY + assert len(clears) == 1 and clears[0][2].get("key") == server._AGENT_BUILD_SLOW_NOTICE_KEY + finally: + server._sessions.pop("sid", None) + + +def test_agent_build_failure_surfaces_error_and_drops_turn(monkeypatch): + """When the build itself FAILS (agent_error set when ready fires), the + prompt must not run and the failure must reach the client as a visible + error event — never a silent drop.""" + threads = [] + emitted = [] + calls = {"run_prompt": 0} + + class _FakeThread: + def __init__(self, target=None, daemon=None): + self.target = target + threads.append(self) + + def start(self): + return None + + def is_alive(self): + return True + + ready = threading.Event() + ready.set() # build finished... + session = _session(agent_ready=ready) + session["agent"] = None + session["agent_error"] = "No LLM provider configured" # ...but failed + server._sessions["sid"] = session + + try: + monkeypatch.setattr(server.threading, "Thread", _FakeThread) + monkeypatch.setattr(server, "_emit", lambda *args, **kwargs: emitted.append(args)) + monkeypatch.setattr(server, "_ensure_session_db_row", lambda session: None) + monkeypatch.setattr(server, "_persist_branch_seed", lambda session: None) + monkeypatch.setattr(server, "_start_agent_build", lambda sid, session: None) + monkeypatch.setattr( + server, + "_run_prompt_submit", + lambda *args, **kwargs: calls.__setitem__( + "run_prompt", calls["run_prompt"] + 1 + ), + ) + + submit = server.handle_request( + { + "id": "1", + "method": "prompt.submit", + "params": {"session_id": "sid", "text": "first message"}, + } + ) + assert submit.get("result"), f"got error: {submit.get('error')}" + + threads[0].target() + + assert calls["run_prompt"] == 0 + assert session["running"] is False + error_events = [e for e in emitted if e and e[0] == "error" and e[1] == "sid"] + assert len(error_events) == 1, f"expected one error event, got: {emitted}" + assert "No LLM provider configured" in error_events[0][2].get("message", "") + finally: + server._sessions.pop("sid", None) + + +def test_dead_build_thread_fails_fast_not_full_cap(monkeypatch): + """A build thread that died without setting agent_ready means the build + died hard — the waiter must fail promptly with a visible error instead of + sitting out the full wait cap on a corpse.""" + emitted = [] + + class _DeadThread: + def is_alive(self): + return False + + ready = threading.Event() # never set + session = _session(agent_ready=ready) + session["agent"] = None + session["running"] = True + session["_agent_build_thread"] = _DeadThread() + session["agent_error"] = "agent init failed: boom" + server._sessions["sid"] = session + + try: + monkeypatch.setattr(server, "_emit", lambda *args, **kwargs: emitted.append(args)) + # Short slices so the test is fast; the dead-thread check fires on the + # first empty slice, far below the cap. + monkeypatch.setattr(server, "_AGENT_BUILD_WAIT_SLICE", 0.01) + + start = time.monotonic() + err = server._wait_agent_for_prompt(session, "rid-1", "sid") + elapsed = time.monotonic() - start + + assert err is not None + assert "boom" in (err.get("error") or {}).get("message", "") + assert elapsed < 5.0, f"dead-thread detection took {elapsed:.1f}s" + finally: + server._sessions.pop("sid", None) + + +def test_wait_agent_for_prompt_honors_cancel_mid_wait(monkeypatch): + """A cancel arriving during the patient wait must end it promptly and + return None (the caller's cancel branch owns the user-visible event).""" + ready = threading.Event() # never set + session = _session(agent_ready=ready) + session["agent"] = None + session["running"] = True + server._sessions["sid"] = session + + try: + monkeypatch.setattr(server, "_AGENT_BUILD_WAIT_SLICE", 0.01) + + def cancel_soon(): + time.sleep(0.05) + with session["history_lock"]: + session["_turn_cancel_requested"] = True + + canceller = threading.Thread(target=cancel_soon) + canceller.start() + start = time.monotonic() + err = server._wait_agent_for_prompt(session, "rid-1", "sid") + elapsed = time.monotonic() - start + canceller.join() + + assert err is None + assert elapsed < 5.0, f"cancel honored only after {elapsed:.1f}s" + finally: + server._sessions.pop("sid", None) + + +def test_agent_build_wait_cap_config_override(monkeypatch): + """agent.build_wait_timeout in config.yaml overrides the default cap; + invalid/absent values fall back to 600s.""" + monkeypatch.setattr(server, "_load_cfg", lambda: {"agent": {"build_wait_timeout": 90}}) + assert server._agent_build_wait_cap() == 90.0 + + monkeypatch.setattr(server, "_load_cfg", lambda: {"agent": {}}) + assert server._agent_build_wait_cap() == 600.0 + + monkeypatch.setattr(server, "_load_cfg", lambda: {"agent": {"build_wait_timeout": 0}}) + assert server._agent_build_wait_cap() == 600.0 + + monkeypatch.setattr(server, "_load_cfg", lambda: {"agent": {"build_wait_timeout": "nonsense"}}) + assert server._agent_build_wait_cap() == 600.0 + + +def test_wait_agent_for_prompt_expires_at_cap(monkeypatch): + """A genuinely hung build (thread alive, never ready) still fails at the + bounded cap with a message that tells the user their text was not sent.""" + class _AliveThread: + def is_alive(self): + return True + + ready = threading.Event() # never set + session = _session(agent_ready=ready) + session["agent"] = None + session["running"] = True + session["_agent_build_thread"] = _AliveThread() + server._sessions["sid"] = session + + try: + monkeypatch.setattr(server, "_AGENT_BUILD_WAIT_SLICE", 0.01) + monkeypatch.setattr(server, "_agent_build_wait_cap", lambda: 0.05) + + err = server._wait_agent_for_prompt(session, "rid-1", "sid") + + assert err is not None + message = (err.get("error") or {}).get("message", "") + assert "timed out" in message and "was not sent" in message + finally: + server._sessions.pop("sid", None) + + def test_clear_pending_without_sid_clears_all(): """_clear_pending(None) is the shutdown path — must still release every pending prompt regardless of owning session.""" diff --git a/tui_gateway/server.py b/tui_gateway/server.py index 433439de8cdc..c9237a6b28ee 100644 --- a/tui_gateway/server.py +++ b/tui_gateway/server.py @@ -1670,6 +1670,124 @@ def _wait_agent(session: dict, rid: str, timeout: float = 30.0) -> dict | None: return _err(rid, 5032, err) if err else None +# The deferred prompt path waits in short slices so a cancel is honored +# promptly and a slow build can be reported to the client exactly once. +_AGENT_BUILD_WAIT_SLICE = 5.0 +_AGENT_BUILD_SLOW_NOTICE_AFTER = 30.0 +_AGENT_BUILD_SLOW_NOTICE_KEY = "agent-build-slow" + + +def _agent_build_wait_cap() -> float: + """Upper bound (seconds) a submitted prompt waits for the deferred agent + build before failing permanently. ``agent.build_wait_timeout`` in + config.yaml overrides the 600s default (raise it for deployments with + many slow/unreachable MCP servers or high-latency provider metadata).""" + try: + agent_cfg = _load_cfg().get("agent") or {} + raw = agent_cfg.get("build_wait_timeout") + if raw is not None: + value = float(raw) + if value > 0: + return value + except Exception: + pass + return 600.0 + + +def _wait_agent_for_prompt(session: dict, rid: str, sid: str) -> dict | None: + """Patient variant of ``_wait_agent`` for the deferred prompt.submit path. + + The flat 30s ``_wait_agent`` ceiling was a message-eating cliff (#63078): + ``prompt.submit`` has already returned ``{"status": "streaming"}``, the + user's first message IS the turn in flight, and the deferred agent build + (MCP discovery with per-server retry backoff, synchronous model-metadata + HTTP, skills scanning) routinely outlives 30 seconds on cold starts. On + timeout the old path emitted an error EVENT and returned without ever + calling ``_run_prompt_submit`` — the first message was permanently + discarded while the build finished successfully in the background, leaving + the blank first session. + + This wait instead: + - keeps the pending prompt attached to this (already off-RPC) thread and + delivers it the moment the still-running build completes; + - waits in short slices so a cancel (session.interrupt / session churn) + is honored promptly instead of after the full timeout; + - tells the client once, via a keyed notice, when the build outlives + ``_AGENT_BUILD_SLOW_NOTICE_AFTER`` — the wait is patient but never + silent; + - fails permanently only when the build itself fails: the build thread + died without signalling ready, or the bounded cap + (``agent.build_wait_timeout``, default 600s — no infinite waits) + expired on a genuinely hung build. + + Returns ``None`` on success OR when the turn was cancelled mid-wait (the + caller's cancel branch owns that messaging), an ``_err`` dict otherwise. + """ + ready = session.get("agent_ready") + if ready is None: + return None + start = time.monotonic() + cap = _agent_build_wait_cap() + notified_slow = False + while not ready.wait(timeout=_AGENT_BUILD_WAIT_SLICE): + with session["history_lock"]: + cancelled = session.get("_turn_cancel_requested") or not session.get( + "running" + ) + if cancelled: + # The caller's cancel/not-running branch emits the user-visible + # event for this — bail without an error of our own. + return None + waited = time.monotonic() - start + if waited >= cap: + return _err( + rid, + 5032, + f"agent initialization timed out after {int(waited)}s — " + "your message was not sent; retry once the session is ready", + ) + build_thread = session.get("_agent_build_thread") + if ( + build_thread is not None + and not build_thread.is_alive() + and not ready.is_set() + ): + # _build's ``finally`` guarantees ready.set(); a dead thread with + # ready still unset means the build died hard (interpreter-level + # kill) — don't wait on a corpse for the rest of the cap. + return _err( + rid, + 5032, + session.get("agent_error") + or "agent initialization failed before completing", + ) + if not notified_slow and waited >= _AGENT_BUILD_SLOW_NOTICE_AFTER: + # One keyed, replace-in-place notice: the desktop shows it as a + # toast, the TUI in its status bar. Without this the extended wait + # would be exactly the silent hang this function exists to fix. + notified_slow = True + _emit( + "notification.show", + sid, + { + "text": ( + "Still starting the agent (tool discovery / model " + "setup) — your message will be sent as soon as it's " + "ready." + ), + "level": "info", + "kind": "agent", + "ttl_ms": None, + "key": _AGENT_BUILD_SLOW_NOTICE_KEY, + "id": _AGENT_BUILD_SLOW_NOTICE_KEY, + }, + ) + if notified_slow: + _emit("notification.clear", sid, {"key": _AGENT_BUILD_SLOW_NOTICE_KEY}) + err = session.get("agent_error") + return _err(rid, 5032, err) if err else None + + def _start_agent_build(sid: str, session: dict) -> None: """Start building the real AIAgent for a TUI session, once. @@ -1846,7 +1964,12 @@ def _start_agent_build(sid: str, session: dict) -> None: pass ready.set() - threading.Thread(target=_build, daemon=True).start() + build_thread = threading.Thread(target=_build, daemon=True) + # Handle for _wait_agent_for_prompt: a dead build thread with agent_ready + # still unset means the build died hard — waiters must not sit out the + # full cap on a corpse. + session["_agent_build_thread"] = build_thread + build_thread.start() def _sess_nowait(params, rid): @@ -10338,7 +10461,12 @@ def _(rid, params: dict) -> dict: _start_agent_build(sid, session) def run_after_agent_ready() -> None: - err = _wait_agent(session, rid) + # Patient wait (#63078): the user's message is already the accepted + # in-flight turn, so a slow deferred build must not eat it. The wait + # delivers the prompt when the still-running build completes, honors a + # cancel promptly, notices the user once past the slow threshold, and + # only errors when the build itself fails or the bounded cap expires. + err = _wait_agent_for_prompt(session, rid, sid) if err: _emit( "error",