From f0526e6d744be0b4ad25fafca1bae77a1e7fd346 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 12:20:47 -0700 Subject: [PATCH] refactor(run_agent): extract CompressionFacadeMixin; collapse 10 more lazy forwarders --- agent/compression_facade.py | 366 +++++++++++++++++++++++++++++ run_agent.py | 443 +----------------------------------- 2 files changed, 378 insertions(+), 431 deletions(-) create mode 100644 agent/compression_facade.py diff --git a/agent/compression_facade.py b/agent/compression_facade.py new file mode 100644 index 0000000000..3af5e4bb7a --- /dev/null +++ b/agent/compression_facade.py @@ -0,0 +1,366 @@ +"""Host-side ``AIAgent._compress_context`` wrapper. + +Publishes the commit fence ``hard_interrupt()`` reads, runs the compressor on a snapshot under the progress +timeout, mirrors ``_DB_PERSISTED_MARKER`` stamps back onto the live lists and rebinds the session context. +Extracted from ``run_agent.py``; every method resolves through ``AIAgent``'s MRO unchanged. +""" +import logging +import copy +import threading + +from agent.session_activity import ActivityProvenance + +# Same logger name as the origin module so log records / caplog filters are unchanged. +logger = logging.getLogger("run_agent") + + +class CompressionFacadeMixin: + """``_compress_context`` (see module docstring).""" + + def _compress_context( + self, + messages: list, + system_message: str, + *, + approx_tokens: int = None, + task_id: str = "default", + focus_topic: str = None, + force: bool = False, + bypass_cooldown: bool = False, + defer_context_engine_notification: bool = False, + commit_fence=None, + ) -> tuple: + """Forwarder — see ``agent.conversation_compression.compress_context``. + + ``force=True`` (manual /compress) bypasses the summary-failure cooldown; ``bypass_cooldown=True`` + (provider-proven overflow recovery) runs one real attempt while the cooldown stays armed. + """ + # Per-attempt timeout signal for turn-start preflight and in-loop consumers (#98424): a stalled + # compression must not be mistaken for a structural no-op. Thread-local + per-agent lock (#98741). + from agent.conversation_compression import ( + CompressionCommitFence, + compress_context, + mark_context_compression_timed_out, + reset_context_compression_timeout_outcome, + resolve_context_compression_timeouts, + run_compress_context_with_progress_timeout, + ) + reset_context_compression_timeout_outcome(self) + from agent.portal_tags import ( + get_affinity_scope, + get_conversation_context, + reset_affinity_scope, + reset_conversation_context, + set_affinity_scope, + set_conversation_context, + ) + from agent.prompt_cache_scope import declared_conversation_scope_safe + # Out-of-turn compaction (/compact, gateway /compress, partial head compression) runs outside + # run_conversation's ambient scope; publish the root as a fallback so the summarizer's call carries + # the conversation tag. No-op for in-turn callers. + token = None + if get_conversation_context() is None: + root = self._conversation_root_id() + if root: + token = set_conversation_context(root) + # Same fallback for the ROUTING scope, only when the host declared one (pre-#96811 fallback + # otherwise). + affinity_token = None + if get_affinity_scope() is None: + declared = declared_conversation_scope_safe(self) + if declared: + affinity_token = set_affinity_scope(declared) + # Every compression has a fence; hard_interrupt() uses this exact instance to serialize cancel + # admission against begin_commit(). + active_fence = commit_fence or CompressionCommitFence() + # Serialize fence publication so overlapping automatic/manual entrypoints cannot replace the + # fence of the attempt currently committing. + fence_registration_lock = vars(self).setdefault( + "_compression_commit_fence_lock", threading.RLock() + ) + with fence_registration_lock: + missing_fence = object() + previous_fence = vars(self).get( + "_active_compression_commit_fence", missing_fence + ) + self._active_compression_commit_fence = active_fence + try: + def _run(fence=None, target_messages=None): + return compress_context( + self, + target_messages if target_messages is not None else messages, + system_message, + approx_tokens=approx_tokens, task_id=task_id, + focus_topic=focus_topic, + force=force, + bypass_cooldown=bypass_cooldown, + defer_context_engine_notification=( + defer_context_engine_notification + ), + commit_fence=fence, + ) + + # Callers that already own a progress-aware wait (gateway session + # hygiene) pass commit_fence and must not be double-wrapped. + direct_path = commit_fence is not None + idle_timeout = total_ceiling = None + if not direct_path: + idle_timeout, total_ceiling = resolve_context_compression_timeouts() + if idle_timeout <= 0: + direct_path = True + + if direct_path: + result = _run(active_fence) + else: + def _snapshot_worker(fence=None): + # #76354 F3: the pooled worker must NEVER share the caller's live transcript — a late + # engine after a host timeout could rewrite it. Deep-snapshot on the worker; results + # publish only via an ADMITTED commit. + snapshot = copy.deepcopy(messages) + result_msgs, result_prompt = _run( + fence, target_messages=snapshot + ) + if result_msgs is snapshot: + # No-op/abort returned the snapshot unchanged: hand back the ORIGINAL list so + # identity-based semantics keep working. + return messages, result_prompt + return result_msgs, result_prompt + + # Resolve the fallback prompt lazily: an eager rebuild would raise before compress_context + # runs when _cached_system_prompt is unset and _build_system_prompt fails. + def _fallback_prompt(): + cached = getattr(self, "_cached_system_prompt", None) + if cached: + return cached + try: + return self._build_system_prompt(system_message) + except Exception: + logger.debug( + "compress_context timeout fallback prompt rebuild " + "failed; using raw system_message", + exc_info=True, + ) + return system_message or "" + + timeout_cause = { + "total_exhausted": False, + "progress_observed": False, + } + + def _on_timeout_cause(total_exhausted, progress_observed): + timeout_cause["total_exhausted"] = total_exhausted + timeout_cause["progress_observed"] = progress_observed + + def _on_timeout(idle, waited, since_progress): + mark_context_compression_timed_out(self) + total_exhausted = timeout_cause["total_exhausted"] + progress_observed = timeout_cause["progress_observed"] + if total_exhausted: + logger.warning( + "Context compression reached its total ceiling " + "after %.1fs (progress observed=%s); continuing " + "without compression", + waited, + progress_observed, + ) + else: + logger.warning( + "Context compression made no progress for %.1fs " + "(total wait %.1fs, ceiling %.1fs); continuing " + "without compression", + since_progress, + waited, + total_ceiling, + ) + touch = getattr(self, "_touch_activity", None) + if callable(touch): + try: + touch( + "context compression timed out", + provenance=ActivityProvenance.AGENT_COMPRESSION_TIMEOUT, + ) + except Exception: + logger.debug( + "compress_context timeout activity touch failed", + exc_info=True, + ) + # Same timeout cooldown ladder as summary-LLM timeouts + # (#62452): avoid re-burning the full idle budget every turn. + compressor = getattr(self, "context_compressor", None) + if compressor is not None: + record = getattr(compressor, "record_timeout_failure", None) + if callable(record): + try: + reason = ( + "host compress_context total ceiling " + "exhausted" + if total_exhausted + else "host compress_context timeout " + "(no summary progress)" + ) + record( + reason, + failure_kind=( + "ceiling_exhausted" + if total_exhausted + else "stalled" + ), + ) + except Exception: + logger.debug( + "failed to record compress_context timeout " + "cooldown", + exc_info=True, + ) + emit = getattr(self, "_emit_warning", None) + if callable(emit): + if total_exhausted: + progress = ( + " after summary output was observed" + if progress_observed + else "" + ) + emit( + "⚠ Context compression reached its total ceiling " + f"after {waited:.1f}s{progress}. No messages were " + "dropped — continuing without compression. Run " + "/compress to retry or /new for a clean session." + ) + else: + emit( + "⚠ Context compression timed out " + f"after {idle:.1f}s with no output from the summary " + "model. No messages were dropped — continuing " + "without compression. Run /compress to retry, /new " + "for a clean session, or check " + "auxiliary.compression." + ) + + def _on_commit_overrun(waited, ceiling): + # Commit-phase ceiling breach: the SessionDB mutation must complete, so this only surfaces + # the overrun. + 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)." + ) + + def _publish_new_fence(): + # The stall-fallback retry (#78981) needs a fence the aborted attempt cannot veto; publish + # it on the slot hard_interrupt() reads. The finally restores the caller's fence. + retry_fence = CompressionCommitFence() + with fence_registration_lock: + self._active_compression_commit_fence = retry_fence + return retry_fence + + result = run_compress_context_with_progress_timeout( + worker=_snapshot_worker, + messages=messages, + system_prompt_fallback=_fallback_prompt, + idle_timeout_seconds=idle_timeout, + total_ceiling_seconds=total_ceiling, + on_timeout=_on_timeout, + on_timeout_cause=_on_timeout_cause, + on_commit_overrun=_on_commit_overrun, + fence=active_fence, + telemetry_agent=self, + new_fence=_publish_new_fence, + ) + # Imported UNCONDITIONALLY: a silent fallback literal would split the stamping key from the + # flush's and resurrect the duplicate-row bug. + from agent.context_compressor import _DB_PERSISTED_MARKER + from agent.conversation_compression import ( + _messages_match_scoped_identity, + + ) + + def _sync_persisted_markers(target_messages, source_messages): + if not isinstance(target_messages, list) or not isinstance( + source_messages, list + ): + return + # Stamps land on the worker's snapshot first; mirror them onto the live lists by scoped + # identity. Timestamp-less repeated content is ambiguous, so every scoped match is stamped. + for source_message in source_messages: + if not ( + isinstance(source_message, dict) + and source_message.get(_DB_PERSISTED_MARKER) + ): + continue + source_timestamp = source_message.get("timestamp") + matched_exact_timestamp = False + if source_timestamp is not None: + for target_message in target_messages: + if not isinstance(target_message, dict): + continue + if target_message.get(_DB_PERSISTED_MARKER): + continue + if not _messages_match_scoped_identity( + target_message, source_message + ): + continue + if target_message.get("timestamp") != source_timestamp: + continue + target_message[_DB_PERSISTED_MARKER] = True + matched_exact_timestamp = True + if matched_exact_timestamp: + continue + for target_message in target_messages: + if not isinstance(target_message, dict): + continue + if target_message.get(_DB_PERSISTED_MARKER): + continue + if not _messages_match_scoped_identity( + target_message, source_message + ): + continue + target_message[_DB_PERSISTED_MARKER] = True + + if isinstance(result, tuple) and result: + result_messages = result[0] + if isinstance(result_messages, list): + # Direct-path callers bypass the snapshot worker but still need the post-publish mirror. + if direct_path or result_messages is not messages: + _sync_persisted_markers(messages, result_messages) + session_messages = getattr(self, "_session_messages", None) + if ( + isinstance(session_messages, list) + and session_messages is not messages + ): + # Durable-parent adoption can leave `_session_messages` on the pre-adoption list; sync + # both. + _sync_persisted_markers(session_messages, result_messages) + # The worker thread rotated hermes_logging's thread-local session id; propagate to this thread + # (#34089). + try: + from hermes_logging import set_session_context + set_session_context(self.session_id) + except Exception: + pass + # #76354 F5: rebind the session ContextVar in the CALLER's context so post-compression tools + # resolve HERMES_SESSION_ID to the child id (idempotent when no rotation happened). + try: + from gateway.session_context import set_current_session_id + if self.session_id: + set_current_session_id(self.session_id) + except Exception: + logger.debug( + "post-compression session ContextVar rebind failed", + exc_info=True, + ) + return result + finally: + with fence_registration_lock: + if previous_fence is missing_fence: + vars(self).pop("_active_compression_commit_fence", None) + else: + self._active_compression_commit_fence = previous_fence + # Restore whatever the caller had, so a compaction never leaks its + # tag into the surrounding scope. + if token is not None: + reset_conversation_context(token) + if affinity_token is not None: + reset_affinity_scope(affinity_token) diff --git a/run_agent.py b/run_agent.py index fb16873eb4..c93b588b85 100644 --- a/run_agent.py +++ b/run_agent.py @@ -185,6 +185,7 @@ from agent.session_persistence import ( # noqa: F401 # re-exported: cli/gatewa _is_ephemeral_scaffolding, _safe_session_filename_component, ) +from agent.compression_facade import CompressionFacadeMixin from agent.lazy_forward import forward as _forward, forward_static as _forward_static from agent.session_activity import ActivityProvenance from agent.model_metadata import ( @@ -311,6 +312,7 @@ class AIAgent( ActivityTrackingMixin, RateLimitCreditsMixin, SessionPersistenceMixin, + CompressionFacadeMixin, ): """AI Agent with tool calling capabilities.""" @@ -724,12 +726,7 @@ class AIAgent( _stream_diag_init = _forward_static("agent.stream_diag", "stream_diag_init") - def _stream_diag_capture_response( - self, diag: Dict[str, Any], http_response: Any - ) -> None: - """Forwarder — see ``agent.stream_diag.stream_diag_capture_response``.""" - from agent.stream_diag import stream_diag_capture_response - stream_diag_capture_response(self, diag, http_response) + _stream_diag_capture_response = _forward("agent.stream_diag", "stream_diag_capture_response") _flatten_exception_chain = _forward_static("agent.stream_diag", "flatten_exception_chain") @@ -748,38 +745,9 @@ class AIAgent( message = str(error).strip().lower() return "expected ident at line" in message - def _log_stream_retry( - self, - *, - kind: str, - error: BaseException, - attempt: int, - max_attempts: int, - mid_tool_call: bool, - diag: Optional[Dict[str, Any]] = None, - ) -> None: - """Forwarder — see ``agent.stream_diag.log_stream_retry``.""" - from agent.stream_diag import log_stream_retry - log_stream_retry( - self, kind=kind, error=error, attempt=attempt, - max_attempts=max_attempts, mid_tool_call=mid_tool_call, diag=diag, - ) + _log_stream_retry = _forward("agent.stream_diag", "log_stream_retry") - def _emit_stream_drop( - self, - *, - error: BaseException, - attempt: int, - max_attempts: int, - mid_tool_call: bool, - diag: Optional[Dict[str, Any]] = None, - ) -> None: - """Forwarder — see ``agent.stream_diag.emit_stream_drop``.""" - from agent.stream_diag import emit_stream_drop - emit_stream_drop( - self, error=error, attempt=attempt, max_attempts=max_attempts, - mid_tool_call=mid_tool_call, diag=diag, - ) + _emit_stream_drop = _forward("agent.stream_diag", "emit_stream_drop") def _emit_auxiliary_failure(self, task: str, exc: BaseException) -> None: """Surface a compact warning for failed auxiliary work.""" @@ -803,15 +771,9 @@ class AIAgent( "auth_mode": getattr(self, "auth_mode", "") or "", } - def _check_compression_model_feasibility(self) -> None: - """Forwarder — see ``agent.conversation_compression.check_compression_model_feasibility``.""" - from agent.conversation_compression import check_compression_model_feasibility - check_compression_model_feasibility(self) + _check_compression_model_feasibility = _forward("agent.conversation_compression", "check_compression_model_feasibility") - def _replay_compression_warning(self) -> None: - """Forwarder — see ``agent.conversation_compression.replay_compression_warning``.""" - from agent.conversation_compression import replay_compression_warning - replay_compression_warning(self) + _replay_compression_warning = _forward("agent.conversation_compression", "replay_compression_warning") def _is_direct_openai_url(self, base_url: str = None) -> bool: """Return True when a base URL targets OpenAI's native API.""" @@ -1158,19 +1120,7 @@ class AIAgent( _COMBINED_REVIEW_PROMPT, ) - @staticmethod - def _summarize_background_review_actions( - review_messages: List[Dict], - prior_snapshot: List[Dict], - notification_mode: str = "on", - ) -> List[str]: - """Forwarder — see ``agent.background_review.summarize_background_review_actions``.""" - from agent.background_review import summarize_background_review_actions - return summarize_background_review_actions( - review_messages, - prior_snapshot, - notification_mode=notification_mode, - ) + _summarize_background_review_actions = _forward_static("agent.background_review", "summarize_background_review_actions") def _spawn_background_review( self, @@ -1331,7 +1281,6 @@ class AIAgent( short aliases existing gateway and delegate readers use. """ from agent.session_activity import ( - ActivityProvenance, build_activity_snapshot, ) @@ -1724,10 +1673,7 @@ class AIAgent( """Check if an interrupt has been requested.""" return self._interrupt_requested - def _build_system_prompt(self, system_message: str = None) -> str: - """Forwarder — see ``agent.system_prompt.build_system_prompt``.""" - from agent.system_prompt import build_system_prompt - return build_system_prompt(self, system_message=system_message) + _build_system_prompt = _forward("agent.system_prompt", "build_system_prompt") @staticmethod def _get_tool_call_id_static(tc) -> str: @@ -1888,10 +1834,7 @@ class AIAgent( _repair_tool_call = _forward("agent.agent_runtime_helpers", "repair_tool_call") - def _invalidate_system_prompt(self): - """Forwarder — see ``agent.system_prompt.invalidate_system_prompt``.""" - from agent.system_prompt import invalidate_system_prompt - invalidate_system_prompt(self) + _invalidate_system_prompt = _forward("agent.system_prompt", "invalidate_system_prompt") @staticmethod def _deterministic_call_id(fn_name: str, arguments: str, index: int = 0) -> str: @@ -2213,18 +2156,7 @@ class AIAgent( ) return summary - def _try_shrink_image_parts_in_messages( - self, - api_messages: list, - *, - max_dimension: int = 8000, - ) -> bool: - """Forwarder — see ``agent.conversation_compression.try_shrink_image_parts_in_messages``.""" - from agent.conversation_compression import try_shrink_image_parts_in_messages - return try_shrink_image_parts_in_messages( - api_messages, - max_dimension=max_dimension, - ) + _try_shrink_image_parts_in_messages = _forward_static("agent.conversation_compression", "try_shrink_image_parts_in_messages") def _try_strip_image_parts_from_tool_messages( self, @@ -2387,10 +2319,7 @@ class AIAgent( content[-1]["cache_control"] = {"type": "ephemeral"} break - def _build_api_kwargs(self, api_messages: list, tools_for_api: Optional[list] = None) -> dict: - """Forwarder — see ``agent.chat_completion_helpers.build_api_kwargs``.""" - from agent.chat_completion_helpers import build_api_kwargs - return build_api_kwargs(self, api_messages, tools_for_api=tools_for_api) + _build_api_kwargs = _forward("agent.chat_completion_helpers", "build_api_kwargs") def _supports_reasoning_extra_body(self) -> bool: """Return True when reasoning extra_body is safe to send for this route/model. @@ -2672,354 +2601,6 @@ class AIAgent( """ return self.api_mode != "codex_responses" - def _compress_context( - self, - messages: list, - system_message: str, - *, - approx_tokens: int = None, - task_id: str = "default", - focus_topic: str = None, - force: bool = False, - bypass_cooldown: bool = False, - defer_context_engine_notification: bool = False, - commit_fence=None, - ) -> tuple: - """Forwarder — see ``agent.conversation_compression.compress_context``. - - ``force=True`` (manual /compress) bypasses the summary-failure cooldown; ``bypass_cooldown=True`` - (provider-proven overflow recovery) runs one real attempt while the cooldown stays armed. - """ - # Per-attempt timeout signal for turn-start preflight and in-loop consumers (#98424): a stalled - # compression must not be mistaken for a structural no-op. Thread-local + per-agent lock (#98741). - from agent.conversation_compression import ( - CompressionCommitFence, - compress_context, - mark_context_compression_timed_out, - reset_context_compression_timeout_outcome, - resolve_context_compression_timeouts, - run_compress_context_with_progress_timeout, - ) - reset_context_compression_timeout_outcome(self) - from agent.portal_tags import ( - get_affinity_scope, - get_conversation_context, - reset_affinity_scope, - reset_conversation_context, - set_affinity_scope, - set_conversation_context, - ) - from agent.prompt_cache_scope import declared_conversation_scope_safe - # Out-of-turn compaction (/compact, gateway /compress, partial head compression) runs outside - # run_conversation's ambient scope; publish the root as a fallback so the summarizer's call carries - # the conversation tag. No-op for in-turn callers. - token = None - if get_conversation_context() is None: - root = self._conversation_root_id() - if root: - token = set_conversation_context(root) - # Same fallback for the ROUTING scope, only when the host declared one (pre-#96811 fallback - # otherwise). - affinity_token = None - if get_affinity_scope() is None: - declared = declared_conversation_scope_safe(self) - if declared: - affinity_token = set_affinity_scope(declared) - # Every compression has a fence; hard_interrupt() uses this exact instance to serialize cancel - # admission against begin_commit(). - active_fence = commit_fence or CompressionCommitFence() - # Serialize fence publication so overlapping automatic/manual entrypoints cannot replace the - # fence of the attempt currently committing. - fence_registration_lock = vars(self).setdefault( - "_compression_commit_fence_lock", threading.RLock() - ) - with fence_registration_lock: - missing_fence = object() - previous_fence = vars(self).get( - "_active_compression_commit_fence", missing_fence - ) - self._active_compression_commit_fence = active_fence - try: - def _run(fence=None, target_messages=None): - return compress_context( - self, - target_messages if target_messages is not None else messages, - system_message, - approx_tokens=approx_tokens, task_id=task_id, - focus_topic=focus_topic, - force=force, - bypass_cooldown=bypass_cooldown, - defer_context_engine_notification=( - defer_context_engine_notification - ), - commit_fence=fence, - ) - - # Callers that already own a progress-aware wait (gateway session - # hygiene) pass commit_fence and must not be double-wrapped. - direct_path = commit_fence is not None - idle_timeout = total_ceiling = None - if not direct_path: - idle_timeout, total_ceiling = resolve_context_compression_timeouts() - if idle_timeout <= 0: - direct_path = True - - if direct_path: - result = _run(active_fence) - else: - def _snapshot_worker(fence=None): - # #76354 F3: the pooled worker must NEVER share the caller's live transcript — a late - # engine after a host timeout could rewrite it. Deep-snapshot on the worker; results - # publish only via an ADMITTED commit. - snapshot = copy.deepcopy(messages) - result_msgs, result_prompt = _run( - fence, target_messages=snapshot - ) - if result_msgs is snapshot: - # No-op/abort returned the snapshot unchanged: hand back the ORIGINAL list so - # identity-based semantics keep working. - return messages, result_prompt - return result_msgs, result_prompt - - # Resolve the fallback prompt lazily: an eager rebuild would raise before compress_context - # runs when _cached_system_prompt is unset and _build_system_prompt fails. - def _fallback_prompt(): - cached = getattr(self, "_cached_system_prompt", None) - if cached: - return cached - try: - return self._build_system_prompt(system_message) - except Exception: - logger.debug( - "compress_context timeout fallback prompt rebuild " - "failed; using raw system_message", - exc_info=True, - ) - return system_message or "" - - timeout_cause = { - "total_exhausted": False, - "progress_observed": False, - } - - def _on_timeout_cause(total_exhausted, progress_observed): - timeout_cause["total_exhausted"] = total_exhausted - timeout_cause["progress_observed"] = progress_observed - - def _on_timeout(idle, waited, since_progress): - mark_context_compression_timed_out(self) - total_exhausted = timeout_cause["total_exhausted"] - progress_observed = timeout_cause["progress_observed"] - if total_exhausted: - logger.warning( - "Context compression reached its total ceiling " - "after %.1fs (progress observed=%s); continuing " - "without compression", - waited, - progress_observed, - ) - else: - logger.warning( - "Context compression made no progress for %.1fs " - "(total wait %.1fs, ceiling %.1fs); continuing " - "without compression", - since_progress, - waited, - total_ceiling, - ) - touch = getattr(self, "_touch_activity", None) - if callable(touch): - try: - touch( - "context compression timed out", - provenance=ActivityProvenance.AGENT_COMPRESSION_TIMEOUT, - ) - except Exception: - logger.debug( - "compress_context timeout activity touch failed", - exc_info=True, - ) - # Same timeout cooldown ladder as summary-LLM timeouts - # (#62452): avoid re-burning the full idle budget every turn. - compressor = getattr(self, "context_compressor", None) - if compressor is not None: - record = getattr(compressor, "record_timeout_failure", None) - if callable(record): - try: - reason = ( - "host compress_context total ceiling " - "exhausted" - if total_exhausted - else "host compress_context timeout " - "(no summary progress)" - ) - record( - reason, - failure_kind=( - "ceiling_exhausted" - if total_exhausted - else "stalled" - ), - ) - except Exception: - logger.debug( - "failed to record compress_context timeout " - "cooldown", - exc_info=True, - ) - emit = getattr(self, "_emit_warning", None) - if callable(emit): - if total_exhausted: - progress = ( - " after summary output was observed" - if progress_observed - else "" - ) - emit( - "⚠ Context compression reached its total ceiling " - f"after {waited:.1f}s{progress}. No messages were " - "dropped — continuing without compression. Run " - "/compress to retry or /new for a clean session." - ) - else: - emit( - "⚠ Context compression timed out " - f"after {idle:.1f}s with no output from the summary " - "model. No messages were dropped — continuing " - "without compression. Run /compress to retry, /new " - "for a clean session, or check " - "auxiliary.compression." - ) - - def _on_commit_overrun(waited, ceiling): - # Commit-phase ceiling breach: the SessionDB mutation must complete, so this only surfaces - # the overrun. - 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)." - ) - - def _publish_new_fence(): - # The stall-fallback retry (#78981) needs a fence the aborted attempt cannot veto; publish - # it on the slot hard_interrupt() reads. The finally restores the caller's fence. - retry_fence = CompressionCommitFence() - with fence_registration_lock: - self._active_compression_commit_fence = retry_fence - return retry_fence - - result = run_compress_context_with_progress_timeout( - worker=_snapshot_worker, - messages=messages, - system_prompt_fallback=_fallback_prompt, - idle_timeout_seconds=idle_timeout, - total_ceiling_seconds=total_ceiling, - on_timeout=_on_timeout, - on_timeout_cause=_on_timeout_cause, - on_commit_overrun=_on_commit_overrun, - fence=active_fence, - telemetry_agent=self, - new_fence=_publish_new_fence, - ) - # Imported UNCONDITIONALLY: a silent fallback literal would split the stamping key from the - # flush's and resurrect the duplicate-row bug. - from agent.context_compressor import _DB_PERSISTED_MARKER - from agent.conversation_compression import ( - _messages_match_scoped_identity, - - ) - - def _sync_persisted_markers(target_messages, source_messages): - if not isinstance(target_messages, list) or not isinstance( - source_messages, list - ): - return - # Stamps land on the worker's snapshot first; mirror them onto the live lists by scoped - # identity. Timestamp-less repeated content is ambiguous, so every scoped match is stamped. - for source_message in source_messages: - if not ( - isinstance(source_message, dict) - and source_message.get(_DB_PERSISTED_MARKER) - ): - continue - source_timestamp = source_message.get("timestamp") - matched_exact_timestamp = False - if source_timestamp is not None: - for target_message in target_messages: - if not isinstance(target_message, dict): - continue - if target_message.get(_DB_PERSISTED_MARKER): - continue - if not _messages_match_scoped_identity( - target_message, source_message - ): - continue - if target_message.get("timestamp") != source_timestamp: - continue - target_message[_DB_PERSISTED_MARKER] = True - matched_exact_timestamp = True - if matched_exact_timestamp: - continue - for target_message in target_messages: - if not isinstance(target_message, dict): - continue - if target_message.get(_DB_PERSISTED_MARKER): - continue - if not _messages_match_scoped_identity( - target_message, source_message - ): - continue - target_message[_DB_PERSISTED_MARKER] = True - - if isinstance(result, tuple) and result: - result_messages = result[0] - if isinstance(result_messages, list): - # Direct-path callers bypass the snapshot worker but still need the post-publish mirror. - if direct_path or result_messages is not messages: - _sync_persisted_markers(messages, result_messages) - session_messages = getattr(self, "_session_messages", None) - if ( - isinstance(session_messages, list) - and session_messages is not messages - ): - # Durable-parent adoption can leave `_session_messages` on the pre-adoption list; sync - # both. - _sync_persisted_markers(session_messages, result_messages) - # The worker thread rotated hermes_logging's thread-local session id; propagate to this thread - # (#34089). - try: - from hermes_logging import set_session_context - set_session_context(self.session_id) - except Exception: - pass - # #76354 F5: rebind the session ContextVar in the CALLER's context so post-compression tools - # resolve HERMES_SESSION_ID to the child id (idempotent when no rotation happened). - try: - from gateway.session_context import set_current_session_id - if self.session_id: - set_current_session_id(self.session_id) - except Exception: - logger.debug( - "post-compression session ContextVar rebind failed", - exc_info=True, - ) - return result - finally: - with fence_registration_lock: - if previous_fence is missing_fence: - vars(self).pop("_active_compression_commit_fence", None) - else: - self._active_compression_commit_fence = previous_fence - # Restore whatever the caller had, so a compaction never leaks its - # tag into the surrounding scope. - if token is not None: - reset_conversation_context(token) - if affinity_token is not None: - reset_affinity_scope(affinity_token) - def _set_tool_guardrail_halt(self, decision: ToolGuardrailDecision) -> None: """Record the first guardrail decision that should stop this turn.""" if decision.should_halt and self._tool_guardrail_halt_decision is None: