From 2fb0aa1c0e8d5dc0cda22111f2020ea63ee0cfa4 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Sat, 1 Aug 2026 11:39:17 -0700 Subject: [PATCH] fix(agent): enforce bounded, surfaced commit-phase waits past context_total_ceiling_seconds MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The post-begin_commit() waiter previously called unbounded future.result(), so the advertised compression.context_total_ceiling_seconds was silently unenforced for commit-phase hangs. The commit still must complete (abandoning an in-flight SessionDB mutation would diverge live messages from durable state), but the wait is now bounded in increments against the remaining ceiling: on ceiling breach the overrun is logged (WARNING escalating to ERROR), surfaced once through the user-visible warning channel via the new on_commit_overrun callback (wired to _emit_warning in run_agent.py), and the host keeps waiting in bounded slices until the commit finishes. Documented guarantee (config comment + docs, en/zh): summary phase bounded by the ceiling; commit phase logged + surfaced if it exceeds it — never silently hung, never abandoned mid-commit. Test updated to assert the surfacing fires (previously accepted a silent over-ceiling wait); adds coverage that a raising overrun callback cannot break the commit wait. --- agent/conversation_compression.py | 78 ++++++++++--- hermes_cli/config_defaults.py | 12 +- run_agent.py | 15 +++ .../test_compress_context_progress_timeout.py | 109 +++++++++++++++--- website/docs/user-guide/configuration.md | 4 +- .../current/user-guide/configuration.md | 4 +- 6 files changed, 187 insertions(+), 35 deletions(-) diff --git a/agent/conversation_compression.py b/agent/conversation_compression.py index ed1f187838..15861bbb23 100644 --- a/agent/conversation_compression.py +++ b/agent/conversation_compression.py @@ -521,6 +521,13 @@ DEFAULT_CONTEXT_TOTAL_CEILING_SECONDS = 600.0 _compress_timeout_executor = None _compress_timeout_executor_lock = threading.Lock() +# Commit-phase overrun wait slice: once an in-flight SessionDB commit runs +# past the total ceiling, keep waiting in bounded increments of this size so +# every overrun window produces a fresh (escalating) log line instead of one +# silent unbounded future.result(). Clamped down to the ceiling for tiny test +# ceilings so overrun reporting stays observable at test timescales. +_COMMIT_OVERRUN_WAIT_SLICE_SECONDS = 30.0 + def _get_compress_timeout_executor(): """Return the process-wide compress-timeout DaemonThreadPoolExecutor.""" @@ -593,6 +600,7 @@ def run_compress_context_with_progress_timeout( idle_timeout_seconds: float, total_ceiling_seconds: float, on_timeout: Optional[Callable[[float, float, float], None]] = None, + on_commit_overrun: Optional[Callable[[float, float], None]] = None, ) -> Tuple[list, str]: """Run ``worker(fence)`` under a sync progress-aware timeout. @@ -610,10 +618,15 @@ def run_compress_context_with_progress_timeout( the **pre-commit** wait only — the summary / stream phase before :meth:`CompressionCommitFence.begin_commit`. Once the worker holds the commit fence, SessionDB mutation is already in flight and cannot be safely - abandoned without risking transcript divergence; the caller therefore waits - for ``future.result()`` with no additional host ceiling (same contract as - gateway session-hygiene after a lost cancel race). A hung commit can still - stall the turn; that is a SessionDB / I/O failure mode outside this wrapper. + abandoned without risking transcript divergence; the commit is therefore + always allowed to complete. The commit-phase wait is still *bounded in + increments* against the remaining total ceiling: if the commit runs past + ``total_ceiling_seconds``, the overrun is logged loudly (escalating from + WARNING to ERROR on repeat) and surfaced once via ``on_commit_overrun``, + while the host keeps waiting in bounded slices until the commit finishes. + The documented guarantee is: **summary phase bounded by the ceiling; + commit phase logged + surfaced if it exceeds it** (never silently hung, + never abandoned mid-commit). ``system_prompt_fallback`` may be a string or a zero-arg callable resolved only on the timeout path, so successful compression never pays for (or @@ -676,16 +689,53 @@ def run_compress_context_with_progress_timeout( if not cancelled: # Pre-commit ceiling already elapsed, but begin_commit() won the race. # Waiting is intentional: SessionDB mutation cannot be fence-cancelled. - waited = time.monotonic() - wait_started - if waited >= ceiling: - logger.warning( - "Context compression crossed the commit boundary after the " - "pre-commit ceiling (waited %.1fs, ceiling %.1fs); waiting for " - "SessionDB commit to finish before continuing", - waited, - ceiling, - ) - return future.result() + # The wait is bounded in increments against the remaining ceiling: a + # commit that overruns total_ceiling_seconds is logged loudly and + # surfaced once (on_commit_overrun), then waited on in bounded slices + # with escalating log level until it completes. Guarantee: summary + # phase bounded by ceiling; commit phase logged + surfaced if it + # exceeds it — never silently hung, never abandoned mid-commit. + overrun_surfaced = False + overrun_reports = 0 + while True: + waited = time.monotonic() - wait_started + remaining = ceiling - waited + if remaining <= 0: + # Ceiling breached while the commit is in flight. Wait in + # bounded increments so each overrun window is visible in + # logs rather than one silent unbounded block. + remaining = min( + _COMMIT_OVERRUN_WAIT_SLICE_SECONDS, + max(ceiling, 0.05), + ) + overrun_reports += 1 + log = logger.warning if overrun_reports <= 2 else logger.error + log( + "Context compression SessionDB commit still running " + "%.1fs past the total ceiling (waited %.1fs, ceiling " + "%.1fs); commit cannot be abandoned mid-flight — " + "continuing to wait (check SessionDB health if this " + "persists)", + waited - ceiling, + waited, + ceiling, + ) + if not overrun_surfaced and on_commit_overrun is not None: + overrun_surfaced = True + try: + on_commit_overrun(waited, ceiling) + except Exception: + logger.debug( + "compress_context commit-overrun callback failed", + exc_info=True, + ) + try: + return future.result(timeout=remaining) + except concurrent.futures.TimeoutError: + # Fence progress (commit-phase touch_progress) is informative + # only — the commit must complete regardless; loop and + # re-report with the updated overrun window. + continue waited = time.monotonic() - wait_started since_progress = fence.seconds_since_progress() diff --git a/hermes_cli/config_defaults.py b/hermes_cli/config_defaults.py index f0ff6505e1..834263a3c5 100644 --- a/hermes_cli/config_defaults.py +++ b/hermes_cli/config_defaults.py @@ -651,8 +651,16 @@ DEFAULT_CONFIG = { # in-agent compress_context wait (summary / # stream phase) even while tokens are still # moving. Clamped to >= context_timeout_seconds - # when the idle budget is > 0. Does NOT bound - # an already-started SessionDB commit fence. + # when the idle budget is > 0. Guarantee: + # the summary phase is bounded by this + # ceiling; an already-started SessionDB + # commit is never abandoned mid-flight — + # if the commit itself runs past the + # ceiling it is logged (WARNING, then + # ERROR) and surfaced to the user via the + # warning channel while the host keeps + # waiting in bounded increments for the + # commit to finish. "protect_first_n": 3, # non-system head messages always preserved # verbatim, in ADDITION to the system prompt # (which is always implicitly protected). Set to diff --git a/run_agent.py b/run_agent.py index fe52149a7e..f7285ae7fa 100644 --- a/run_agent.py +++ b/run_agent.py @@ -7199,6 +7199,20 @@ class AIAgent: "session, or check auxiliary.compression." ) + def _on_commit_overrun(waited, ceiling): + # Commit-phase ceiling breach: the SessionDB mutation is in + # flight and must complete (abandoning it mid-commit would + # diverge live messages from durable session state), so this + # only surfaces the overrun — it never cancels the commit. + emit = getattr(self, "_emit_warning", None) + if callable(emit): + emit( + "⚠ Context compression commit is taking unusually " + f"long ({waited:.0f}s, ceiling {ceiling:.0f}s). " + "Waiting for it to finish safely — if this persists, " + "check SessionDB health (disk / lock contention)." + ) + result = run_compress_context_with_progress_timeout( worker=_run, messages=messages, @@ -7206,6 +7220,7 @@ class AIAgent: idle_timeout_seconds=idle_timeout, total_ceiling_seconds=total_ceiling, on_timeout=_on_timeout, + on_commit_overrun=_on_commit_overrun, ) # compress_context ran on a daemon pool worker thread; the session # id rotation updated hermes_logging._session_context (a diff --git a/tests/agent/test_compress_context_progress_timeout.py b/tests/agent/test_compress_context_progress_timeout.py index 5d7488a8e3..5f7a5eb897 100644 --- a/tests/agent/test_compress_context_progress_timeout.py +++ b/tests/agent/test_compress_context_progress_timeout.py @@ -149,13 +149,19 @@ class TestRunCompressContextWithProgressTimeout: assert result_prompt == "committed" def test_never_finishing_commit_waits_past_pre_commit_ceiling(self): - """Once begin_commit() wins, the host waits without a ceiling. + """Once begin_commit() wins, the commit is waited on to completion — + but NOT silently. - context_total_ceiling_seconds only bounds the pre-commit (summary) - phase. A hung SessionDB commit cannot be fence-cancelled; returning - early would diverge live messages from durable session state. This - pins that contract so docs and the wrapper stay aligned. + context_total_ceiling_seconds bounds the pre-commit (summary) phase. + A hung SessionDB commit cannot be fence-cancelled; returning early + would diverge live messages from durable session state. The guarantee + is: summary phase bounded by ceiling; commit phase logged + surfaced + (on_commit_overrun + escalating log) if it exceeds it. This pins both + halves: the waiter blocks past the ceiling AND the overrun is loudly + reported, never silent. """ + import logging + original = [{"role": "user", "content": "a"}] compressed = [{"role": "assistant", "content": "late-commit"}] entered = threading.Event() @@ -165,7 +171,7 @@ class TestRunCompressContextWithProgressTimeout: assert fence.begin_commit() entered.set() try: - assert release.wait(timeout=2) + assert release.wait(timeout=5) return (compressed, "committed-late") finally: fence.finish_commit() @@ -173,6 +179,7 @@ class TestRunCompressContextWithProgressTimeout: ceiling = 0.05 started = time.monotonic() done = {} + overruns = [] def run(): done["result"] = run_compress_context_with_progress_timeout( @@ -181,21 +188,93 @@ class TestRunCompressContextWithProgressTimeout: system_prompt_fallback="fallback", idle_timeout_seconds=ceiling, total_ceiling_seconds=ceiling, + on_commit_overrun=lambda waited, ceil: overruns.append( + (waited, ceil) + ), ) - t = threading.Thread(target=run, name="commit-hang-waiter") - t.start() - assert entered.wait(timeout=1) - # Still blocked past the pre-commit ceiling while commit holds the fence. - time.sleep(ceiling + 0.15) - assert t.is_alive(), "waiter must block on an in-flight commit past ceiling" - release.set() - t.join(timeout=2) - assert not t.is_alive() + records = [] + + class _Capture(logging.Handler): + def emit(self, record): + records.append(record) + + comp_logger = logging.getLogger("agent.conversation_compression") + handler = _Capture(level=logging.WARNING) + comp_logger.addHandler(handler) + try: + t = threading.Thread(target=run, name="commit-hang-waiter") + t.start() + assert entered.wait(timeout=1) + # Still blocked past the pre-commit ceiling while commit holds + # the fence. + time.sleep(ceiling + 0.25) + assert t.is_alive(), ( + "waiter must block on an in-flight commit past ceiling" + ) + release.set() + t.join(timeout=5) + assert not t.is_alive() + finally: + comp_logger.removeHandler(handler) + waited = time.monotonic() - started assert waited >= ceiling + 0.1 assert done["result"][0] == compressed assert done["result"][1] == "committed-late" + # The over-ceiling commit wait must NOT be silent: the overrun + # callback fires exactly once and a WARNING+ log line reports the + # in-flight commit running past the ceiling. + assert len(overruns) == 1, overruns + assert overruns[0][1] == pytest.approx(ceiling) + assert overruns[0][0] >= ceiling + overrun_logs = [ + r + for r in records + if r.levelno >= logging.WARNING + and "past the total ceiling" in r.getMessage() + ] + assert overrun_logs, ( + "expected a WARNING+ log surfacing the commit-phase ceiling " + f"overrun; got: {[r.getMessage() for r in records]}" + ) + + def test_commit_overrun_callback_failure_does_not_break_wait(self): + """A raising on_commit_overrun callback must not abort the commit wait.""" + original = [{"role": "user", "content": "a"}] + compressed = [{"role": "assistant", "content": "ok"}] + release = threading.Event() + + def worker(fence: CompressionCommitFence): + assert fence.begin_commit() + try: + assert release.wait(timeout=5) + return (compressed, "done") + finally: + fence.finish_commit() + + def boom(waited, ceiling): + raise RuntimeError("callback exploded") + + done = {} + + def run(): + done["result"] = run_compress_context_with_progress_timeout( + worker=worker, + messages=original, + system_prompt_fallback="fallback", + idle_timeout_seconds=0.05, + total_ceiling_seconds=0.05, + on_commit_overrun=boom, + ) + + t = threading.Thread(target=run) + t.start() + time.sleep(0.3) + release.set() + t.join(timeout=5) + assert not t.is_alive() + assert done["result"] == (compressed, "done") def test_rejects_non_positive_idle(self): with pytest.raises(ValueError): diff --git a/website/docs/user-guide/configuration.md b/website/docs/user-guide/configuration.md index be9ab99b0b..a65f8e88ed 100644 --- a/website/docs/user-guide/configuration.md +++ b/website/docs/user-guide/configuration.md @@ -796,7 +796,7 @@ compression: hygiene_total_ceiling_seconds: 600 # Absolute cap on the hygiene wait even while tokens are still streaming hygiene_failure_cooldown_seconds: 300 # Skip repeated failed hygiene attempts for this session context_timeout_seconds: 120 # Inactivity budget for in-agent compress_context (loop /compress / preflight) — see below - context_total_ceiling_seconds: 600 # Absolute cap on the *pre-commit* in-agent compress_context wait even while tokens are still streaming (does not bound an already-started SessionDB commit) + context_total_ceiling_seconds: 600 # Absolute cap on the *pre-commit* in-agent compress_context wait even while tokens are still streaming (an already-started SessionDB commit is never abandoned; overruns are logged + surfaced) proactive_prune_tokens: 0 # Opt-in tokens trigger for the no-LLM tool-result prune (0 = off; see below) proactive_prune_min_result_chars: 8000 # Prune's summarize pass only touches tool results larger than this (clamped >= 200) proactive_prune_min_reclaim_tokens: 4096 # Prune only commits when it reclaims at least this many tokens (0 = commit any) @@ -825,7 +825,7 @@ Older configs with `compression.summary_model`, `compression.summary_provider`, `context_timeout_seconds` (default `120`) is the same **inactivity budget** for in-agent `compress_context` — the conversation loop, preflight compaction, and manual `/compress` — so a hung summary model cannot stall a session indefinitely. Streamed summary tokens extend the wait; only a silent worker is cut off. On timeout Hermes skips compaction, keeps the existing messages, and warns the user. Set to `0` to disable. Gateway session hygiene keeps its own `hygiene_timeout_seconds` path and is not double-wrapped. -`context_total_ceiling_seconds` (default `600`) bounds the in-agent **pre-commit** wait (summary / stream phase) even while tokens are still moving. It is clamped to at least `context_timeout_seconds`. Once the worker has entered the compression commit fence and SessionDB mutation is in flight, the host waits for that commit to finish without an additional ceiling — abandoning mid-commit would risk transcript divergence (same contract as gateway session hygiene). +`context_total_ceiling_seconds` (default `600`) bounds the in-agent **pre-commit** wait (summary / stream phase) even while tokens are still moving. It is clamped to at least `context_timeout_seconds`. The exact guarantee: **the summary phase is bounded by this ceiling; the commit phase is logged and surfaced if it exceeds it.** Once the worker has entered the compression commit fence and SessionDB mutation is in flight, the commit is never abandoned mid-flight — that would risk transcript divergence — but the wait is no longer silent: if the commit runs past the ceiling, Hermes logs the overrun (WARNING, escalating to ERROR on repeat), sends a one-shot warning through the user-visible warning channel, and keeps waiting in bounded increments until the commit completes. `protect_first_n` controls how many **non-system** head messages are pinned across every compaction. Default `3` — the opening user/assistant exchange survives every summarizer pass so the original goal stays visible. On long-running rolling-compaction sessions where the opening turn is no longer relevant, set `protect_first_n: 0` to pin nothing but the system prompt + summary + tail. The system prompt itself is always preserved regardless of this setting. diff --git a/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/configuration.md b/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/configuration.md index f3e8293361..5ec1239468 100644 --- a/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/configuration.md +++ b/website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/configuration.md @@ -601,7 +601,7 @@ compression: protect_last_n: 20 # 保持未压缩的最少最近消息数 hygiene_hard_message_limit: 5000 # Gateway 安全阀 —— 见下文 context_timeout_seconds: 120 # Agent 侧 compress_context 无进展超时（秒）—— 见下文 - context_total_ceiling_seconds: 600 # Agent 侧 compress_context 预提交等待上限（秒；不含已进入 SessionDB commit fence 的等待） + context_total_ceiling_seconds: 600 # Agent 侧 compress_context 预提交等待上限（秒；已开始的 SessionDB 提交不会被放弃，超限会记录日志并告警） # 摘要模型/provider 在 auxiliary: 下配置： auxiliary: @@ -619,7 +619,7 @@ auxiliary: `context_timeout_seconds`（默认 `120`）是 agent 侧 `compress_context`（对话循环、预检压缩、手动 `/compress`）的**无进展超时**，语义与 gateway 会话预压缩（session hygiene）的 inactivity 预算相同：摘要模型仍在流式出 token 时会延长等待；仅当完全无输出时才跳过压缩并保留原消息。设为 `0` 可关闭。Gateway 会话预压缩仍使用自己的 `hygiene_timeout_seconds`，不会被双重包装。 -`context_total_ceiling_seconds`（默认 `600`）限制即使仍有 token 推进时的 agent 侧**预提交**等待时间（摘要 / 流式阶段），并会被钳制为至少等于 `context_timeout_seconds`。一旦 worker 已进入 compression commit fence 且 SessionDB 变更正在进行，宿主会无额外上限地等待该提交完成——中途放弃会导致 transcript 分叉（与 gateway 会话 hygiene 同一契约）。 +`context_total_ceiling_seconds`（默认 `600`）限制即使仍有 token 推进时的 agent 侧**预提交**等待时间（摘要 / 流式阶段），并会被钳制为至少等于 `context_timeout_seconds`。确切保证：**摘要阶段受该上限约束；提交阶段若超出上限则记录日志并向用户告警。**一旦 worker 已进入 compression commit fence 且 SessionDB 变更正在进行，提交绝不会被中途放弃（那会导致 transcript 分叉），但等待不再是静默的：若提交超过上限，Hermes 会记录超时（WARNING，重复时升级为 ERROR），通过用户可见的警告通道发送一次性提醒，并以有界增量继续等待直到提交完成。 :::tip Gateway 热重载压缩和上下文长度 从最近的版本开始，在运行中的 gateway 上编辑 `config.yaml` 中的 `model.context_length` 或任何 `compression.*` 键将在下一条消息时生效 —— 无需 gateway 重启、`/reset` 或会话轮换。缓存的 agent 签名包含这些键，因此 gateway 在检测到更改时会透明地重建 agent。API 密钥和工具/技能配置仍需要通常的重载路径。