mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-24 16:54:43 +00:00
fix(agent): defer turns during compression lock contention instead of exhausting
A lock-loser compression pass returns its input unchanged, which the automatic compression sites misread as 'cannot compress further': the preflight loop armed the insufficient-progress blocker, the pre-API gate burned a shared attempt, and a lock-contended 413/overflow retried into the attempt cap and returned compression_exhausted — which the gateway answers with a full session auto-reset (#9893/#35809). A temporary concurrent-compression defer wiped the session. Consume the landed #69870 lock-skip signal on every automatic path (preflight in turn_context, pre-API pressure gate, 413 handler, overflow handler, post-tool compaction): when a pass no-ops AND the type-pinned lock-skip flag is set, refund the attempt (never count it toward the cap or the insufficient-progress blocker), and when the turn cannot proceed (provider already proved the request does not fit) end it with a soft compression_deferred result — distinct from compression_exhausted — so the gateway keeps the session intact and the next message retries after the concurrent compressor finishes. The new compression_skipped_due_to_lock() reader is type-pinned (is True or isinstance(str)) per the MagicMock auto-attribute rule, and compress_context() now also clears the signal at the very top of every attempt (per-attempt state rule, #58629/#69853) so a stale value can never make a later breaker/codex no-op look like lock contention. Salvaged from PR #49874; rebuilt on main's #69870 _compression_skipped_due_to_lock signal instead of the PR's parallel _compression_deferred_by_lock triple.
This commit is contained in:
helix4u
Teknium
parent
fdefb2d38c
commit
056a40aa4d
3 changed files with 178 additions and 27 deletions
|
|
@ -355,6 +355,24 @@ def _emit_compression_attempt_telemetry(
|
|||
logger.debug("failed to emit compression attempt telemetry: %s", exc)
|
||||
|
||||
|
||||
def compression_skipped_due_to_lock(agent: Any) -> bool:
|
||||
"""Type-pinned read of the #69870 lock-skip signal.
|
||||
|
||||
``agent._compression_skipped_due_to_lock`` is set by ``compress_context``
|
||||
when a compression pass no-ops because another path holds the per-session
|
||||
compression lock (holder string when the holder was confirmed, ``True``
|
||||
otherwise) and cleared to ``None`` at the entry of every call.
|
||||
|
||||
The read MUST be type-pinned (``is True or isinstance(x, str)``), never
|
||||
bare truthiness: MagicMock test-double agents auto-create truthy
|
||||
attributes, and a bare ``if getattr(agent, ...)`` would hijack every
|
||||
mocked agent in sibling suites into the lock-skip branch (the
|
||||
#69870 × #69840 type-ahead incident).
|
||||
"""
|
||||
_sig = getattr(agent, "_compression_skipped_due_to_lock", None)
|
||||
return _sig is True or isinstance(_sig, str)
|
||||
|
||||
|
||||
def _compression_lock_holder(agent: Any) -> str:
|
||||
"""Build a unique holder id for the lock: pid:tid:agent-instance:uuid.
|
||||
|
||||
|
|
@ -1143,6 +1161,14 @@ def compress_context(
|
|||
# boundary, so the previous flush baseline remains authoritative.
|
||||
agent._last_compression_attempt_recorded = True
|
||||
agent._last_compression_attempt_in_place = None
|
||||
# Clear the lock-skip signal at the VERY TOP, before the codex route and
|
||||
# the breaker gates below can early-return (per-attempt state rule,
|
||||
# #58630/#69853). A stale ``True``/holder value from a prior lock-skip
|
||||
# must never make a later breaker/codex no-op look like lock contention
|
||||
# to the automatic-path consumers (compression_deferred, #49874) — the
|
||||
# second clear before lock acquisition below stays for the same reason
|
||||
# it was added in #69870 and is simply idempotent now.
|
||||
agent._compression_skipped_due_to_lock = None
|
||||
|
||||
_attempt_started_at = time.monotonic()
|
||||
_attempt_id = uuid.uuid4().hex
|
||||
|
|
|
|||
|
|
@ -34,6 +34,7 @@ from agent.conversation_compression import (
|
|||
COMPRESSION_RETRY_TOKENS_STATUS_TEMPLATE,
|
||||
COMPRESSION_RETRY_TOO_LARGE_STATUS_TEMPLATE,
|
||||
PRE_API_COMPRESSION_STATUS_TEMPLATE,
|
||||
compression_skipped_due_to_lock,
|
||||
conversation_history_after_compression,
|
||||
)
|
||||
from agent.context_engine import automatic_compaction_status_message
|
||||
|
|
@ -640,6 +641,55 @@ def _content_policy_blocked_result(
|
|||
}
|
||||
|
||||
|
||||
def _compression_deferred_result(
|
||||
agent,
|
||||
messages: List[Dict],
|
||||
api_call_count: int,
|
||||
) -> Dict[str, Any]:
|
||||
"""Build the soft turn result for a lock-contended compression defer.
|
||||
|
||||
Another path (a sibling turn, a background review fork, a manual
|
||||
``/compress``) holds this session's compression lock, so every
|
||||
compression pass this turn no-oped and the request still does not fit.
|
||||
This is a TEMPORARY condition — the lock winner is actively shrinking
|
||||
the same session — so the turn must end as a soft defer
|
||||
(``compression_deferred``), never as ``compression_exhausted``: the
|
||||
gateway auto-resets (wipes) the session on exhaustion (#9893/#35809),
|
||||
which would destroy a session that the concurrent compressor is about
|
||||
to make healthy again.
|
||||
|
||||
``failed`` stays False so the gateway persists the user turn (transient
|
||||
branch) and retry-next-message semantics apply.
|
||||
"""
|
||||
holder = getattr(agent, "_compression_skipped_due_to_lock", None)
|
||||
logger.info(
|
||||
"turn deferred: compression lock held by another path "
|
||||
"(session=%s holder=%s) — not counting as compression exhaustion",
|
||||
agent.session_id or "none",
|
||||
holder if isinstance(holder, str) else "unconfirmed",
|
||||
)
|
||||
try:
|
||||
agent._flush_status_buffer()
|
||||
except Exception:
|
||||
pass
|
||||
_final = (
|
||||
"Context compression is already running for this session. "
|
||||
"Please retry in a moment — your next message will be processed "
|
||||
"once the concurrent compression finishes."
|
||||
)
|
||||
return {
|
||||
"final_response": _final,
|
||||
"messages": messages,
|
||||
"completed": False,
|
||||
"api_calls": api_call_count,
|
||||
"error": _final,
|
||||
"partial": True,
|
||||
"failed": False,
|
||||
"compression_deferred": True,
|
||||
"session_id": agent.session_id,
|
||||
}
|
||||
|
||||
|
||||
def _sync_failover_system_message(agent, api_messages, active_system_prompt):
|
||||
"""Refresh the in-flight system message after a provider failover.
|
||||
|
||||
|
|
@ -1354,36 +1404,52 @@ def run_conversation(
|
|||
if _pre_api_status:
|
||||
agent._emit_status(_pre_api_status)
|
||||
_last_preflight_pressure = request_pressure_tokens
|
||||
_pre_api_input = messages
|
||||
messages, active_system_prompt = agent._compress_context(
|
||||
messages,
|
||||
system_message,
|
||||
approx_tokens=request_pressure_tokens,
|
||||
task_id=effective_task_id,
|
||||
)
|
||||
# Reset retry/empty-response state so the compacted request
|
||||
# gets a fresh chance instead of inheriting stale recovery
|
||||
# counters from the pre-compaction history.
|
||||
agent._empty_content_retries = 0
|
||||
agent._thinking_prefill_retries = 0
|
||||
agent._last_content_with_tools = None
|
||||
agent._last_content_tools_all_housekeeping = False
|
||||
agent._mute_post_response = False
|
||||
# Re-baseline the flush cursor for the compaction mode that just
|
||||
# ran. Legacy session-rotation returns None (the child session has
|
||||
# not seen the compacted transcript, so the next flush writes it
|
||||
# whole); in-place compaction returns list(messages) because the
|
||||
# compacted rows are already persisted under the same session id —
|
||||
# leaving None there would re-append them, doubling the active
|
||||
# context and retriggering compression. Mirrors the post-response
|
||||
# and preflight compaction sites; see
|
||||
# conversation_history_after_compression().
|
||||
conversation_history = conversation_history_after_compression(
|
||||
agent, messages, conversation_history
|
||||
)
|
||||
api_call_count -= 1
|
||||
agent._api_call_count = api_call_count
|
||||
agent.iteration_budget.refund()
|
||||
continue
|
||||
if messages is _pre_api_input and compression_skipped_due_to_lock(agent):
|
||||
# #69870 lock-skip: another path holds this session's
|
||||
# compression lock, so this pass no-oped. That is a temporary
|
||||
# DEFER, not evidence about compressibility — refund the
|
||||
# attempt (it must not burn the shared overflow-recovery
|
||||
# budget toward compression_exhausted → gateway auto-reset,
|
||||
# #9893/#35809) and leave the insufficient-progress blocker
|
||||
# unarmed. Proceed with the current request: if it truly does
|
||||
# not fit, the provider's 413/overflow handler returns the
|
||||
# soft compression_deferred result with that stronger signal.
|
||||
compression_attempts -= 1
|
||||
_last_preflight_pressure = None
|
||||
if pending_moa_prepared_request is _moa_prepared_request:
|
||||
pending_moa_prepared_request = None
|
||||
else:
|
||||
# Reset retry/empty-response state so the compacted request
|
||||
# gets a fresh chance instead of inheriting stale recovery
|
||||
# counters from the pre-compaction history.
|
||||
agent._empty_content_retries = 0
|
||||
agent._thinking_prefill_retries = 0
|
||||
agent._last_content_with_tools = None
|
||||
agent._last_content_tools_all_housekeeping = False
|
||||
agent._mute_post_response = False
|
||||
# Re-baseline the flush cursor for the compaction mode that just
|
||||
# ran. Legacy session-rotation returns None (the child session has
|
||||
# not seen the compacted transcript, so the next flush writes it
|
||||
# whole); in-place compaction returns list(messages) because the
|
||||
# compacted rows are already persisted under the same session id —
|
||||
# leaving None there would re-append them, doubling the active
|
||||
# context and retriggering compression. Mirrors the post-response
|
||||
# and preflight compaction sites; see
|
||||
# conversation_history_after_compression().
|
||||
conversation_history = conversation_history_after_compression(
|
||||
agent, messages, conversation_history
|
||||
)
|
||||
api_call_count -= 1
|
||||
agent._api_call_count = api_call_count
|
||||
agent.iteration_budget.refund()
|
||||
continue
|
||||
elif (
|
||||
agent.compression_enabled
|
||||
and len(messages) > 1
|
||||
|
|
@ -3841,10 +3907,23 @@ def run_conversation(
|
|||
|
||||
original_len = len(messages)
|
||||
original_tokens = estimate_messages_tokens_rough(messages)
|
||||
_overflow_input = messages
|
||||
messages, active_system_prompt = agent._compress_context(
|
||||
messages, system_message, approx_tokens=approx_tokens,
|
||||
task_id=effective_task_id,
|
||||
)
|
||||
if messages is _overflow_input and compression_skipped_due_to_lock(agent):
|
||||
# #69870 lock-skip: the provider proved the request
|
||||
# does not fit, but this compression pass no-oped only
|
||||
# because another path holds the session's compression
|
||||
# lock. Temporary defer, not exhaustion — refund the
|
||||
# attempt and end the turn softly so the gateway does
|
||||
# NOT auto-reset the session (#9893/#35809).
|
||||
compression_attempts -= 1
|
||||
agent._persist_session(messages, conversation_history)
|
||||
return _compression_deferred_result(
|
||||
agent, messages, api_call_count
|
||||
)
|
||||
conversation_history = conversation_history_after_compression(
|
||||
agent, messages, conversation_history
|
||||
)
|
||||
|
|
@ -4082,10 +4161,23 @@ def run_conversation(
|
|||
|
||||
original_len = len(messages)
|
||||
original_tokens = estimate_messages_tokens_rough(messages)
|
||||
_overflow_input = messages
|
||||
messages, active_system_prompt = agent._compress_context(
|
||||
messages, system_message, approx_tokens=approx_tokens,
|
||||
task_id=effective_task_id,
|
||||
)
|
||||
if messages is _overflow_input and compression_skipped_due_to_lock(agent):
|
||||
# #69870 lock-skip: the provider proved the request
|
||||
# does not fit, but this compression pass no-oped only
|
||||
# because another path holds the session's compression
|
||||
# lock. Temporary defer, not exhaustion — refund the
|
||||
# attempt and end the turn softly so the gateway does
|
||||
# NOT auto-reset the session (#9893/#35809).
|
||||
compression_attempts -= 1
|
||||
agent._persist_session(messages, conversation_history)
|
||||
return _compression_deferred_result(
|
||||
agent, messages, api_call_count
|
||||
)
|
||||
conversation_history = conversation_history_after_compression(
|
||||
agent, messages, conversation_history
|
||||
)
|
||||
|
|
@ -5488,14 +5580,27 @@ def run_conversation(
|
|||
if callable(_clear_warn):
|
||||
_clear_warn()
|
||||
agent._safe_print(" ⟳ compacting context…")
|
||||
_post_tool_input = messages
|
||||
messages, active_system_prompt = agent._compress_context(
|
||||
messages, system_message,
|
||||
approx_tokens=agent.context_compressor.last_prompt_tokens,
|
||||
task_id=effective_task_id,
|
||||
)
|
||||
conversation_history = conversation_history_after_compression(
|
||||
agent, messages, conversation_history
|
||||
)
|
||||
if (
|
||||
messages is _post_tool_input
|
||||
and compression_skipped_due_to_lock(agent)
|
||||
):
|
||||
# #69870 lock-skip: this pass no-oped because another
|
||||
# path holds the session's compression lock — a
|
||||
# temporary defer, not evidence about compressibility.
|
||||
# Refund the attempt so a lock-loser tool loop does not
|
||||
# burn the shared per-turn budget toward
|
||||
# compression_exhausted (#9893/#35809).
|
||||
compression_attempts -= 1
|
||||
else:
|
||||
conversation_history = conversation_history_after_compression(
|
||||
agent, messages, conversation_history
|
||||
)
|
||||
elif agent.compression_enabled:
|
||||
# Over threshold but compression is blocked (summary-LLM
|
||||
# cooldown or anti-thrashing). Surface a deduped warning so
|
||||
|
|
|
|||
|
|
@ -34,6 +34,7 @@ from typing import Any, Dict, List, Mapping, Optional
|
|||
from agent.conversation_compression import (
|
||||
IDLE_COMPACTION_STATUS_TEMPLATE,
|
||||
PREFLIGHT_COMPRESSION_STATUS_TEMPLATE,
|
||||
compression_skipped_due_to_lock,
|
||||
conversation_history_after_compression,
|
||||
)
|
||||
from agent.context_engine import automatic_compaction_status_message
|
||||
|
|
@ -833,10 +834,29 @@ def build_turn_context(
|
|||
for _pass in range(_max_preflight_passes):
|
||||
_orig_len = len(messages)
|
||||
_orig_tokens = _preflight_tokens
|
||||
_preflight_input = messages
|
||||
messages, active_system_prompt = agent._compress_context(
|
||||
messages, system_message, approx_tokens=_preflight_tokens,
|
||||
task_id=effective_task_id,
|
||||
)
|
||||
if (
|
||||
messages is _preflight_input
|
||||
and compression_skipped_due_to_lock(agent)
|
||||
):
|
||||
# #69870 lock-skip: another path holds this session's
|
||||
# compression lock, so the pass no-oped. That is a
|
||||
# temporary DEFER, not proof the transcript cannot
|
||||
# compress — do NOT arm the insufficient-progress
|
||||
# blocker (the loop's error handlers must keep their
|
||||
# provider-proven retry budget) and stop preflight
|
||||
# passes for this turn; the lock winner is shrinking
|
||||
# the same session concurrently.
|
||||
logger.info(
|
||||
"Preflight compression deferred: compression lock "
|
||||
"held by another path (session %s)",
|
||||
agent.session_id or "none",
|
||||
)
|
||||
break
|
||||
# Re-estimate now so size-only compression (same row count,
|
||||
# lower token count — e.g. summarising tool outputs) is
|
||||
# recognised as progress instead of being misread as
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue