fix(clarify): honor agent.clarify_timeout and support unlimited wait (#7163)
The WebUI clarify bridge resolved the prompt timeout from a local, divergent copy (legacy clarify.timeout, 120s fallback) and treated 0 as invalid, so agent.clarify_timeout: 0 (unlimited) could not be honored and there was no way to make a clarification card wait for the user. Resolve the timeout with the agent core's canonical resolve_clarify_timeout() (clarify.timeout -> agent.clarify_timeout -> 3600, <= 0 = unlimited) against the session's own profile config (the streaming worker thread does not inherit the request's per-profile context), wait with a null deadline when unlimited, and stop the frontend rendering an expired countdown for non-positive timeouts. - api/streaming.py: per-session profile config resolution, canonical resolver with a compatibility fallback, and an extracted _await_clarify_response() helper that polls in 1s slices and honours cancellation. - api/clarify.py: preserve timeout_seconds <= 0 and emit no expires_at. - static/messages.js: _clarifyExpiryMs() returns 0 for non-positive timeouts so the card stays up without a countdown. - tests/test_sprint42.py: resolution matrix, per-session profile divergence, unlimited resolve-later / cancel-exit, and no-expiry payload. Closes #7159 Co-authored-by: nesquena-hermes <nesquena+hermes@gmail.com>
This commit is contained in:
1 parent
c6842e29ee
commit
0b5d2f3698
4 files changed
+197
-34
No files matched your search
+8
-3
@@ -86,11 +86,16 @@ def clear_pending(session_key: str) -> int:
|
|||||||
def _with_timeout_metadata(data: dict) -> dict:
|
def _with_timeout_metadata(data: dict) -> dict:
|
||||||
item = dict(data or {})
|
item = dict(data or {})
|
||||||
requested_at = float(item.get("requested_at") or time.time())
|
requested_at = float(item.get("requested_at") or time.time())
|
||||||
timeout_seconds = int(item.get("timeout_seconds") or DEFAULT_TIMEOUT_SECONDS)
|
raw_timeout = item.get("timeout_seconds")
|
||||||
expires_at = float(item.get("expires_at") or requested_at + timeout_seconds)
|
timeout_seconds = int(raw_timeout) if raw_timeout is not None else DEFAULT_TIMEOUT_SECONDS
|
||||||
item["requested_at"] = requested_at
|
item["requested_at"] = requested_at
|
||||||
item["timeout_seconds"] = timeout_seconds
|
item["timeout_seconds"] = timeout_seconds
|
||||||
item["expires_at"] = expires_at
|
if timeout_seconds <= 0:
|
||||||
|
# Unlimited: no expiry. The frontend renders no countdown and the card
|
||||||
|
# waits until the user answers (or the run is cancelled).
|
||||||
|
item["expires_at"] = 0
|
||||||
|
return item
|
||||||
|
item["expires_at"] = float(item.get("expires_at") or requested_at + timeout_seconds)
|
||||||
return item
|
return item
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
+77
-26
@@ -790,17 +790,82 @@ def _is_quota_error_text(err_text: str) -> bool:
|
|||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
def _clarify_timeout_seconds(default: int = 120) -> int:
|
def _clarify_session_config(sid: str) -> dict | None:
|
||||||
"""Resolve clarify timeout from config, with bounded fallback."""
|
"""Config dict for the clarify session's profile, or None to use ambient.
|
||||||
|
|
||||||
|
The agent runs on a detached worker thread that does not inherit the
|
||||||
|
per-request TLS profile context, so the ambient ``get_config()`` would
|
||||||
|
resolve to the process-global profile (usually ``default``) even for a
|
||||||
|
session running under a named profile (issue #3294 pattern). Resolve the
|
||||||
|
session's own profile home instead, mirroring how the rest of the run is
|
||||||
|
configured.
|
||||||
|
"""
|
||||||
try:
|
try:
|
||||||
cfg = get_config()
|
from api.models import _get_profile_home
|
||||||
raw = cfg.get("clarify", {}).get("timeout", default)
|
from api.config import get_config_for_profile_home
|
||||||
timeout_seconds = int(raw)
|
session = get_session(sid)
|
||||||
if timeout_seconds <= 0:
|
if session is None:
|
||||||
return default
|
return None
|
||||||
return timeout_seconds
|
return get_config_for_profile_home(_get_profile_home(getattr(session, "profile", None)))
|
||||||
except Exception:
|
except Exception:
|
||||||
return default
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
def _clarify_timeout_seconds(config: dict | None = None, default: int = 3600) -> int:
|
||||||
|
"""Resolve the clarify timeout (seconds) for WebUI clarify prompts.
|
||||||
|
|
||||||
|
Uses the agent core's canonical resolver when available
|
||||||
|
(``tools.clarify_gateway.resolve_clarify_timeout`` — the same function the
|
||||||
|
CLI and messaging gateways use), falling back to an inline mirror of its
|
||||||
|
resolution order when the agent package is not importable. Resolution:
|
||||||
|
legacy top-level ``clarify.timeout`` if explicitly set, else the canonical
|
||||||
|
``agent.clarify_timeout``, else 3600. ``<= 0`` is preserved verbatim and
|
||||||
|
means *unlimited*: the prompt waits until the user answers or the run is
|
||||||
|
cancelled — it is never auto-skipped.
|
||||||
|
|
||||||
|
``config`` may be an explicit profile config dict (see
|
||||||
|
:func:`_clarify_session_config`); when None, the ambient ``get_config()``
|
||||||
|
is used.
|
||||||
|
"""
|
||||||
|
cfg = config if config is not None else (get_config() or {})
|
||||||
|
try:
|
||||||
|
from tools.clarify_gateway import resolve_clarify_timeout
|
||||||
|
return int(resolve_clarify_timeout(cfg))
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
try:
|
||||||
|
raw = (cfg.get("clarify") or {}).get("timeout")
|
||||||
|
if raw is None:
|
||||||
|
raw = (cfg.get("agent") or {}).get("clarify_timeout", 3600)
|
||||||
|
return int(raw)
|
||||||
|
except Exception:
|
||||||
|
return int(default)
|
||||||
|
|
||||||
|
|
||||||
|
def _await_clarify_response(entry, timeout, cancel_evt) -> tuple[str, bool]:
|
||||||
|
"""Block until a clarify entry resolves, or (finite timeout only) expires.
|
||||||
|
|
||||||
|
``timeout <= 0`` means *unlimited*: no deadline — the prompt waits until
|
||||||
|
the user answers or the run is cancelled. Polls the entry's event in ~1s
|
||||||
|
slices so a cancelled run unblocks promptly, and still honours the
|
||||||
|
``is_interrupted``-style cancel check each slice.
|
||||||
|
|
||||||
|
Returns ``(response, expired)``: ``expired`` is True when it returned
|
||||||
|
because of cancellation or a finite timeout without a user response (the
|
||||||
|
caller should clear pending and fall back), False when the user answered.
|
||||||
|
"""
|
||||||
|
deadline = None if timeout <= 0 else time.monotonic() + timeout
|
||||||
|
while True:
|
||||||
|
if cancel_evt.is_set():
|
||||||
|
return "", True
|
||||||
|
wait_for = 1.0
|
||||||
|
if deadline is not None:
|
||||||
|
remaining = deadline - time.monotonic()
|
||||||
|
if remaining <= 0:
|
||||||
|
return "", True
|
||||||
|
wait_for = min(1.0, remaining)
|
||||||
|
if entry.event.wait(timeout=wait_for):
|
||||||
|
return str(entry.result or "").strip(), False
|
||||||
|
|
||||||
|
|
||||||
_CANCEL_MARKER_PATTERNS = ('task cancelled', 'task canceled', 'response interrupted')
|
_CANCEL_MARKER_PATTERNS = ('task cancelled', 'task canceled', 'response interrupted')
|
||||||
@@ -8859,7 +8924,7 @@ def _run_agent_streaming(
|
|||||||
|
|
||||||
def _clarify_callback_impl(question, choices, sid, cancel_evt, put_event):
|
def _clarify_callback_impl(question, choices, sid, cancel_evt, put_event):
|
||||||
"""Bridge Hermes clarify prompts to the WebUI."""
|
"""Bridge Hermes clarify prompts to the WebUI."""
|
||||||
timeout = _clarify_timeout_seconds()
|
timeout = _clarify_timeout_seconds(_clarify_session_config(sid))
|
||||||
choices_list = [str(choice) for choice in (choices or [])]
|
choices_list = [str(choice) for choice in (choices or [])]
|
||||||
data = {
|
data = {
|
||||||
'question': str(question or ''),
|
'question': str(question or ''),
|
||||||
@@ -8878,23 +8943,9 @@ def _run_agent_streaming(
|
|||||||
)
|
)
|
||||||
|
|
||||||
entry = _submit_clarify_pending(sid, data)
|
entry = _submit_clarify_pending(sid, data)
|
||||||
deadline = time.monotonic() + timeout
|
response, expired = _await_clarify_response(entry, timeout, cancel_evt)
|
||||||
while True:
|
if expired:
|
||||||
if cancel_evt.is_set():
|
|
||||||
_clear_clarify_pending(sid)
|
_clear_clarify_pending(sid)
|
||||||
return (
|
|
||||||
"The user did not provide a response within the time limit. "
|
|
||||||
"Use your best judgement to make the choice and proceed."
|
|
||||||
)
|
|
||||||
remaining = deadline - time.monotonic()
|
|
||||||
if remaining <= 0:
|
|
||||||
_clear_clarify_pending(sid)
|
|
||||||
return (
|
|
||||||
"The user did not provide a response within the time limit. "
|
|
||||||
"Use your best judgement to make the choice and proceed."
|
|
||||||
)
|
|
||||||
if entry.event.wait(timeout=min(1.0, remaining)):
|
|
||||||
response = str(entry.result or "").strip()
|
|
||||||
return (
|
return (
|
||||||
response
|
response
|
||||||
or "The user did not provide a response within the time limit. "
|
or "The user did not provide a response within the time limit. "
|
||||||
|
|||||||
+1
-1
@@ -8605,7 +8605,7 @@ function _clarifyExpiryMs(pending) {
|
|||||||
if (Number.isFinite(expiresAt) && expiresAt > 0) return expiresAt * 1000;
|
if (Number.isFinite(expiresAt) && expiresAt > 0) return expiresAt * 1000;
|
||||||
const requestedAt = Number(pending && pending.requested_at);
|
const requestedAt = Number(pending && pending.requested_at);
|
||||||
const timeoutSeconds = Number(pending && pending.timeout_seconds);
|
const timeoutSeconds = Number(pending && pending.timeout_seconds);
|
||||||
if (Number.isFinite(requestedAt) && Number.isFinite(timeoutSeconds)) {
|
if (Number.isFinite(requestedAt) && Number.isFinite(timeoutSeconds) && timeoutSeconds > 0) {
|
||||||
return (requestedAt + timeoutSeconds) * 1000;
|
return (requestedAt + timeoutSeconds) * 1000;
|
||||||
}
|
}
|
||||||
return 0;
|
return 0;
|
||||||
|
|||||||
+111
-4
@@ -10,6 +10,7 @@ Covers:
|
|||||||
"""
|
"""
|
||||||
import ast
|
import ast
|
||||||
import threading
|
import threading
|
||||||
|
import time
|
||||||
import pathlib
|
import pathlib
|
||||||
import re
|
import re
|
||||||
import queue
|
import queue
|
||||||
@@ -465,6 +466,113 @@ class TestRuntimeRouteInjection(unittest.TestCase):
|
|||||||
|
|
||||||
def test_clarify_callback_passes_configured_timeout_seconds(self):
|
def test_clarify_callback_passes_configured_timeout_seconds(self):
|
||||||
"""clarify prompt data should use clarify.timeout from config when present."""
|
"""clarify prompt data should use clarify.timeout from config when present."""
|
||||||
|
result = self._capture_clarify_timeout({"clarify": {"timeout": 300}})
|
||||||
|
self.assertEqual(result["clarify_result"], "selected")
|
||||||
|
self.assertEqual(len(result["payloads"]), 1)
|
||||||
|
self.assertEqual(result["payloads"][0]["timeout_seconds"], 300)
|
||||||
|
|
||||||
|
def test_clarify_callback_zero_timeout_is_unlimited(self):
|
||||||
|
"""0 (or negative) must pass through as unlimited — never fall back to a default."""
|
||||||
|
for config in ({"clarify": {"timeout": 0}}, {"agent": {"clarify_timeout": 0}}):
|
||||||
|
with self.subTest(config=config):
|
||||||
|
result = self._capture_clarify_timeout(config)
|
||||||
|
self.assertEqual(len(result["payloads"]), 1)
|
||||||
|
self.assertEqual(result["payloads"][0]["timeout_seconds"], 0)
|
||||||
|
|
||||||
|
def test_clarify_timeout_falls_back_to_agent_clarify_timeout(self):
|
||||||
|
"""Without the legacy clarify.timeout key, agent.clarify_timeout governs (canonical order)."""
|
||||||
|
import api.streaming as streaming
|
||||||
|
with mock.patch.object(streaming, "get_config", return_value={"agent": {"clarify_timeout": 45}}):
|
||||||
|
self.assertEqual(streaming._clarify_timeout_seconds(), 45)
|
||||||
|
with mock.patch.object(streaming, "get_config", return_value={}):
|
||||||
|
self.assertEqual(streaming._clarify_timeout_seconds(), 3600)
|
||||||
|
|
||||||
|
def test_clarify_session_profile_beats_ambient(self):
|
||||||
|
"""A non-default session's profile config must win over the ambient
|
||||||
|
(process-default) config — the exact divergence this fix resolves."""
|
||||||
|
import api.streaming as streaming
|
||||||
|
fake_session = type("FakeSession", (), {"profile": "ops"})()
|
||||||
|
with mock.patch.object(streaming, "get_session", return_value=fake_session), \
|
||||||
|
mock.patch("api.models._get_profile_home", return_value=pathlib.Path("/fake/profiles/ops")), \
|
||||||
|
mock.patch("api.config.get_config_for_profile_home", return_value={"agent": {"clarify_timeout": 0}}), \
|
||||||
|
mock.patch.object(streaming, "get_config", return_value={"agent": {"clarify_timeout": 600}}):
|
||||||
|
# session profile says 0 (unlimited); ambient says 600 → session wins
|
||||||
|
self.assertEqual(
|
||||||
|
streaming._clarify_timeout_seconds(streaming._clarify_session_config("sess-x")),
|
||||||
|
0,
|
||||||
|
)
|
||||||
|
|
||||||
|
def test_clarify_timeout_resolution_matrix(self):
|
||||||
|
"""The resolver must match the Agent core's canonical resolution order:
|
||||||
|
legacy clarify.timeout wins, else agent.clarify_timeout, else 3600;
|
||||||
|
non-positive values stay unlimited, invalid values fall back to 3600."""
|
||||||
|
import api.streaming as streaming
|
||||||
|
cases = [
|
||||||
|
({"clarify": {"timeout": 120}}, 120), # legacy wins
|
||||||
|
({"clarify": {"timeout": 120}, "agent": {"clarify_timeout": 0}}, 120), # legacy beats canonical
|
||||||
|
({"agent": {"clarify_timeout": 45}}, 45), # canonical fallback
|
||||||
|
({}, 3600), # absent → 3600
|
||||||
|
({"agent": {"clarify_timeout": "abc"}}, 3600), # invalid → 3600
|
||||||
|
({"clarify": {"timeout": "abc"}}, 3600), # invalid legacy → 3600
|
||||||
|
({"agent": {"clarify_timeout": 0}}, 0), # unlimited
|
||||||
|
({"agent": {"clarify_timeout": -5}}, -5), # negative → unlimited
|
||||||
|
({"clarify": {"timeout": -1}}, -1), # legacy negative → unlimited
|
||||||
|
]
|
||||||
|
for cfg, expected in cases:
|
||||||
|
with self.subTest(config=cfg):
|
||||||
|
self.assertEqual(streaming._clarify_timeout_seconds(cfg), expected)
|
||||||
|
|
||||||
|
def test_await_clarify_unlimited_resolves_later(self):
|
||||||
|
"""An unlimited (timeout<=0) prompt keeps waiting until the user answers."""
|
||||||
|
import api.streaming as streaming
|
||||||
|
entry = types.SimpleNamespace(result="yes", event=threading.Event())
|
||||||
|
cancel = threading.Event()
|
||||||
|
threading.Timer(0.05, entry.event.set).start()
|
||||||
|
response, expired = streaming._await_clarify_response(entry, 0, cancel)
|
||||||
|
self.assertEqual(response, "yes")
|
||||||
|
self.assertFalse(expired)
|
||||||
|
|
||||||
|
def test_await_clarify_unlimited_cancel_exits(self):
|
||||||
|
"""Cancellation still unblocks an unlimited wait promptly."""
|
||||||
|
import api.streaming as streaming
|
||||||
|
entry = types.SimpleNamespace(result=None, event=threading.Event())
|
||||||
|
cancel = threading.Event()
|
||||||
|
cancel.set()
|
||||||
|
started = time.monotonic()
|
||||||
|
response, expired = streaming._await_clarify_response(entry, 0, cancel)
|
||||||
|
self.assertTrue(expired)
|
||||||
|
self.assertLess(time.monotonic() - started, 1.0)
|
||||||
|
|
||||||
|
def test_await_clarify_finite_times_out(self):
|
||||||
|
"""A finite timeout still expires without a response (fail-safe kept)."""
|
||||||
|
import api.streaming as streaming
|
||||||
|
entry = types.SimpleNamespace(result=None, event=threading.Event())
|
||||||
|
response, expired = streaming._await_clarify_response(entry, 0.05, threading.Event())
|
||||||
|
self.assertTrue(expired)
|
||||||
|
|
||||||
|
def test_await_clarify_finite_resolves(self):
|
||||||
|
"""A finite timeout with a user response resolves normally."""
|
||||||
|
import api.streaming as streaming
|
||||||
|
entry = types.SimpleNamespace(result="ok", event=threading.Event())
|
||||||
|
entry.event.set()
|
||||||
|
response, expired = streaming._await_clarify_response(entry, 300, threading.Event())
|
||||||
|
self.assertEqual(response, "ok")
|
||||||
|
self.assertFalse(expired)
|
||||||
|
|
||||||
|
def test_clarify_metadata_unlimited_has_no_expiry(self):
|
||||||
|
"""timeout_seconds <= 0 must not fall back to the 120s default or set an expires_at."""
|
||||||
|
from api.clarify import _with_timeout_metadata
|
||||||
|
item = _with_timeout_metadata({"timeout_seconds": 0, "requested_at": 1234.0})
|
||||||
|
self.assertEqual(item["timeout_seconds"], 0)
|
||||||
|
self.assertEqual(item["expires_at"], 0)
|
||||||
|
# Finite timeouts still get an expires_at.
|
||||||
|
item2 = _with_timeout_metadata({"timeout_seconds": 300, "requested_at": 1234.0})
|
||||||
|
self.assertEqual(item2["expires_at"], 1534.0)
|
||||||
|
|
||||||
|
def _capture_clarify_timeout(self, config):
|
||||||
|
"""Drive _run_agent_streaming with a fake agent that invokes the clarify
|
||||||
|
callback, capturing the payload the bridge submits. Returns a dict with
|
||||||
|
``clarify_result`` and ``payloads``."""
|
||||||
import api.streaming as streaming
|
import api.streaming as streaming
|
||||||
|
|
||||||
captured = {}
|
captured = {}
|
||||||
@@ -572,7 +680,8 @@ class TestRuntimeRouteInjection(unittest.TestCase):
|
|||||||
with mock.patch.object(streaming, "get_session", return_value=fake_session), \
|
with mock.patch.object(streaming, "get_session", return_value=fake_session), \
|
||||||
mock.patch.object(streaming, "_get_ai_agent", return_value=CapturingAgent), \
|
mock.patch.object(streaming, "_get_ai_agent", return_value=CapturingAgent), \
|
||||||
mock.patch.object(streaming, "resolve_model_provider", return_value=("gpt-5.4", "openai-codex", None)), \
|
mock.patch.object(streaming, "resolve_model_provider", return_value=("gpt-5.4", "openai-codex", None)), \
|
||||||
mock.patch.object(streaming, "get_config", return_value={"clarify": {"timeout": 300}}), \
|
mock.patch.object(streaming, "get_config", return_value=config), \
|
||||||
|
mock.patch("api.config.get_config_for_profile_home", return_value=config), \
|
||||||
mock.patch("api.config._resolve_cli_toolsets", return_value=[]), \
|
mock.patch("api.config._resolve_cli_toolsets", return_value=[]), \
|
||||||
mock.patch("api.clarify.submit_pending", side_effect=fake_submit_pending), \
|
mock.patch("api.clarify.submit_pending", side_effect=fake_submit_pending), \
|
||||||
mock.patch.dict(sys.modules, {
|
mock.patch.dict(sys.modules, {
|
||||||
@@ -589,9 +698,7 @@ class TestRuntimeRouteInjection(unittest.TestCase):
|
|||||||
stream_id=fake_stream_id,
|
stream_id=fake_stream_id,
|
||||||
)
|
)
|
||||||
|
|
||||||
self.assertEqual(captured["clarify_result"], "selected")
|
return {"clarify_result": captured.get("clarify_result"), "payloads": submit_payloads}
|
||||||
self.assertEqual(len(submit_payloads), 1)
|
|
||||||
self.assertEqual(submit_payloads[0]["timeout_seconds"], 300)
|
|
||||||
|
|
||||||
|
|
||||||
class TestSessionDBAST(unittest.TestCase):
|
class TestSessionDBAST(unittest.TestCase):
|
||||||
|
|||||||
Reference in new issue
Block a user