mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-11-06 05:17:15 +00:00
Teknium
18d6041890
A kill that lands during the retry backoff (the attempt id is unchanged until the next attempt starts) hit an attempt the worker already returned from, yet counted toward the wedged-worker give-up: a healthy call was abandoned and its orphaned worker sent one more request. With the cap at the stale timeout, the stale and cap checks also counted the same silent window twice and the cap check abandoned again after the call ended, replacing the stale error with the cap copy. The worker now records the attempt it caught an error for; kills of that attempt are not counted and the cap does not cut it. A stale kill restarts the cap clock, and the cap check stops once the call is done.
303 lines
17 KiB
Python
303 lines
17 KiB
Python
"""Display and heartbeat phase of the request-local streaming monitor.
|
|
|
|
Also holds the stale-kill follow-up (re-arm, abandon a wedged worker, stop claiming liveness
|
|
for a wedged inline call) and the shared "no visible text was delivered" check
|
|
(``_forget_invisible_deltas``) used by the error and abandon paths.
|
|
"""
|
|
|
|
import contextlib
|
|
import logging
|
|
import time
|
|
from types import SimpleNamespace
|
|
from urllib.parse import urlsplit
|
|
|
|
from agent import chat_completion_wait_notice as wn
|
|
from agent.model_metadata import is_local_endpoint
|
|
from agent.proxy_bypass import is_loopback_host
|
|
from agent.sleep_clock import SleepWatch
|
|
from agent.stream_max_lifetime import StreamLifetime, StreamLifetimeExceeded, resolve_stream_max_lifetime
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Kills of one dispatched attempt whose worker never unwinds before the monitor gives up on it.
|
|
_STALE_ABANDON_KILLS = 3
|
|
|
|
|
|
class StreamingWaitMonitor:
|
|
# Stale-kill bookkeeping: consecutive kills of the SAME dispatched attempt, and the error
|
|
# that ends a call whose worker those kills could not unwind.
|
|
_last_killed_attempt = 0
|
|
_same_attempt_kills = 0
|
|
_abandon_error = None
|
|
# Inline call wedged past the limit: the stale-timer re-arm time after which activity touches
|
|
# are withheld (None = claim liveness as usual). A real chunk after it lifts the hold.
|
|
_liveness_withheld_at = None
|
|
_lifetime = StreamLifetime(0.0) # opt-in agent.stream_max_lifetime; real clock set when the monitor starts
|
|
_lifetime_kill = None # (attempt id, elapsed s) of the latest cap kill
|
|
_released_attempt = 0 # the attempt the worker's except path last caught (it let go of it)
|
|
|
|
def _claims_liveness(self) -> bool:
|
|
since = self._liveness_withheld_at
|
|
return since is None or self.last_chunk_time["t"] > since
|
|
|
|
def _rearm_or_abandon_after_kill(self, elapsed: float) -> None:
|
|
"""After a stale kill: re-arm the stale timer and announce the reconnect, unless the
|
|
kill bookkeeping gave up on a wedged attempt (see ``_note_attempt_kill``)."""
|
|
# Reset the timer so we don't kill repeatedly while the worker unwinds; the cap clock
|
|
# too, so a cap at the stale timeout does not cut (and count) the same window again.
|
|
self.last_chunk_time["t"] = self._lifetime.started = time.time()
|
|
if self._note_attempt_kill(elapsed, "stale"):
|
|
return
|
|
if self._claims_liveness(): # the reconnect notice touches activity too (_emit_wait_notice)
|
|
self.agent._emit_diagnostic_wait(f"⚠ no output from provider for {int(elapsed)}s — reconnecting...")
|
|
self.agent._touch_activity(f"stale stream detected after {int(elapsed)}s, reconnecting")
|
|
|
|
def _note_attempt_kill(self, elapsed: float, kind: str) -> bool:
|
|
"""Count a stale or ``stream_max_lifetime`` kill of the current attempt. Once the same
|
|
dispatched attempt has been killed ``_STALE_ABANDON_KILLS`` times without unwinding,
|
|
the request is parked where a socket shutdown cannot reach it, and re-killing it every
|
|
window would stall the turn forever: a worker thread is abandoned (returns True); an
|
|
inline call (cron, delegated children) runs on the caller's thread and cannot be, so
|
|
the monitor stops touching activity and the owner's inactivity watchdog ends it."""
|
|
with self.stream_attempt_lock:
|
|
attempt = int(self.stream_attempt_state["current"])
|
|
if attempt == self._released_attempt: # retry backoff: the worker is not stuck in it
|
|
return False
|
|
rekill = bool(attempt) and attempt == self._last_killed_attempt
|
|
self._last_killed_attempt = attempt
|
|
self._same_attempt_kills = self._same_attempt_kills + 1 if rekill else 1
|
|
if not rekill:
|
|
self._liveness_withheld_at = None
|
|
if not attempt or self._same_attempt_kills < _STALE_ABANDON_KILLS:
|
|
return False
|
|
if self.worker is not None:
|
|
self._abandon_wedged_worker(elapsed, attempt, kind)
|
|
return True
|
|
if self._liveness_withheld_at is None:
|
|
logger.error(
|
|
"Inline streaming call did not unwind after %d %s kills of attempt %s; no longer "
|
|
"reporting it as active, so the inactivity watchdog can end it. model=%s",
|
|
self._same_attempt_kills, kind, attempt, self.api_kwargs.get("model", "unknown"))
|
|
self._liveness_withheld_at = self.last_chunk_time["t"]
|
|
return False
|
|
|
|
def _forget_invisible_deltas(self, exc) -> bool:
|
|
"""Deltas fired but nothing visible reached a consumer (whitespace/think-only deltas,
|
|
or no display consumer at all) and no tool call is in flight: from the user's and the
|
|
model's point of view NOTHING was delivered. The "partial delivery" stub would be
|
|
EMPTY and the loop would ask the model to continue from nowhere, so it repeats the
|
|
lost step (#112419). Reset delivery so the failure is handled as undelivered (retry
|
|
the unchanged request) instead. Returns whether a partial tool call is in flight."""
|
|
tool_in_flight = bool(self.result.get("partial_tool_names")) or self.provider_tool_in_flight["yes"]
|
|
if self.deltas_were_sent["yes"] and not tool_in_flight and not self._visible_text_delivered():
|
|
logging.getLogger("agent.chat_completion_helpers").warning(
|
|
"Stream died after deltas but before any visible text was delivered (0 chars, "
|
|
"no tool call in flight); treating as an undelivered stream failure: %s", exc)
|
|
self._quiet(self.agent._reset_stream_delivery_tracking)
|
|
self.deltas_were_sent["yes"] = False
|
|
self.first_delta_fired["done"] = False
|
|
return tool_in_flight
|
|
|
|
def _abandon_wedged_worker(self, elapsed: float, attempt: int, kind: str) -> None:
|
|
"""Like the /stop path, leave the (daemon) worker behind and end this call with a
|
|
retryable timeout; ``run()`` discards anything the worker writes later."""
|
|
logger.error(
|
|
"Streaming worker did not unwind after %d %s kills of attempt %s (%.0fs); "
|
|
"abandoning the request. model=%s", self._same_attempt_kills, kind, attempt, elapsed,
|
|
self.api_kwargs.get("model", "unknown"))
|
|
self._request_cancelled["value"] = True
|
|
if kind == "stale":
|
|
self._abandon_error = TimeoutError(
|
|
f"Provider stream stalled: no chunks and the request worker did not unwind after "
|
|
f"{self._same_attempt_kills} stale-stream kills; request abandoned.")
|
|
else: # the copy names the setting, like a cap kill that did unwind
|
|
self._abandon_error = StreamLifetimeExceeded(elapsed, self._lifetime.cap)
|
|
self._call_done.set()
|
|
|
|
def _poll_local_load_notice(self, now: float) -> bool:
|
|
"""Managed local server: surface a cold model's weight-load progress
|
|
instead of the 60s neutral "waiting on <model>" notice. Polled ~1s only while no
|
|
REAL chunk arrived for 2s+ (never during healthy token flow); in-memory,
|
|
no network. True while loading = heartbeat liveness, skip the rest of
|
|
this iteration (the stale detector's local floor dwarfs any load)."""
|
|
from agent.chat_completion_local import _managed_local_load_notice
|
|
|
|
m = self._mon
|
|
if self._show_prompt_progress(now):
|
|
return True
|
|
if now - self.last_chunk_time["t"] < 2.0 or now - m.last_load_poll < 1.0:
|
|
return False
|
|
m.last_load_poll = now
|
|
_load_notice = _managed_local_load_notice(self.agent, self.api_kwargs)
|
|
if _load_notice is not None:
|
|
m.wait_notice_started_ts = None # The local loader now owns the display.
|
|
m.wait_notice.reset()
|
|
self.agent._emit_wait_notice(_load_notice)
|
|
self.agent._touch_activity("local model loading")
|
|
m.load_notice_shown, m.load_notice_misses, m.last_heartbeat = True, 0, now # loading IS liveness
|
|
return True
|
|
if m.load_notice_shown:
|
|
# One missed sample is routine (probe timeout under load); clearing on it strobed the line.
|
|
m.load_notice_misses += 1
|
|
if m.load_notice_misses >= 3:
|
|
m.load_notice_shown, m.load_notice_misses = False, 0
|
|
self.agent._emit_wait_notice("")
|
|
return False
|
|
|
|
def _show_prompt_progress(self, now: float) -> bool:
|
|
"""Prompt progress straight from the stream's own progress chunks (llama-server with
|
|
``return_progress``), at most once a second until the first token. Clears the line
|
|
when the first token arrives. True when it showed progress."""
|
|
from agent.local_prompt_progress import prompt_progress_percent
|
|
|
|
m = self._mon
|
|
state = getattr(self, "prompt_progress", None)
|
|
diag = getattr(getattr(self, "clients", None), "diag", None)
|
|
first_token = isinstance(diag, dict) and bool(diag.get("first_chunk_at"))
|
|
if not state or state["progress"] is None:
|
|
return False
|
|
if first_token:
|
|
if m.load_notice_shown:
|
|
m.load_notice_shown, m.load_notice_misses = False, 0
|
|
self.agent._emit_wait_notice("")
|
|
return False
|
|
if now - state["t"] > 60.0 or now - m.last_load_poll < 1.0:
|
|
return False
|
|
m.last_load_poll = now
|
|
percent = prompt_progress_percent(state["progress"])
|
|
m.wait_notice_started_ts = None # The prompt-progress line owns the display.
|
|
m.wait_notice.reset()
|
|
self.agent._emit_wait_notice(
|
|
f"⚙ processing prompt — {percent}%" if percent is not None else "⚙ processing prompt")
|
|
m.load_notice_shown, m.load_notice_misses, m.last_heartbeat = True, 0, now
|
|
return True
|
|
|
|
def _skip_sleep(self, slept: float) -> None:
|
|
"""The computer slept and froze this request with a server on the same machine: move the
|
|
stall clock past the sleep, so the request gets its full patience again after wake. Never
|
|
past now: a chunk can land between the sleep reading and this write."""
|
|
if slept <= 0:
|
|
return
|
|
now = time.time()
|
|
self.last_chunk_time["t"] = min(self.last_chunk_time["t"] + slept, now)
|
|
self._mon.last_heartbeat = min(self._mon.last_heartbeat + slept, now)
|
|
logger.info("Computer slept for %.0fs during a model request; the stall timer skips that time.", slept)
|
|
self.agent._touch_activity(f"resumed after the computer slept for {int(slept)}s")
|
|
|
|
def _heartbeat(self, waiting_secs: int) -> None:
|
|
"""Gateway inactivity heartbeat: the start-to-first-chunk gap (thinking,
|
|
local prefill) can exceed the gateway timeout. Withheld for a wedged inline call."""
|
|
if not self._claims_liveness():
|
|
return
|
|
if waiting_secs >= 60.0:
|
|
# No chunks for 60s+: say WHAT the wait is and WHEN recovery kicks in —
|
|
# once per silence, not every heartbeat (#92550).
|
|
stale = self._stream_stale_timeout
|
|
watchdog = ("stream stale", stale - waiting_secs) if stale is not None and stale != float("inf") else None
|
|
diag = getattr(getattr(self, "clients", None), "diag", None)
|
|
phase = "post_chunk" if isinstance(diag, dict) and diag.get("first_chunk_at") else "first_chunk"
|
|
if not self._mon.wait_notice.should_emit(phase, watchdog):
|
|
self.agent._touch_activity(f"waiting for stream response ({waiting_secs}s, {phase})")
|
|
return
|
|
self._mon.wait_notice_started_ts = self._mon.last_heartbeat
|
|
self.agent._emit_wait_notice(wn.wait_notice_text(
|
|
self.api_kwargs.get('model', 'the provider'), waiting_secs, phase, watchdog))
|
|
else:
|
|
# Chunks are flowing — keep the tracker fresh, leave the display alone.
|
|
self.agent._touch_activity(f"waiting for stream response ({waiting_secs}s, no chunks yet)")
|
|
|
|
def _check_stream_lifetime(self) -> None:
|
|
"""Opt-in ``agent.stream_max_lifetime``: a stream that keeps sending chunks but never
|
|
finishes never trips the stale check. Kill it like a stale kill (cancel the attempt,
|
|
abort the request-local client) so the retry loop or the partial-delivery continuation
|
|
takes over. Never counted in the cross-turn stale breaker: the provider WAS responding,
|
|
and a streak of cap kills would refuse the next healthy request with nothing sent."""
|
|
if not self._lifetime.cap or self._call_done.is_set(): # off (the default) or call over
|
|
return
|
|
with self.stream_attempt_lock:
|
|
attempt = int(self.stream_attempt_state["current"])
|
|
# A released attempt (retry backoff) has no stream left to cut.
|
|
elapsed = self._lifetime.exceeded(attempt) if attempt and attempt != self._released_attempt else None
|
|
if elapsed is None:
|
|
return
|
|
model = self.api_kwargs.get("model", "unknown")
|
|
logger.warning("Stream still running after %.0fs (stream_max_lifetime %.0fs) — killing connection. "
|
|
"model=%s", elapsed, self._lifetime.cap, model)
|
|
self.agent._buffer_diagnostic_status(
|
|
f"⚠️ Response still streaming after {int(elapsed)}s (stream_max_lifetime "
|
|
f"{int(self._lifetime.cap)}s, model: {model}). Aborting...")
|
|
killed = self._attempt_stream_response
|
|
self._lifetime_kill = (attempt, elapsed) # before the abort: the woken worker reads it at once
|
|
with contextlib.suppress(Exception):
|
|
self._cancel_current_stream_attempt("stream_max_lifetime_kill")
|
|
self.clients.close_once("stream_max_lifetime_kill")
|
|
self._shutdown_stale_attempt_socket(killed)
|
|
# Accounted, never bumped: a stale re-kill or /stop while it unwinds must not strike it either.
|
|
self._stale_counted_attempts.add(attempt)
|
|
self.last_chunk_time["t"] = time.time() # the stale timer restarts with the kill
|
|
# A worker the abort cannot unwind would be re-cut every window forever (the stale
|
|
# detector may be off): same give-up as a stale kill, without its reconnect notice.
|
|
self._note_attempt_kill(elapsed, "stream_max_lifetime")
|
|
|
|
def _as_lifetime_error(self, e: Exception) -> Exception:
|
|
"""The worker's transport error for an attempt the cap killed, replaced by one that names
|
|
``agent.stream_max_lifetime`` (the retry, exhausted and partial-delivery lines show it)."""
|
|
kill = self._lifetime_kill
|
|
if kill is None or isinstance(e, StreamLifetimeExceeded): # no cap kill: touch nothing else
|
|
return e
|
|
with self.stream_attempt_lock:
|
|
if kill[0] != int(self.stream_attempt_state["current"]):
|
|
return e
|
|
err = StreamLifetimeExceeded(kill[1], self._lifetime.cap)
|
|
err.__cause__ = e
|
|
return err
|
|
|
|
def _monitor_loop(self) -> None:
|
|
try:
|
|
self._poll_until_done()
|
|
except Exception:
|
|
# A dying watchdog otherwise ends stale/cap/interrupt monitoring with no trace.
|
|
logger.error("stream watchdog monitor loop aborted abnormally", exc_info=True)
|
|
raise
|
|
|
|
def _poll_until_done(self) -> None:
|
|
_HEARTBEAT_INTERVAL = 30.0 # seconds between gateway activity touches
|
|
self._mon = SimpleNamespace(
|
|
last_heartbeat=time.time(), last_load_poll=0.0,
|
|
load_notice_shown=False, load_notice_misses=0, wait_notice_started_ts=None,
|
|
wait_notice=wn.WaitNoticeState(),
|
|
)
|
|
_is_local_base = bool(self.agent.base_url) and is_local_endpoint(self.agent.base_url)
|
|
# Only a server on this machine sleeps with it; a remote one kept working (or died).
|
|
sleep_watch = (SleepWatch(interval=0.3) if self.agent.base_url
|
|
and is_loopback_host(urlsplit(self.agent.base_url).hostname) else None)
|
|
self._lifetime = StreamLifetime(resolve_stream_max_lifetime(self._stream_stale_timeout))
|
|
logger.debug("stream watchdog armed: stale=%.0fs cap=%.0fs model=%s", self._stream_stale_timeout,
|
|
self._lifetime.cap, self.api_kwargs.get("model", "unknown"))
|
|
while not self._call_done.is_set():
|
|
self._call_done.wait(timeout=0.3)
|
|
if sleep_watch is not None:
|
|
self._skip_sleep(sleep_watch.slept())
|
|
_hb_now = time.time()
|
|
if _is_local_base and self._poll_local_load_notice(_hb_now):
|
|
continue
|
|
# Reasoning callbacks do not clear the classic CLI spinner. The empty
|
|
# protocol payload resets status without adding synthetic reasoning.
|
|
if (self._mon.wait_notice_started_ts is not None
|
|
and self.last_chunk_time["t"] > self._mon.wait_notice_started_ts):
|
|
self.agent._emit_wait_notice("")
|
|
self._mon.wait_notice_started_ts = None
|
|
self._mon.wait_notice.reset()
|
|
if _hb_now - self._mon.last_heartbeat >= _HEARTBEAT_INTERVAL:
|
|
self._mon.last_heartbeat = _hb_now
|
|
self._heartbeat(int(_hb_now - self.last_chunk_time["t"]))
|
|
_stale_elapsed = time.time() - self.last_chunk_time["t"]
|
|
if _stale_elapsed > self._stream_stale_timeout:
|
|
self._mon.wait_notice_started_ts = None # Reconnect status has its own owner.
|
|
self._mon.wait_notice.reset()
|
|
self._kill_stale_stream(_stale_elapsed)
|
|
self._check_stream_lifetime()
|
|
if self.agent._interrupt_requested:
|
|
self._abort_for_interrupt(_stale_elapsed)
|
|
return
|
|
|