_drain_webui_process_notifications now drops (consumes, never requeues) a completion whose completed_at is older than a configurable cap (HERMES_WEBUI_STALE_COMPLETION_MAX_AGE_SECONDS, default 6h; 0 disables). Fail-open: events without completed_at (legacy agent) are never dropped. Additive (+192/-0). Closes #4029. Co-authored-by: allenliang2022 Co-authored-by: nesquena-hermes <agent@nesquena-hermes>
This commit is contained in:
4 files changed
+196
No files matched your search
@@ -7,6 +7,12 @@
|
||||
|
||||
- **Windows pytest-harness compatibility (#3664).** Hardened the test suite to run on Windows: profile-home fallback paths are path-normalized, strict POSIX file-mode (`0o600`) assertions are gated behind `os.name != "nt"` (Linux still asserts them at full strictness), the conftest cleanup handles Windows process-tree/port teardown and the Py3.12+ `shutil.rmtree` `onexc` shim, and tests that require `fork`/`fcntl` carry `@requires_fork` / `@requires_fcntl` markers (which never skip on Linux). Test-only — no runtime or app behavior change, no Linux CI behavior change. (#4254, #4255, #4256, #4257, #4259, #4263, #4266, #4274)
|
||||
|
||||
## [v0.51.455] — 2026-06-16 — Release PP (drop stale background-process completions instead of prepending them to a later turn)
|
||||
|
||||
### Fixed
|
||||
|
||||
- **A background task that finishes long after you've moved on no longer hijacks your next message.** When a `notify_on_complete` background task's completion fired after the user had left the session idle (or it sat queued), the WebUI prepended its `[IMPORTANT: … completed]` notification to whatever the user typed next — so the agent answered the *old* task and already-delivered completion notices reappeared on later turns. The notification drain now age-gates completions: one whose enqueue time is older than a configurable cap (`HERMES_WEBUI_STALE_COMPLETION_MAX_AGE_SECONDS`, default 6h; set `0` to disable) is silently consumed instead of delivered. Completions without a timestamp (older agent builds) are never dropped, so behavior is unchanged until the agent side stamps `completed_at`. (#4306, #4029)
|
||||
|
||||
## [v0.51.454] — 2026-06-16 — Release PO (complete the Japanese UI translation)
|
||||
|
||||
### Changed
|
||||
|
||||
@@ -1536,6 +1536,31 @@ def _bind_turn_session_identity(session_id: str):
|
||||
_reset_turn_session_identity(tokens)
|
||||
|
||||
|
||||
def _stale_completion_max_age_seconds() -> float:
|
||||
"""Max age (seconds) a background-process completion may sit in the queue
|
||||
before the WebUI drain treats it as stale and drops it instead of
|
||||
prepending it to the user's next turn.
|
||||
|
||||
Completions older than this are silently consumed (not requeued) so a
|
||||
notification that finally fires long after the user moved on cannot
|
||||
contaminate an unrelated later turn. See nesquena/hermes-webui#4029.
|
||||
|
||||
Configurable via HERMES_WEBUI_STALE_COMPLETION_MAX_AGE_SECONDS. A value of
|
||||
0 (or negative) disables age-gating and restores the legacy drain-all
|
||||
behavior. Defaults to 6 hours.
|
||||
"""
|
||||
raw = os.environ.get("HERMES_WEBUI_STALE_COMPLETION_MAX_AGE_SECONDS")
|
||||
if raw is not None:
|
||||
try:
|
||||
return float(raw)
|
||||
except (TypeError, ValueError):
|
||||
logger.warning(
|
||||
"Invalid HERMES_WEBUI_STALE_COMPLETION_MAX_AGE_SECONDS=%r; using default",
|
||||
raw,
|
||||
)
|
||||
return 6 * 60 * 60 # 6 hours
|
||||
|
||||
|
||||
def _format_process_notification(evt: dict) -> str:
|
||||
"""Format a completed background process notification for agent input."""
|
||||
if not isinstance(evt, dict):
|
||||
@@ -1584,6 +1609,10 @@ def _drain_webui_process_notifications(session_id: str) -> list[str]:
|
||||
if completion_queue is None:
|
||||
return []
|
||||
|
||||
# Computed once per drain (not per event): reads/validates the env cap a
|
||||
# single time so an invalid value logs at most one warning per drain.
|
||||
stale_completion_max_age = _stale_completion_max_age_seconds()
|
||||
|
||||
while True:
|
||||
try:
|
||||
evt = completion_queue.get_nowait()
|
||||
@@ -1607,6 +1636,25 @@ def _drain_webui_process_notifications(session_id: str) -> list[str]:
|
||||
skipped_events.append(evt)
|
||||
continue
|
||||
|
||||
# Age-gate stale completions: a completion that fires long after the
|
||||
# user moved on must not be prepended to an unrelated later turn
|
||||
# (nesquena/hermes-webui#4029). Drop (consume, do not requeue) any
|
||||
# completion whose enqueue time is older than the configured cap.
|
||||
# Events without a 'completed_at' (older agent builds) are never
|
||||
# dropped here, preserving backward-compatible behavior.
|
||||
if stale_completion_max_age > 0 and isinstance(evt, dict):
|
||||
completed_at = evt.get('completed_at')
|
||||
if isinstance(completed_at, (int, float)) and completed_at > 0:
|
||||
age = time.time() - completed_at
|
||||
if age > stale_completion_max_age:
|
||||
logger.info(
|
||||
"Dropping stale background-process completion for "
|
||||
"session %s (age %.0fs > cap %.0fs)",
|
||||
evt_sid, age, stale_completion_max_age,
|
||||
)
|
||||
_mark_process_completion_consumed(process_registry, evt_sid)
|
||||
continue
|
||||
|
||||
notification = _format_process_notification(evt)
|
||||
if notification:
|
||||
notifications.append(notification)
|
||||
|
||||
@@ -0,0 +1,125 @@
|
||||
"""Behavioral tests for issue #4029 stale-completion age gate.
|
||||
|
||||
These exercise the real `_drain_webui_process_notifications` against a fake
|
||||
process_registry, proving that:
|
||||
- a completion older than the cap is dropped (consumed, not requeued),
|
||||
- a fresh completion is still delivered,
|
||||
- events without `completed_at` are never dropped (backward compat),
|
||||
- the env override (incl. disable via 0) is honored.
|
||||
"""
|
||||
import importlib
|
||||
import queue
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
import types
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
def _install_fake_registry(monkeypatch, events):
|
||||
"""Build a fake `tools.process_registry` module whose `process_registry`
|
||||
exposes the surface `_drain_webui_process_notifications` touches."""
|
||||
q = queue.Queue()
|
||||
for e in events:
|
||||
q.put(e)
|
||||
|
||||
class _FakeProc:
|
||||
def __init__(self, session_key):
|
||||
self.session_key = session_key
|
||||
|
||||
class _FakeRegistry:
|
||||
def __init__(self):
|
||||
self.completion_queue = q
|
||||
self._lock = threading.RLock()
|
||||
self._completion_consumed = set()
|
||||
# map session_id -> session_key so ownership check passes
|
||||
self._owner = {}
|
||||
|
||||
def is_completion_consumed(self, sid):
|
||||
return sid in self._completion_consumed
|
||||
|
||||
def get(self, sid):
|
||||
key = self._owner.get(sid)
|
||||
return _FakeProc(key) if key is not None else None
|
||||
|
||||
reg = _FakeRegistry()
|
||||
mod = types.ModuleType("tools.process_registry")
|
||||
mod.process_registry = reg
|
||||
# ensure parent package exists
|
||||
if "tools" not in sys.modules:
|
||||
pkg = types.ModuleType("tools")
|
||||
pkg.__path__ = []
|
||||
monkeypatch.setitem(sys.modules, "tools", pkg)
|
||||
monkeypatch.setitem(sys.modules, "tools.process_registry", mod)
|
||||
return reg
|
||||
|
||||
|
||||
def _make_event(sid, completed_at, session_key="websess-1"):
|
||||
return {
|
||||
"type": "completion",
|
||||
"session_id": sid,
|
||||
"session_key": session_key,
|
||||
"command": f"cmd-{sid}",
|
||||
"exit_code": 0,
|
||||
"output": f"out-{sid}",
|
||||
**({"completed_at": completed_at} if completed_at is not None else {}),
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def streaming():
|
||||
return importlib.import_module("api.streaming")
|
||||
|
||||
|
||||
def test_stale_completion_is_dropped_and_fresh_is_delivered(streaming, monkeypatch):
|
||||
now = time.time()
|
||||
fresh = _make_event("fresh", now - 60) # 1 min old -> keep
|
||||
stale = _make_event("stale", now - 7 * 3600) # 7 h old -> drop (cap 6h)
|
||||
reg = _install_fake_registry(monkeypatch, [stale, fresh])
|
||||
reg._owner["fresh"] = "websess-1"
|
||||
reg._owner["stale"] = "websess-1"
|
||||
# default cap = 6h
|
||||
monkeypatch.delenv("HERMES_WEBUI_STALE_COMPLETION_MAX_AGE_SECONDS", raising=False)
|
||||
|
||||
out = streaming._drain_webui_process_notifications("websess-1")
|
||||
|
||||
joined = "\n".join(out)
|
||||
assert "fresh" in joined, "fresh completion should be delivered"
|
||||
assert "stale" not in joined, "stale completion must be dropped"
|
||||
# stale was consumed (won't come back), fresh consumed after delivery
|
||||
assert "stale" in reg._completion_consumed
|
||||
assert "fresh" in reg._completion_consumed
|
||||
# nothing belonging to this session is left requeued
|
||||
assert reg.completion_queue.empty()
|
||||
|
||||
|
||||
def test_event_without_completed_at_is_never_dropped(streaming, monkeypatch):
|
||||
legacy = _make_event("legacy", None) # no completed_at -> backward compat keep
|
||||
reg = _install_fake_registry(monkeypatch, [legacy])
|
||||
reg._owner["legacy"] = "websess-1"
|
||||
monkeypatch.delenv("HERMES_WEBUI_STALE_COMPLETION_MAX_AGE_SECONDS", raising=False)
|
||||
|
||||
out = streaming._drain_webui_process_notifications("websess-1")
|
||||
|
||||
assert any("legacy" in n for n in out), "legacy event (no timestamp) must be delivered"
|
||||
|
||||
|
||||
def test_env_zero_disables_age_gate(streaming, monkeypatch):
|
||||
now = time.time()
|
||||
ancient = _make_event("ancient", now - 10 * 24 * 3600) # 10 days old
|
||||
reg = _install_fake_registry(monkeypatch, [ancient])
|
||||
reg._owner["ancient"] = "websess-1"
|
||||
monkeypatch.setenv("HERMES_WEBUI_STALE_COMPLETION_MAX_AGE_SECONDS", "0")
|
||||
|
||||
out = streaming._drain_webui_process_notifications("websess-1")
|
||||
|
||||
assert any("ancient" in n for n in out), "age gate disabled -> even ancient delivered"
|
||||
|
||||
|
||||
def test_helper_reads_env_override(streaming, monkeypatch):
|
||||
monkeypatch.setenv("HERMES_WEBUI_STALE_COMPLETION_MAX_AGE_SECONDS", "120")
|
||||
assert streaming._stale_completion_max_age_seconds() == 120.0
|
||||
monkeypatch.setenv("HERMES_WEBUI_STALE_COMPLETION_MAX_AGE_SECONDS", "not-a-number")
|
||||
# invalid -> falls back to default 6h
|
||||
assert streaming._stale_completion_max_age_seconds() == 6 * 60 * 60
|
||||
@@ -28,3 +28,20 @@ def test_webui_sets_gateway_session_platform_for_background_watchers():
|
||||
assert "os.environ['HERMES_SESSION_PLATFORM'] = 'webui'" in src
|
||||
assert "old_session_platform = os.environ.get('HERMES_SESSION_PLATFORM')" in src
|
||||
assert "os.environ.pop('HERMES_SESSION_PLATFORM', None)" in src
|
||||
|
||||
|
||||
def test_webui_age_gates_stale_background_completion_events():
|
||||
"""Issue #4029: drain must drop completions older than the configured cap
|
||||
so stale notifications can't be prepended to an unrelated later turn."""
|
||||
src = Path("api/streaming.py").read_text(encoding="utf-8")
|
||||
|
||||
# The age-gate helper + its env override exist.
|
||||
assert "def _stale_completion_max_age_seconds()" in src
|
||||
assert "HERMES_WEBUI_STALE_COMPLETION_MAX_AGE_SECONDS" in src
|
||||
# The drain reads completed_at and drops over-age events without requeueing.
|
||||
assert "completed_at = evt.get('completed_at')" in src
|
||||
assert "age = time.time() - completed_at" in src
|
||||
assert "if age > stale_completion_max_age:" in src
|
||||
# Over-age events are consumed (marked), not requeued, so they vanish.
|
||||
assert "_mark_process_completion_consumed(process_registry, evt_sid)" in src
|
||||
|
||||
Reference in new issue
Block a user