fix(agent): delegated children and cron turns stream again — inline, no worker
should_use_direct_api_call() contexts (gateway cron turns #62151, delegate_task children #60203) were short-circuited onto the NON-streaming wire because the interrupt worker wedges inside their nested thread pools. That dropped every liveness property streaming provides: edge proxies kill the silent POST (z.ai HTTP 524 — three retries later the child dies as "max_iterations"), and the non-stream stale watchdog cannot tell a reasoning model's thinking phase from a hung provider, so children die at exactly stale_timeout (#100260). Keep those contexts on interruptible_streaming_api_call. The request now runs INLINE on the conversation thread (no worker → the deadlock class stays closed) while the existing poll loop — 30s heartbeat, stale-stream detector, cross-thread interrupt abort — moves onto a monitor thread that only ever aborts sockets, never dispatches (same shape as direct_api_call's watchdog timer). Interactive sessions are unchanged: worker + poll loop as before. should_use_direct_api_call() itself is untouched; only what it routes to. Live A/B (real SSE server, real AIAgent.run_conversation): before: subagent/cron wire stream=None, request on conversation thread after: subagent/cron wire stream=True, request on conversation thread cli unchanged (stream=True, spawned worker) inline stale detector kills a one-chunk-then-silence stream at budget; AIAgent.interrupt() from another thread unwinds the inline stream in 0.6s. Co-authored-by: Expri-commits <184641533+Expri-commits@users.noreply.github.com>
This commit is contained in:
Teknium
+223
-167
@@ -3546,14 +3546,11 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta=
|
||||
if emit is not None:
|
||||
emit(final_text=final_text, finished=finished, error=error)
|
||||
|
||||
# Cron and other non-interactive, nested-pool contexts deadlock on the
|
||||
# spawned worker thread (#62151). They also have no stream consumer, so the
|
||||
# deltas this path produces go nowhere. Delegate to the non-streaming entry
|
||||
# (which runs inline via should_use_direct_api_call) exactly like the codex
|
||||
# branch below — routing through the _interruptible_api_call method keeps the
|
||||
# outer loop's per-request retry/refresh seam intact.
|
||||
if should_use_direct_api_call(agent):
|
||||
return agent._interruptible_api_call(api_kwargs)
|
||||
# Cron turns and delegated children (should_use_direct_api_call) used to be
|
||||
# short-circuited here onto the NON-streaming wire. They now stay on this
|
||||
# streaming path and run the request inline — see the ``_inline`` block
|
||||
# before the poll loop below. Only the codex branch still detours through
|
||||
# _interruptible_api_call (it streams internally).
|
||||
|
||||
if agent.api_mode == "codex_responses":
|
||||
# Codex streams internally via _run_codex_stream. The main dispatch
|
||||
@@ -5371,8 +5368,42 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta=
|
||||
if _reasoning_floor is not None:
|
||||
_stream_stale_timeout = max(_stream_stale_timeout, _reasoning_floor)
|
||||
|
||||
t = threading.Thread(target=_context_thread_target(_call), daemon=True)
|
||||
t.start()
|
||||
# Delegated children and gateway cron turns run the streaming request
|
||||
# INLINE on the conversation thread: spawning the interrupt worker inside
|
||||
# their nested thread pools wedges before the socket opens (#62151,
|
||||
# #60203). They used to be routed to the non-streaming wire for that
|
||||
# reason — but streaming is also the transport keepalive and the
|
||||
# liveness signal: a non-streaming POST that stays silent through a
|
||||
# reasoning model's thinking phase is killed by edge proxies (z.ai 524,
|
||||
# #90202) and by our own stale watchdog, which cannot tell thinking from
|
||||
# a hang when no bytes ever arrive (#100260). Inline mode keeps the
|
||||
# stream (per-token liveness) and moves ONLY the lightweight poll loop
|
||||
# below — heartbeat, stale detector, interrupt abort — onto a monitor
|
||||
# thread. The monitor never issues a request, so the no-worker property
|
||||
# that fixes the deadlock class is preserved (same shape as the
|
||||
# direct_api_call watchdog timer).
|
||||
_inline = should_use_direct_api_call(agent)
|
||||
_call_done = threading.Event()
|
||||
_monitor_interrupted = {"yes": False}
|
||||
|
||||
def _run_call():
|
||||
try:
|
||||
_call()
|
||||
finally:
|
||||
_call_done.set()
|
||||
|
||||
if _inline:
|
||||
t = None
|
||||
else:
|
||||
t = threading.Thread(target=_context_thread_target(_run_call), daemon=True)
|
||||
t.start()
|
||||
|
||||
def _call_alive() -> bool:
|
||||
return not _call_done.is_set()
|
||||
|
||||
def _wait_call(timeout: float) -> None:
|
||||
_call_done.wait(timeout=timeout)
|
||||
|
||||
_last_heartbeat = time.time()
|
||||
_HEARTBEAT_INTERVAL = 30.0 # seconds between gateway activity touches
|
||||
# Managed local server: a cold model streams weights off disk for tens
|
||||
@@ -5385,173 +5416,198 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta=
|
||||
_load_notice_shown = False
|
||||
_load_notice_misses = 0
|
||||
_is_local_base = bool(agent.base_url) and is_local_endpoint(agent.base_url)
|
||||
while t.is_alive():
|
||||
t.join(timeout=0.3)
|
||||
|
||||
_hb_now = time.time()
|
||||
# Cold-load window: last_chunk_time is touched at request-client
|
||||
# creation and then only by REAL chunks, so "no chunk for 2s+" is
|
||||
# true through a model load (nothing can stream while the child is
|
||||
# still mapping weights) and false during healthy token flow —
|
||||
# which is what keeps this poll off the streaming hot path. The
|
||||
# probe itself is an in-memory snapshot read.
|
||||
if (
|
||||
_is_local_base
|
||||
and _hb_now - last_chunk_time["t"] >= 2.0
|
||||
and _hb_now - _last_load_poll >= 1.0
|
||||
):
|
||||
_last_load_poll = _hb_now
|
||||
_load_notice = _managed_local_load_notice(agent, api_kwargs)
|
||||
if _load_notice is not None:
|
||||
agent._emit_wait_notice(_load_notice)
|
||||
agent._touch_activity("local model loading")
|
||||
_load_notice_shown = True
|
||||
_load_notice_misses = 0
|
||||
# Loading IS liveness for the heartbeat; the stale detector
|
||||
# needs no help — the local floor (900s) dwarfs any load.
|
||||
_last_heartbeat = _hb_now
|
||||
continue
|
||||
if _load_notice_shown:
|
||||
# One missed sample is routine (a /slots read straddling a
|
||||
# batch boundary, a 2s probe timeout under load) — clearing
|
||||
# on it made the status line strobe blank once every few
|
||||
# seconds mid-prefill. Only a SUSTAINED absence means the
|
||||
# phase really ended.
|
||||
_load_notice_misses += 1
|
||||
if _load_notice_misses >= 3:
|
||||
_load_notice_shown = False
|
||||
def _monitor_loop() -> None:
|
||||
nonlocal _last_heartbeat, _last_load_poll, _load_notice_shown, _load_notice_misses
|
||||
while _call_alive():
|
||||
_wait_call(0.3)
|
||||
|
||||
_hb_now = time.time()
|
||||
# Cold-load window: last_chunk_time is touched at request-client
|
||||
# creation and then only by REAL chunks, so "no chunk for 2s+" is
|
||||
# true through a model load (nothing can stream while the child is
|
||||
# still mapping weights) and false during healthy token flow —
|
||||
# which is what keeps this poll off the streaming hot path. The
|
||||
# probe itself is an in-memory snapshot read.
|
||||
if (
|
||||
_is_local_base
|
||||
and _hb_now - last_chunk_time["t"] >= 2.0
|
||||
and _hb_now - _last_load_poll >= 1.0
|
||||
):
|
||||
_last_load_poll = _hb_now
|
||||
_load_notice = _managed_local_load_notice(agent, api_kwargs)
|
||||
if _load_notice is not None:
|
||||
agent._emit_wait_notice(_load_notice)
|
||||
agent._touch_activity("local model loading")
|
||||
_load_notice_shown = True
|
||||
_load_notice_misses = 0
|
||||
agent._emit_wait_notice("")
|
||||
# Loading IS liveness for the heartbeat; the stale detector
|
||||
# needs no help — the local floor (900s) dwarfs any load.
|
||||
_last_heartbeat = _hb_now
|
||||
continue
|
||||
if _load_notice_shown:
|
||||
# One missed sample is routine (a /slots read straddling a
|
||||
# batch boundary, a 2s probe timeout under load) — clearing
|
||||
# on it made the status line strobe blank once every few
|
||||
# seconds mid-prefill. Only a SUSTAINED absence means the
|
||||
# phase really ended.
|
||||
_load_notice_misses += 1
|
||||
if _load_notice_misses >= 3:
|
||||
_load_notice_shown = False
|
||||
_load_notice_misses = 0
|
||||
agent._emit_wait_notice("")
|
||||
|
||||
# Periodic heartbeat: touch the agent's activity tracker so the
|
||||
# gateway's inactivity monitor knows we're alive while waiting
|
||||
# for stream chunks. Without this, long thinking pauses (e.g.
|
||||
# reasoning models) or slow prefill on local providers (Ollama)
|
||||
# trigger false inactivity timeouts. The _call thread touches
|
||||
# activity on each chunk, but the gap between API call start
|
||||
# and first chunk can exceed the gateway timeout — especially
|
||||
# when the stale-stream timeout is disabled (local providers).
|
||||
if _hb_now - _last_heartbeat >= _HEARTBEAT_INTERVAL:
|
||||
_last_heartbeat = _hb_now
|
||||
_waiting_secs = int(_hb_now - last_chunk_time["t"])
|
||||
if _waiting_secs >= _HEARTBEAT_INTERVAL:
|
||||
# No chunks for 30s+ — rewrite the live spinner/status line
|
||||
# so CLI/TUI/Desktop users see WHAT the wait is (slow or
|
||||
# overloaded provider / long thinking pause) instead of an
|
||||
# unexplained generic spinner, and WHEN recovery kicks in.
|
||||
if (
|
||||
_stream_stale_timeout is not None
|
||||
and _stream_stale_timeout != float("inf")
|
||||
):
|
||||
_recovery = f"; auto-reconnect at {int(_stream_stale_timeout)}s"
|
||||
# Periodic heartbeat: touch the agent's activity tracker so the
|
||||
# gateway's inactivity monitor knows we're alive while waiting
|
||||
# for stream chunks. Without this, long thinking pauses (e.g.
|
||||
# reasoning models) or slow prefill on local providers (Ollama)
|
||||
# trigger false inactivity timeouts. The _call thread touches
|
||||
# activity on each chunk, but the gap between API call start
|
||||
# and first chunk can exceed the gateway timeout — especially
|
||||
# when the stale-stream timeout is disabled (local providers).
|
||||
if _hb_now - _last_heartbeat >= _HEARTBEAT_INTERVAL:
|
||||
_last_heartbeat = _hb_now
|
||||
_waiting_secs = int(_hb_now - last_chunk_time["t"])
|
||||
if _waiting_secs >= _HEARTBEAT_INTERVAL:
|
||||
# No chunks for 30s+ — rewrite the live spinner/status line
|
||||
# so CLI/TUI/Desktop users see WHAT the wait is (slow or
|
||||
# overloaded provider / long thinking pause) instead of an
|
||||
# unexplained generic spinner, and WHEN recovery kicks in.
|
||||
if (
|
||||
_stream_stale_timeout is not None
|
||||
and _stream_stale_timeout != float("inf")
|
||||
):
|
||||
_recovery = f"; auto-reconnect at {int(_stream_stale_timeout)}s"
|
||||
else:
|
||||
_recovery = ""
|
||||
agent._emit_wait_notice(
|
||||
f"⏳ waiting on {api_kwargs.get('model', 'the provider')} — "
|
||||
f"{_waiting_secs}s with no output yet (provider may be "
|
||||
f"slow or overloaded, or the model is thinking{_recovery})"
|
||||
)
|
||||
else:
|
||||
_recovery = ""
|
||||
# Chunks are flowing — keep the activity tracker fresh but
|
||||
# leave the live display alone.
|
||||
agent._touch_activity(
|
||||
f"waiting for stream response ({_waiting_secs}s, no chunks yet)"
|
||||
)
|
||||
|
||||
# Detect stale streams: connections kept alive by SSE pings
|
||||
# but delivering no real chunks. Kill the client so the
|
||||
# inner retry loop can start a fresh connection.
|
||||
_stale_elapsed = time.time() - last_chunk_time["t"]
|
||||
if _stale_elapsed > _stream_stale_timeout:
|
||||
_est_ctx = estimate_request_context_tokens(api_kwargs)
|
||||
logger.warning(
|
||||
"Stream stale for %.0fs (threshold %.0fs) — no chunks received. "
|
||||
"model=%s context=~%s tokens. Killing connection.",
|
||||
_stale_elapsed, _stream_stale_timeout,
|
||||
api_kwargs.get("model", "unknown"), f"{_est_ctx:,}",
|
||||
)
|
||||
agent._buffer_status(
|
||||
f"⚠️ No response from provider for {int(_stale_elapsed)}s "
|
||||
f"(model: {api_kwargs.get('model', 'unknown')}, "
|
||||
f"context: ~{_est_ctx:,} tokens). "
|
||||
f"Reconnecting..."
|
||||
)
|
||||
try:
|
||||
_cancel_current_stream_attempt("stale_stream_kill")
|
||||
_close_request_client_once("stale_stream_kill")
|
||||
except Exception:
|
||||
pass
|
||||
# Circuit breaker (#58962): count the stale kill. See the
|
||||
# canonical comment block above ``_stale_streak()``.
|
||||
_bump_stale_streak(agent)
|
||||
# Rebuild the primary client too — its connection pool
|
||||
# may hold dead sockets from the same provider outage.
|
||||
if agent.api_mode == "anthropic_messages":
|
||||
# #67142: the stale stream ran on a request-local anthropic
|
||||
# client, already socket-aborted above via
|
||||
# _close_request_client_once (which unblocks the worker and
|
||||
# preserves the #28161 no-hang guarantee). The shared
|
||||
# _anthropic_client is NOT the in-flight transport, so we must
|
||||
# not close it from this poll (stranger) thread — that was the
|
||||
# FD-recycle corruption vector. Nothing further is needed.
|
||||
pass
|
||||
else:
|
||||
# #70773: same FD-recycle corruption vector as #67142.
|
||||
# The shared OpenAI client's connection pool must NOT be
|
||||
# closed from this watchdog/poll thread — worker threads
|
||||
# from previous stale-killed attempts may still be
|
||||
# unwinding their SSL BIOs. The request-local client is
|
||||
# already closed above via _close_request_client_once.
|
||||
# The shared client will be replaced lazily by
|
||||
# _ensure_primary_openai_client on the next request.
|
||||
pass
|
||||
# Reset the timer so we don't kill repeatedly while
|
||||
# the inner thread processes the closure.
|
||||
last_chunk_time["t"] = time.time()
|
||||
agent._emit_wait_notice(
|
||||
f"⏳ waiting on {api_kwargs.get('model', 'the provider')} — "
|
||||
f"{_waiting_secs}s with no output yet (provider may be "
|
||||
f"slow or overloaded, or the model is thinking{_recovery})"
|
||||
f"⚠ no output from provider for {int(_stale_elapsed)}s — "
|
||||
f"reconnecting..."
|
||||
)
|
||||
else:
|
||||
# Chunks are flowing — keep the activity tracker fresh but
|
||||
# leave the live display alone.
|
||||
agent._touch_activity(
|
||||
f"waiting for stream response ({_waiting_secs}s, no chunks yet)"
|
||||
f"stale stream detected after {int(_stale_elapsed)}s, reconnecting"
|
||||
)
|
||||
|
||||
# Detect stale streams: connections kept alive by SSE pings
|
||||
# but delivering no real chunks. Kill the client so the
|
||||
# inner retry loop can start a fresh connection.
|
||||
_stale_elapsed = time.time() - last_chunk_time["t"]
|
||||
if _stale_elapsed > _stream_stale_timeout:
|
||||
_est_ctx = estimate_request_context_tokens(api_kwargs)
|
||||
logger.warning(
|
||||
"Stream stale for %.0fs (threshold %.0fs) — no chunks received. "
|
||||
"model=%s context=~%s tokens. Killing connection.",
|
||||
_stale_elapsed, _stream_stale_timeout,
|
||||
api_kwargs.get("model", "unknown"), f"{_est_ctx:,}",
|
||||
)
|
||||
agent._buffer_status(
|
||||
f"⚠️ No response from provider for {int(_stale_elapsed)}s "
|
||||
f"(model: {api_kwargs.get('model', 'unknown')}, "
|
||||
f"context: ~{_est_ctx:,} tokens). "
|
||||
f"Reconnecting..."
|
||||
)
|
||||
try:
|
||||
_cancel_current_stream_attempt("stale_stream_kill")
|
||||
_close_request_client_once("stale_stream_kill")
|
||||
except Exception:
|
||||
pass
|
||||
# Circuit breaker (#58962): count the stale kill. See the
|
||||
# canonical comment block above ``_stale_streak()``.
|
||||
_bump_stale_streak(agent)
|
||||
# Rebuild the primary client too — its connection pool
|
||||
# may hold dead sockets from the same provider outage.
|
||||
if agent.api_mode == "anthropic_messages":
|
||||
# #67142: the stale stream ran on a request-local anthropic
|
||||
# client, already socket-aborted above via
|
||||
# _close_request_client_once (which unblocks the worker and
|
||||
# preserves the #28161 no-hang guarantee). The shared
|
||||
# _anthropic_client is NOT the in-flight transport, so we must
|
||||
# not close it from this poll (stranger) thread — that was the
|
||||
# FD-recycle corruption vector. Nothing further is needed.
|
||||
pass
|
||||
else:
|
||||
# #70773: same FD-recycle corruption vector as #67142.
|
||||
# The shared OpenAI client's connection pool must NOT be
|
||||
# closed from this watchdog/poll thread — worker threads
|
||||
# from previous stale-killed attempts may still be
|
||||
# unwinding their SSL BIOs. The request-local client is
|
||||
# already closed above via _close_request_client_once.
|
||||
# The shared client will be replaced lazily by
|
||||
# _ensure_primary_openai_client on the next request.
|
||||
pass
|
||||
# Reset the timer so we don't kill repeatedly while
|
||||
# the inner thread processes the closure.
|
||||
last_chunk_time["t"] = time.time()
|
||||
agent._emit_wait_notice(
|
||||
f"⚠ no output from provider for {int(_stale_elapsed)}s — "
|
||||
f"reconnecting..."
|
||||
)
|
||||
agent._touch_activity(
|
||||
f"stale stream detected after {int(_stale_elapsed)}s, reconnecting"
|
||||
)
|
||||
|
||||
if agent._interrupt_requested:
|
||||
# The stale branch above already counted this iteration when its
|
||||
# deadline won the race; do not double-count a simultaneous stop.
|
||||
if _stale_elapsed <= _stream_stale_timeout:
|
||||
_record_interrupted_provider_wait(
|
||||
agent,
|
||||
_stale_elapsed,
|
||||
response_started=deltas_were_sent["yes"],
|
||||
if agent._interrupt_requested:
|
||||
# The stale branch above already counted this iteration when its
|
||||
# deadline won the race; do not double-count a simultaneous stop.
|
||||
if _stale_elapsed <= _stream_stale_timeout:
|
||||
_record_interrupted_provider_wait(
|
||||
agent,
|
||||
_stale_elapsed,
|
||||
response_started=deltas_were_sent["yes"],
|
||||
)
|
||||
# Mark THIS request cancelled before force-closing so the worker's
|
||||
# exception handler recognizes the forced transport error as a
|
||||
# cancel and exits without retrying or surfacing a network error.
|
||||
# (#6600)
|
||||
_request_cancelled["value"] = True
|
||||
logger.debug(
|
||||
"Force-closing streaming httpx client due to interrupt "
|
||||
"(not a network error)."
|
||||
)
|
||||
# Mark THIS request cancelled before force-closing so the worker's
|
||||
# exception handler recognizes the forced transport error as a
|
||||
# cancel and exits without retrying or surfacing a network error.
|
||||
# (#6600)
|
||||
_request_cancelled["value"] = True
|
||||
logger.debug(
|
||||
"Force-closing streaming httpx client due to interrupt "
|
||||
"(not a network error)."
|
||||
)
|
||||
try:
|
||||
_cancel_current_stream_attempt("stream_interrupt_abort")
|
||||
# #67142: kind-aware — anthropic aborts the request-local
|
||||
# client's socket from this poll thread; the shared
|
||||
# _anthropic_client is never closed here.
|
||||
_close_request_client_once("stream_interrupt_abort")
|
||||
except Exception:
|
||||
pass
|
||||
# Wait for the worker to unwind Relay-managed stream scopes
|
||||
# (physical LLM + deferred logical) before surfacing
|
||||
# InterruptedError. Raising immediately lets turn teardown
|
||||
# (finish_logical_calls / end_turn / close_session) race a
|
||||
# still-open physical scope and corrupt the LIFO stack —
|
||||
# "scope handle is not at the top of the stack" → CLI EIO /
|
||||
# redraw storm (#81521). No-op when Relay managed execution
|
||||
# is not live.
|
||||
_join_worker_for_relay_teardown(t, label="Streaming")
|
||||
raise InterruptedError("Agent interrupted during streaming API call")
|
||||
try:
|
||||
_cancel_current_stream_attempt("stream_interrupt_abort")
|
||||
# #67142: kind-aware — anthropic aborts the request-local
|
||||
# client's socket from this poll thread; the shared
|
||||
# _anthropic_client is never closed here.
|
||||
_close_request_client_once("stream_interrupt_abort")
|
||||
except Exception:
|
||||
pass
|
||||
# Wait for the worker to unwind Relay-managed stream scopes
|
||||
# (physical LLM + deferred logical) before surfacing
|
||||
# InterruptedError. Raising immediately lets turn teardown
|
||||
# (finish_logical_calls / end_turn / close_session) race a
|
||||
# still-open physical scope and corrupt the LIFO stack —
|
||||
# "scope handle is not at the top of the stack" → CLI EIO /
|
||||
# redraw storm (#81521). No-op when Relay managed execution
|
||||
# is not live. (Inline mode has no worker: the request runs
|
||||
# on the caller's thread and has already unwound by the time
|
||||
# the InterruptedError below is raised.)
|
||||
if t is not None:
|
||||
_join_worker_for_relay_teardown(t, label="Streaming")
|
||||
_monitor_interrupted["yes"] = True
|
||||
return
|
||||
|
||||
if _inline:
|
||||
# Request on THIS thread; heartbeat / stale / interrupt monitor on a
|
||||
# side thread that only ever aborts sockets (never dispatches).
|
||||
monitor = threading.Thread(
|
||||
target=_context_thread_target(_monitor_loop),
|
||||
name="stream-inline-monitor",
|
||||
daemon=True,
|
||||
)
|
||||
monitor.start()
|
||||
try:
|
||||
_run_call()
|
||||
finally:
|
||||
monitor.join(timeout=2.0)
|
||||
else:
|
||||
_monitor_loop()
|
||||
if _monitor_interrupted["yes"]:
|
||||
raise InterruptedError("Agent interrupted during streaming API call")
|
||||
# Worker thread exited before the main thread's poll loop could check
|
||||
# the interrupt flag. If the worker returned early due to an interrupt
|
||||
# (e.g. _call_anthropic() detected _interrupt_requested and returned
|
||||
|
||||
@@ -0,0 +1,243 @@
|
||||
"""Delegated children and cron turns stream on the wire (#90202, #100260).
|
||||
|
||||
``should_use_direct_api_call`` contexts (gateway cron turns, delegate_task
|
||||
children) must not spawn the interrupt worker — it wedges inside their nested
|
||||
thread pools (#62151, #60203). The original fix short-circuited them onto the
|
||||
NON-streaming wire, which silently dropped every liveness property streaming
|
||||
provides: edge proxies killed the silent POST (z.ai HTTP 524, #90202), and the
|
||||
non-stream stale watchdog could not tell a reasoning model's thinking phase
|
||||
from a hung provider (#100260 — children died at exactly ``stale_timeout``).
|
||||
|
||||
These tests pin the replacement contract: those contexts stay on the streaming
|
||||
path, issue ``stream=True`` on the calling thread (no worker), and keep the
|
||||
stale detector + cross-thread interrupt abort working from the monitor thread.
|
||||
"""
|
||||
|
||||
import json
|
||||
import threading
|
||||
import time
|
||||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
import run_agent
|
||||
from agent import chat_completion_helpers as helpers
|
||||
from agent.chat_completion_helpers import (
|
||||
interruptible_streaming_api_call,
|
||||
should_use_direct_api_call,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Real OpenAI-wire SSE server: records the wire ``stream`` flag per request.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class _Wire:
|
||||
def __init__(self, *, stall_after_first_chunk: bool = False):
|
||||
self.requests: list[dict] = []
|
||||
self.stall = stall_after_first_chunk
|
||||
self.hits = threading.Semaphore(0)
|
||||
wire = self
|
||||
|
||||
class Handler(BaseHTTPRequestHandler):
|
||||
def log_message(self, *_a):
|
||||
pass
|
||||
|
||||
def do_POST(self):
|
||||
n = int(self.headers.get("content-length", 0))
|
||||
body = json.loads(self.rfile.read(n) or b"{}")
|
||||
if not self.path.endswith("/chat/completions"):
|
||||
# Local-endpoint capability probes (/api/show etc.) —
|
||||
# answer fast so agent construction never waits on the
|
||||
# stalling stream below.
|
||||
self.send_response(404)
|
||||
self.end_headers()
|
||||
return
|
||||
wire.requests.append(body)
|
||||
wire.hits.release()
|
||||
self.send_response(200)
|
||||
self.send_header("content-type", "text/event-stream")
|
||||
self.end_headers()
|
||||
first = {
|
||||
"id": "c1", "object": "chat.completion.chunk", "created": 1, "model": "m",
|
||||
"choices": [{"index": 0, "delta": {"role": "assistant", "content": "hello"},
|
||||
"finish_reason": None}],
|
||||
}
|
||||
self.wfile.write(f"data: {json.dumps(first)}\n\n".encode())
|
||||
self.wfile.flush()
|
||||
if wire.stall:
|
||||
try:
|
||||
for _ in range(400):
|
||||
time.sleep(0.05)
|
||||
self.wfile.write(b": keepalive\n\n")
|
||||
self.wfile.flush()
|
||||
except Exception:
|
||||
pass
|
||||
return
|
||||
second = {
|
||||
"id": "c1", "object": "chat.completion.chunk", "created": 1, "model": "m",
|
||||
"choices": [{"index": 0, "delta": {"content": " world"}, "finish_reason": None}],
|
||||
}
|
||||
fin = {
|
||||
"id": "c1", "object": "chat.completion.chunk", "created": 1, "model": "m",
|
||||
"choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}],
|
||||
"usage": {"prompt_tokens": 3, "completion_tokens": 2, "total_tokens": 5},
|
||||
}
|
||||
for c in (second, fin):
|
||||
self.wfile.write(f"data: {json.dumps(c)}\n\n".encode())
|
||||
self.wfile.write(b"data: [DONE]\n\n")
|
||||
self.wfile.flush()
|
||||
|
||||
self.server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
|
||||
threading.Thread(target=self.server.serve_forever, daemon=True).start()
|
||||
self.base_url = f"http://127.0.0.1:{self.server.server_address[1]}/v1"
|
||||
|
||||
def close(self):
|
||||
self.server.shutdown()
|
||||
self.server.server_close()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def wire():
|
||||
w = _Wire()
|
||||
yield w
|
||||
w.close()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def stalling_wire():
|
||||
w = _Wire(stall_after_first_chunk=True)
|
||||
yield w
|
||||
w.close()
|
||||
|
||||
|
||||
def _make_agent(base_url: str, *, platform: str):
|
||||
return run_agent.AIAgent(
|
||||
api_key="test-key",
|
||||
base_url=base_url,
|
||||
model="m",
|
||||
provider="custom",
|
||||
platform=platform,
|
||||
quiet_mode=True,
|
||||
skip_context_files=True,
|
||||
skip_memory=True,
|
||||
enabled_toolsets=[],
|
||||
max_iterations=1,
|
||||
)
|
||||
|
||||
|
||||
_KW = {"model": "m", "messages": [{"role": "user", "content": "hi"}]}
|
||||
|
||||
|
||||
@pytest.mark.parametrize("platform", ["subagent", "cron"])
|
||||
def test_direct_contexts_stream_on_the_wire_and_on_the_calling_thread(wire, platform):
|
||||
agent = _make_agent(wire.base_url, platform=platform)
|
||||
assert should_use_direct_api_call(agent) is True
|
||||
|
||||
issued_on = {}
|
||||
real_create = agent._create_request_openai_client
|
||||
|
||||
def spy(*a, **k):
|
||||
issued_on["tid"] = threading.get_ident()
|
||||
return real_create(*a, **k)
|
||||
|
||||
agent._create_request_openai_client = spy
|
||||
|
||||
response = interruptible_streaming_api_call(agent, dict(_KW))
|
||||
|
||||
completions = [r for r in wire.requests if "messages" in r]
|
||||
assert completions, "no chat completion reached the wire"
|
||||
assert completions[-1].get("stream") is True, (
|
||||
f"{platform} turn went out non-streaming: stream={completions[-1].get('stream')!r}"
|
||||
)
|
||||
# No interrupt worker: the request was dispatched from the caller's thread
|
||||
# (the #62151 / #60203 deadlock class needs the request on a spawned worker).
|
||||
assert issued_on["tid"] == threading.get_ident()
|
||||
assert response.choices[0].message.content == "hello world"
|
||||
assert response.choices[0].finish_reason == "stop"
|
||||
|
||||
|
||||
def test_interactive_platform_still_uses_the_worker_thread(wire):
|
||||
"""Regression guard for the refactor: non-direct contexts keep the
|
||||
interrupt worker (interactive /stop responsiveness relies on it)."""
|
||||
agent = _make_agent(wire.base_url, platform="cli")
|
||||
assert should_use_direct_api_call(agent) is False
|
||||
|
||||
issued_on = {}
|
||||
real_create = agent._create_request_openai_client
|
||||
|
||||
def spy(*a, **k):
|
||||
issued_on["tid"] = threading.get_ident()
|
||||
return real_create(*a, **k)
|
||||
|
||||
agent._create_request_openai_client = spy
|
||||
response = interruptible_streaming_api_call(agent, dict(_KW))
|
||||
|
||||
assert issued_on["tid"] != threading.get_ident()
|
||||
assert response.choices[0].message.content == "hello world"
|
||||
|
||||
|
||||
def test_inline_stream_stale_detector_still_fires_from_monitor_thread(
|
||||
stalling_wire, monkeypatch
|
||||
):
|
||||
"""The stale-stream detector moved onto a monitor thread for inline
|
||||
mode; a stream that sends one chunk then only keep-alives must still be
|
||||
killed at the stale budget instead of hanging until the socket dies."""
|
||||
monkeypatch.setenv("HERMES_STREAM_STALE_TIMEOUT", "1.0")
|
||||
monkeypatch.setenv("HERMES_STREAM_RETRIES", "0")
|
||||
agent = _make_agent(stalling_wire.base_url, platform="subagent")
|
||||
|
||||
started = time.time()
|
||||
response = interruptible_streaming_api_call(agent, dict(_KW))
|
||||
elapsed = time.time() - started
|
||||
|
||||
assert elapsed < 6.0, f"inline stream was not bounded by the stale detector ({elapsed:.1f}s)"
|
||||
# A partial delta was delivered → the loop gets the length-truncated
|
||||
# partial-stream stub (same contract as the worker path).
|
||||
assert getattr(response, "id", None) == helpers.PARTIAL_STREAM_STUB_ID
|
||||
assert response.choices[0].finish_reason == helpers.FINISH_REASON_LENGTH
|
||||
|
||||
|
||||
def test_inline_stream_cross_thread_interrupt_aborts_promptly(stalling_wire, monkeypatch):
|
||||
"""``AIAgent.interrupt()`` from another thread (cron watchdog, delegation
|
||||
stall monitor) must abort the inline stream and surface InterruptedError
|
||||
— the property the direct_api_call path guaranteed via
|
||||
``_active_request_abort``."""
|
||||
monkeypatch.setenv("HERMES_STREAM_STALE_TIMEOUT", "60")
|
||||
monkeypatch.setenv("HERMES_STREAM_RETRIES", "0")
|
||||
agent = _make_agent(stalling_wire.base_url, platform="cron")
|
||||
box: dict = {}
|
||||
|
||||
def _run():
|
||||
t0 = time.time()
|
||||
try:
|
||||
interruptible_streaming_api_call(agent, dict(_KW))
|
||||
box["outcome"] = "returned"
|
||||
except BaseException as exc: # noqa: BLE001 — record whatever surfaces
|
||||
box["outcome"] = type(exc).__name__
|
||||
box["elapsed"] = time.time() - t0
|
||||
|
||||
worker = threading.Thread(target=_run, daemon=True)
|
||||
worker.start()
|
||||
assert stalling_wire.hits.acquire(timeout=5.0), "request never reached the wire"
|
||||
time.sleep(0.3) # let the first chunk land
|
||||
agent.interrupt("test interrupt")
|
||||
worker.join(timeout=10.0)
|
||||
|
||||
assert not worker.is_alive(), "inline stream did not unwind after interrupt"
|
||||
assert box["outcome"] == "InterruptedError"
|
||||
assert box["elapsed"] < 5.0
|
||||
|
||||
|
||||
def test_should_use_direct_api_call_gate_is_unchanged():
|
||||
"""The routing predicate itself is untouched — only what it routes to."""
|
||||
def mk(platform, api_mode="chat_completions", provider="openrouter"):
|
||||
return SimpleNamespace(platform=platform, api_mode=api_mode, provider=provider)
|
||||
|
||||
assert should_use_direct_api_call(mk("cron")) is True
|
||||
assert should_use_direct_api_call(mk("subagent")) is True
|
||||
assert should_use_direct_api_call(mk("cli")) is False
|
||||
assert should_use_direct_api_call(mk("cron", api_mode="anthropic_messages")) is False
|
||||
assert should_use_direct_api_call(mk("cron", provider="moa")) is False
|
||||
@@ -1152,7 +1152,9 @@ The **stale stream detection** kills connections that receive SSE keep-alive pin
|
||||
|
||||
The **stale non-stream detection** kills non-streaming calls that produce no response for too long. By default Hermes disables this on local endpoints to avoid false positives during long prefills. If you explicitly set `providers.<id>.stale_timeout_seconds`, `providers.<id>.models.<model>.stale_timeout_seconds`, or `HERMES_API_CALL_STALE_TIMEOUT`, that explicit value is honored even on local endpoints.
|
||||
|
||||
This budget bounds every non-streaming call, including the ones cron jobs and delegated subagents run inline. A provider that accepts a request and then goes silent — connection held open, no bytes, no error — is aborted at the stale timeout and retried, rather than hanging until the much longer socket read timeout (or, for an unattended cron run, until something external kills the process).
|
||||
This budget bounds every non-streaming call. A provider that accepts a request and then goes silent — connection held open, no bytes, no error — is aborted at the stale timeout and retried, rather than hanging until the much longer socket read timeout (or, for an unattended cron run, until something external kills the process).
|
||||
|
||||
Cron jobs and delegated subagents stream too. They run the request inline on their own thread (the interrupt worker other sessions use wedges inside the gateway's nested thread pools), but the wire request is still `stream: true`, so the **stale stream detection** budget above governs them — every token counts as liveness, so a reasoning model that thinks for minutes is not mistaken for a hung provider, and edge proxies that kill silent connections keep seeing bytes.
|
||||
|
||||
## Context Pressure Warnings
|
||||
|
||||
|
||||
Reference in New Issue
Block a user