diff --git a/agent/compression_facade.py b/agent/compression_facade.py index 5a6eea86c8..e90d09cbb1 100644 --- a/agent/compression_facade.py +++ b/agent/compression_facade.py @@ -18,7 +18,6 @@ logger = logging.getLogger("run_agent") def _timeout_fallback_prompt(agent, system_message: str) -> str: """Cached prompt, else a fresh build, else the raw ``system_message`` (never raises). - Resolved lazily by the timeout wrapper: an eager rebuild would raise before compress_context runs when ``_cached_system_prompt`` is unset and the builder fails.""" if cached := getattr(agent, "_cached_system_prompt", None): @@ -93,7 +92,6 @@ def _warn_commit_overrun(agent, waited: float, ceiling: float) -> None: def _sync_persisted_markers(target_messages, source_messages) -> None: """Mirror ``_DB_PERSISTED_MARKER`` stamps from the worker's snapshot onto a live list. - Matched by scoped identity; timestamp-less repeated content is ambiguous, so every scoped match is stamped. Imported UNCONDITIONALLY: a silent fallback literal would split the stamping key from the flush's and resurrect the duplicate-row bug.""" @@ -110,7 +108,6 @@ def _run_under_progress_timeout( agent, run, messages, system_message, *, active_fence, fence_registration_lock, idle_timeout, total_ceiling ): """Run ``run(fence, target_messages=snapshot)`` on the pool under the progress-aware timeout. - The pooled worker must NEVER share the caller's live transcript — a late engine after a host timeout could rewrite it. It deep-snapshots on the worker and publishes only via an ADMITTED commit; a no-op/abort returns the snapshot unchanged, so the ORIGINAL list is handed back to keep identity semantics.""" @@ -165,7 +162,6 @@ def _mirror_result_onto_live_lists(agent, result, messages, *, direct_path: bool def _rebind_caller_session_context(agent) -> None: """Propagate a rotated session id to the CALLER's thread/ContextVar (idempotent otherwise). - The worker thread rotated hermes_logging's thread-local id; post-compression tools must resolve HERMES_SESSION_ID to the child id.""" with contextlib.suppress(Exception): diff --git a/agent/conversation_compression.py b/agent/conversation_compression.py index 47888e478e..096d60b8c0 100644 --- a/agent/conversation_compression.py +++ b/agent/conversation_compression.py @@ -73,7 +73,6 @@ COMPACTION_DONE_STATUS = "✓ Context compaction complete — continuing turn... def _strip_marker_for_comparison(msgs: Any) -> Any: """Copy ``msgs`` with the ``_db_persisted`` marker removed for no-op comparison. - Live dicts carry the marker while ``compress()`` output is swept, so a raw ``==`` would misclassify an identical no-op copy as progress. Non-list inputs and non-dict entries pass through unchanged.""" from agent.context_compressor import _DB_PERSISTED_MARKER @@ -134,7 +133,6 @@ ROUTINE_COMPRESSION_STATUS_SAMPLES = ( def is_compaction_progress_status(text: str | None) -> bool: """True for in-progress auto-compaction lifecycle lines (not the done edge). - The gateway re-tags matches as ``kind="compacting"`` for the whole pause; matching only the marker left idle/preflight/retry lines looking hung. ``COMPACTION_DONE_STATUS`` is emitted as ``kind="compacted"`` and must not match here.""" @@ -156,7 +154,6 @@ def is_compaction_progress_status(text: str | None) -> bool: def _refresh_agent_tool_definitions(agent) -> bool: """Rebuild agent.tools at the compaction commit boundary. - Forever-sessions never restart, so this is the only moment config changes reach the frozen dynamic tool schemas; the prompt cache is already invalid. Delegates to refresh_agent_mcp_tools in content_aware mode (swaps on schema CONTENT change). Returns True when tools were added. Never raises.""" @@ -184,7 +181,6 @@ _COMPRESSOR_COOLDOWN_STATE_FIELDS = ("_summary_failure_cooldown_until", "_last_s def _snapshot_compressor_attempt_state(compressor: Any) -> dict[str, Any]: """Copy only the mutable bookkeeping owned by one compression attempt. - The allow-list avoids copying clients, DB handles, locks and plugin resources; missing fields are ignored so legacy/third-party compressors keep working.""" try: @@ -205,7 +201,6 @@ _COMPRESSOR_ATTEMPT_LOCK = threading.Lock() def _claim_compressor_attempt(compressor: Any) -> int: """Claim the compressor for a new attempt; return its monotonic generation id. - Restores or cancelled-check mutations stamped with an OLDER generation no-op, so a detached late attempt cannot clobber its successor's state.""" with _COMPRESSOR_ATTEMPT_LOCK: @@ -237,7 +232,6 @@ def _install_compression_cancelled_check(compressor: Any, check: Any, generation def _clear_compression_cancelled_check_if_owner(compressor: Any, generation: int) -> bool: """Clear the cancellation consult only when *generation* installed it. - Prevents a detached late primary from tearing down a newer fallback's callback. Returns True when cleared.""" with _COMPRESSOR_ATTEMPT_LOCK: owner = getattr(compressor, "_compression_cancelled_check_owner", None) @@ -253,7 +247,6 @@ def _rollback_durable_cooldown( compressor: Any, snapshot: dict[str, Any], authoritative: Optional[bool], durable_state: Optional[dict[str, Any]] ) -> None: """Recreate/clear the durable cooldown row from the attempt snapshot. - Authoritative captures use the exact raw-row restore API (verifies read-back, propagates failure); the legacy path re-derives deadline/error best-effort.""" session_db = vars(compressor).get("_session_db") @@ -284,7 +277,6 @@ def _restore_compressor_attempt_state( durable_cooldown_state: Optional[dict[str, Any]] = None, attempt_generation: Optional[int] = None, ) -> None: """Restore the per-attempt snapshot after a pre-commit hard cancel. - A restore stamped with a stale ``attempt_generation`` no-ops so a timed-out primary's late unwind cannot roll back state owned by the fallback attempt.""" if attempt_generation is not None and not _compressor_attempt_is_current(compressor, attempt_generation): @@ -322,7 +314,6 @@ def _capture_authoritative_cooldown_under_lease( compressor: Any, attempt_snapshot: dict[str, Any] ) -> tuple[Optional[bool], Optional[dict[str, Any]]]: """Refresh and snapshot built-in durable cooldown state under the lease. - Third-party compressors are not invoked: plugin code must not run under the lease. Returns ``False`` on durable read failure (rollback must not mistake unknown state for an empty row) and ``None`` when the legacy API is absent.""" @@ -359,7 +350,6 @@ def _capture_authoritative_cooldown_under_lease( class CompressionCommitFence: """Fence timeout cancellation against post-summary session mutation. - The sync worker thread cannot be killed; the fence makes the commit boundary deterministic: cancellation wins before mutation starts, or waits for an already-started commit to finish completely.""" @@ -607,7 +597,6 @@ _CANCELLED_WORKER_TEARDOWN_GRACE_SECONDS = 5.0 def _join_cancelled_worker(future: Any, grace_seconds: float) -> bool: """Best-effort bounded join of a fence-cancelled compression worker. - Returns True when the future settled within ``grace_seconds`` (thread provably exited); False for a still-running worker, which the caller must treat as an orphan behind the poison fence.""" grace = 0.0 @@ -673,7 +662,6 @@ def _get_compress_timeout_executor(): def resolve_context_compression_timeouts(compression_cfg: Optional[dict] = None) -> Tuple[float, float]: """Return ``(idle_timeout_seconds, total_ceiling_seconds)``. - ``idle_timeout_seconds <= 0`` disables the progress-aware wrapper. The ceiling is clamped to at least one idle window when the idle budget is positive.""" idle = DEFAULT_CONTEXT_TIMEOUT_SECONDS @@ -703,7 +691,6 @@ def compression_attempt_stalled( *, commit_fence: Optional[CompressionCommitFence], started_at: float, idle_timeout_seconds: Optional[float] = None ) -> bool: """Return whether a pre-commit cancel landed after the stall window. - An early ``/stop`` stays cooldown-neutral; an interrupt after the inactivity budget counts as a stall so the next automatic turn does not blindly retry.""" idle = idle_timeout_seconds @@ -742,7 +729,6 @@ def _record_stall_interrupted_backoff( approx_tokens: Optional[int], ) -> bool: """Persist a stall-interrupted cooldown after snapshot restore. - Must run *after* ``_restore_compressor_attempt_state`` so rollback cannot wipe the new row. Returns True when the backoff was recorded.""" if not compression_attempt_stalled(commit_fence=commit_fence, started_at=started_at): @@ -766,7 +752,6 @@ def _record_stall_interrupted_backoff( def resolve_compression_fallback_route() -> Optional[dict]: """Return the first usable ``auxiliary.compression.fallback_chain`` entry. - The aux client applies the chain only from its exception handler, so a silent stall never reaches it; this pins the route onto one bounded retry instead. Only the first complete entry: if it errors, the aux client's own exception path walks the rest. ``None`` when none is usable (skip compression).""" @@ -814,7 +799,6 @@ def _retry_compression_on_fallback_chain( new_fence: Optional[Callable[[], CompressionCommitFence]] = None, ) -> Optional[Tuple[list, str]]: """Re-run an aborted compression once with the summary route pinned. - Returns ``(messages, system_prompt)`` on real compression, else ``None`` and the caller degrades as before. The entry's ``timeout`` sets the idle window. Re-runs the whole worker, so pre-compression callbacks must be idempotent.""" @@ -956,7 +940,6 @@ def _release_cancelled_worker( future: Any, fence: CompressionCommitFence, *, total_exhausted: bool, ceiling: float ) -> None: """Idle-timeout unwind: free the worker's durable lease via the holder-qualified hook. - Total-ceiling only: bounded grace for the worker to exit (it checks the fence between provider phases; an uninterruptible call is orphaned). Idle-stall skips the join: the worker is hung, the fallback needs a prompt return, the fence guards.""" @@ -985,7 +968,6 @@ def run_compress_context_with_progress_timeout( new_fence: Optional[Callable[[], CompressionCommitFence]] = None, ) -> Tuple[list, str]: """Run ``worker(fence)`` under a sync progress-aware (idle + ceiling) timeout. - Budgets bound the PRE-commit phase only: an admitted commit always completes (overrun logged, surfaced once via ``on_commit_overrun``). A pre-commit cancel returns ``(messages, system_prompt_fallback)`` (lazy callable), detaching the worker; a stall first retries the chain once on ``new_fence``, then on_timeout""" @@ -1118,7 +1100,6 @@ def _checkpoint_blocked(reason: str) -> CompressionCheckpointUnavailable: def _lock_api_is_absent_on_session_db(lock_db: Any) -> bool: """Whether the live in-memory SessionDB class structurally predates locks. - Only the exact old ``hermes_state.SessionDB`` class (hot-reload skew) may fail open; proxies, lookalikes, non-callables and descriptor failures fail closed.""" try: @@ -1209,7 +1190,6 @@ def _restore_messages_snapshot(messages: list, snapshot: Optional[list]) -> None def _restore_prune_rearm_tokens(compressor: Any, snapshot: dict) -> None: """Restore ONLY the prune runway from the attempt snapshot. - compress() zeroes it in memory while the durable copy only clears on a successful commit; a kept transcript keeps its cached prefix, and 0 would let the next prune break that cache.""" if "_proactive_prune_rearm_tokens" in snapshot: @@ -1218,7 +1198,6 @@ def _restore_prune_rearm_tokens(compressor: Any, snapshot: dict) -> None: def compression_skipped_due_to_lock(agent: Any) -> bool: """Type-pinned read of the per-session lock-skip signal. - ``agent._compression_skipped_due_to_lock`` is a holder string or ``True`` when a pass no-oped because the lock was held, ``None`` otherwise. Pinning avoids MagicMock auto-attributes hijacking mocked agents into the lock-skip branch.""" @@ -1245,7 +1224,6 @@ def _get_context_compression_timeout_state( def _set_context_compression_timeout_outcome(agent: Any, timed_out: bool) -> None: """Write this thread's owned-compression timeout outcome. - The ``agent._last_compression_timed_out`` mirror stays authoritative for minimal agent doubles that do not support ``vars()``.""" lock, state = _get_context_compression_timeout_state(agent, create=True) or (None, None) @@ -1269,7 +1247,6 @@ def mark_context_compression_timed_out(agent: Any) -> None: def context_compression_timed_out(agent: Any) -> bool: """Return whether this thread's owned compression hit its host timeout. - Thread-local so overlapping automatic/manual entrypoints cannot hide each other's timeout; attribute fallback for minimal doubles; reads type-pinned.""" locked_state = _get_context_compression_timeout_state(agent, create=False) @@ -1283,7 +1260,6 @@ def context_compression_timed_out(agent: Any) -> bool: def _automatic_compression_gate_blocks(agent: Any, bypass_cooldown: bool, *, include_cooldown: bool = True) -> bool: """Refresh durable guards, then evaluate the compressor's automatic breaker gate. - ``bypass_cooldown`` ignores the cooldown when the gate accepts ``ignore_cooldown`` (engines predating it get the legacy no-argument call). When blocked, the transient-block signal is published for automatic-path consumers. """ @@ -1304,7 +1280,6 @@ def _automatic_compression_gate_blocks(agent: Any, bypass_cooldown: bool, *, inc def compression_blocked_transiently(agent: Any) -> bool: """Type-pinned read of the transient-block signal. - Set when an automatic pass no-ops on a TRANSIENT guard (summary-failure cooldown or structural backoff). Consumers must defer, not count it toward ``compression_exhausted``, or an overflow auto-reset wipes a session that was merely cooling down. The permanent ``ineffective`` breaker never sets it.""" @@ -1314,7 +1289,6 @@ def compression_blocked_transiently(agent: Any) -> bool: def _mark_compression_blocked_transient(agent: Any, compressor: Any) -> None: """Publish the transient-block signal when the active guard is transient. - Classification comes from ``_compression_block_reason``: ``cooldown:*`` and ``structural_backoff:*`` are transient; ``ineffective`` stays unmarked.""" reason_fn = getattr(compressor, "_compression_block_reason", None) @@ -1350,7 +1324,6 @@ def _adopt_live_compression_child( agent: Any, session_db: Any, parent_session_id: str ) -> Optional[List[Dict[str, Any]]]: """Move a stale compression contender onto the live continuation tip. - Resolve and load first, then mutate the agent, so ambiguous lineage or an unreadable handoff fails closed. Uses the transitive ``get_compression_tip`` walk; a tip is adopted only while its row is still live.""" resolver = getattr(type(session_db), "get_compression_tip", None) @@ -1447,7 +1420,6 @@ def recover_rotated_compression_session(agent: Any) -> Optional[List[Dict[str, A def _compression_lock_holder(agent: Any) -> str: """Build a unique lock holder id: ``pid:tid:agent-instance:uuid``. - pid+tid tell crashed holders apart in diagnostics; instance id and per-acquire uuid disambiguate co-resident agents on one thread or pooled compressions.""" return f"pid={os.getpid()}:tid={threading.get_ident()}:agent={id(agent):x}:nonce={uuid.uuid4().hex[:8]}" @@ -1458,7 +1430,6 @@ def _supported_compression_kwargs( memory_context: str, bypass_cooldown: bool = False, ) -> dict: """Return only compression kwargs accepted by an engine callable. - Inspecting first keeps older plugin signatures compatible without catching ``TypeError`` and running a stateful compressor twice.""" candidates = {"current_tokens": current_tokens, "focus_topic": focus_topic, "force": force} @@ -1551,7 +1522,6 @@ class _CompressionActivityHeartbeat: def _direct_messages_for_pre_compress_memory(messages: Any) -> list[dict[str, Any]]: """Return direct user/assistant evidence safe for memory checkpointing. - Summaries, tool rows and system messages are omitted; assistant prose is kept with ``tool_calls`` stripped, and pure tool-call wrappers are dropped.""" # Deferred import: context_compressor → turn_context → this module would form an import cycle. @@ -1632,7 +1602,6 @@ def _lower_threshold_to_aux_context( agent: Any, *, aux_model: str, aux_context: int, aux_provider: str, aux_base_url: str ) -> None: """Lower the live threshold to the aux model's window and tell the user how to fix config. - The summariser sends one user prompt (no system/tools), so threshold == aux_context is safe. tail_token_budget and threshold_percent are kept in lockstep (as update_model does) or the 1.5x tail ceiling exceeds the trigger and re-fires.""" @@ -1700,7 +1669,6 @@ def _lower_threshold_to_aux_context( def check_compression_model_feasibility(agent: Any) -> None: """Warn at session start if the aux compression context is below the threshold. - Called from ``AIAgent.__init__`` (CLI sees it via ``_vprint``); the gateway wires ``status_callback`` later, so ``replay_compression_warning`` resends it.""" if not agent.compression_enabled: @@ -1790,7 +1758,6 @@ def check_compression_model_feasibility(agent: Any) -> None: def replay_compression_warning(agent: Any) -> None: """Re-send the stored compression warning through ``status_callback``. - Called once at the start of the first ``run_conversation()``, when the gateway callback (absent during ``__init__``) is finally wired.""" msg = getattr(agent, "_compression_warning", None) @@ -1803,7 +1770,6 @@ def conversation_history_after_compression( agent: Any, messages: list, previous_history: Optional[list] = None ) -> Optional[list]: """Return the correct flush baseline after a compression boundary. - Session rotation returns ``None`` so the child gets the full compacted list. In-place compaction returns a shallow copy of the already-persisted rows (else the identity flush re-appends them). Aborted/no-op attempts keep the baseline: marking all persisted drops unflushed turns; clearing re-appends rows.""" @@ -1845,7 +1811,6 @@ _SYNTHETIC_USER_FLAGS = ( def _is_real_user_message(message: Any) -> bool: """Distinguish human intent from user-role runtime scaffolding. - A compaction summary flipped to ``role="user"`` for alternation is scaffolding and must not short-circuit anchor restoration.""" if not isinstance(message, dict) or message.get("role") != "user": @@ -1874,7 +1839,6 @@ def _steer_markers() -> Tuple[str, str]: def _message_contains_busy_steer(message: Any) -> bool: """Return whether *message* carries a busy-steer marker. - Steer follow-ups live as markers inside ``role=tool`` results, so they carry user intent that ``_is_real_user_message`` alone would miss.""" text = _message_text(message) @@ -1910,7 +1874,6 @@ def _extract_steer_text_from_message(message: Any) -> Optional[str]: def _compressed_has_busy_steer(messages: list) -> bool: """Whether *messages* already carries a steer marker in a ``role=tool`` row. - Only tool rows count, so a summary merely quoting the marker text is not mistaken for live intent.""" return any( isinstance(msg, dict) and msg.get("role") == "tool" and _message_contains_busy_steer(msg) for msg in messages @@ -1919,7 +1882,6 @@ def _compressed_has_busy_steer(messages: list) -> bool: def _strip_stale_todo_snapshot(content: Any) -> Any: """Remove a previously merged todo-snapshot block from message content. - Snapshots are appended to the trailing user turn, so a surviving header is stale; stripping before re-injection prevents accumulation across boundaries.""" from tools.todo_tool import TODO_INJECTION_HEADER @@ -1943,7 +1905,6 @@ def _strip_stale_todo_snapshot(content: Any) -> Any: def _todo_snapshot_is_only_content(content: Any, stripped: Any) -> bool: """Return whether stripping the snapshot leaves no structured content. - Text snapshots trail a string; structured ones occupy their own text part, so only an empty remainder proves the row was scaffolding alone. Text extraction is deliberately not used: image, audio and other non-text parts must survive.""" @@ -1969,7 +1930,6 @@ _PRUNED_SKILL_RELOAD_NOTICE_HEADER = "[Skills pruned during compression — relo def _pruned_skill_reload_notice(compressed: list) -> str: """Reload notice for skills whose bodies were pruned, or ``""``. - Scans ``[SKILL_PRUNED: ...]`` markers in the post-compression transcript; first-seen order, deduplicated, capped at ``_MAX_PRUNED_SKILL_MARKERS``.""" from agent.context_compressor import _MAX_PRUNED_SKILL_MARKERS, _extract_pruned_skill_names @@ -1996,7 +1956,6 @@ def _pruned_skill_reload_notice(compressed: list) -> str: def _merge_anchor_into_user_message(target: dict, anchor: dict) -> None: """Fold the human anchor into an existing user-role scaffolding turn. - Used only when any insertion would create consecutive user turns. Anchor text leads, scaffolding follows, and synthetic flags are cleared.""" anchor_content = anchor.get("content") @@ -2088,7 +2047,6 @@ def _messages_match_scoped_identity(left: Any, right: Any) -> bool: def _stamp_scoped_twins(targets: list, source: dict, *, exact_counts_stamped: bool = False) -> None: """Stamp ``_db_persisted`` on every unstamped scoped twin of ``source`` in ``targets``. - Exact-timestamp twins are preferred: when the source carries a timestamp and any exact twin was stamped (or, with ``exact_counts_stamped``, merely exists), the broad scoped pass is skipped so a content-equal old duplicate is left alone.""" @@ -2172,7 +2130,6 @@ def finalize_context_engine_compression_notification(agent: Any, *, committed: b class _CompactionLifecycle: """Owns the one-shot terminal edge of the compaction status lifecycle. - ``commit_status`` is rebound to "committed" only on success and read at ``complete()`` time, so abort paths keep the terminal edge suppressed.""" @@ -2195,7 +2152,6 @@ class _CompactionLifecycle: class _CompressionLease: """The per-attempt durable compression lock plus its lifecycle plumbing. - ``holder`` is None when no durable lock is owned (legacy DB, no session db); ``watermark`` is MAX(id) of active rows at lease start (None = archive everything, no concurrent-tail preservation this cycle).""" @@ -2276,7 +2232,6 @@ class _CompressionLease: def _resolve_lock_api(lock_db: Any) -> Tuple[Any, Optional[Exception]]: """Return ``(try_acquire_compression_lock, lookup_error)`` for ``lock_db``. - ``(None, None)`` = no db or legacy SessionDB without the lock API (fail open); ``(None, exc)`` = lookup itself failed (caller fails closed).""" if lock_db is None: @@ -2306,7 +2261,6 @@ def _abort_lease( def _try_acquire_durable_lock(lease: _CompressionLease, try_acquire: Any, commit_fence: Any) -> bool: """Acquire the durable lock for ``lease.holder`` and capture the start watermark. - Watermark = MAX(id) of active rows at START: appends aren't blocked during summary; later rows are concurrent tail that archive_and_compact re-sequences. Capture is safety-additive (fallback archives everything), so its failure never aborts. An acquire that raises is not version skew: fail closed and @@ -2375,7 +2329,6 @@ def _acquire_compression_lease( system_message: str, approx_tokens: Optional[int], attempt_started_at: float, ) -> Tuple[Optional[_CompressionLease], Optional[str]]: """Take the per-session compression lock; ``(None, prompt)`` means sit out. - Two AIAgents sharing a session_id (e.g. background review fork) would both rotate and orphan a child. Keyed on the OLD id (what rivals read from SessionEntry). Loser sits out: messages unchanged, caller sees no-op. Only structural absence of the lock API (version skew) fails open; once resolved, any exception @@ -2449,7 +2402,6 @@ def _adopt_if_parent_rotated( agent: Any, lease: _CompressionLease, messages: list, system_message: str ) -> Optional[Tuple[list, str]]: """Sit out (or adopt the live child) when the parent was already rotated. - A late contender can take the parent lock after the winner released it and rotated; holding the lock does not prove this agent still owns a live parent. Returns the ``compress_context`` result to hand back, or None to proceed.""" @@ -2481,7 +2433,6 @@ def _adopt_if_parent_rotated( def _adopt_grown_durable_parent(agent: Any, lease: _CompressionLease, messages: list) -> Optional[list]: """Return the durable parent transcript when it outgrew the in-memory snapshot. - Rotation only (in-place never loses rows). The snapshot predates the lease: if durable grew, a writer committed a turn — ADOPT it (aborting wedged busy sessions forever). Length check only: in-memory edits of past turns are legal.""" @@ -2522,7 +2473,6 @@ def _adopt_grown_durable_parent(agent: Any, lease: _CompressionLease, messages: def _pre_compress_memory_context(agent: Any, messages: list, checkpoint_required: bool) -> str: """Provider ``on_pre_compress()`` insights to surface in the summary ("" if none). - Raw messages stay the API v1 provider contract; normalized evidence goes only to API v2+ checkpoint providers inside MemoryManager.on_pre_compress(). Raises :class:`CompressionCheckpointUnavailable` when a required checkpoint cannot be taken.""" @@ -2734,7 +2684,6 @@ def _salvage_or_refuse_grown_transcript( attempt_snapshot: dict, ) -> Tuple[Optional[list], Optional[str]]: """Anti-growth guard at the COMMIT SITE (in-place commits before the gateway can inspect). - Compares like-for-like rough estimates; on growth tries one mechanical salvage pass, else treats the attempt as a refused no-op. Returns ``(compressed, None)`` to proceed or ``(None, prompt)`` when refused (caller releases the lease).""" @@ -2797,7 +2746,6 @@ def _parent_deliberately_ended(session_db: Any, session_id: str) -> bool: def _carry_session_state_to_child(agent: Any, old_session_id: str, old_title: Any) -> None: """Migrate /goal, /heartbeat, /loop state and the title from the parent to the child. - Each lookup is a flat per-session read with no parent walk, so state would silently die at the boundary. The title is carried unchanged (renumbering per rotation made one session look like many); its provenance is read BEFORE the transfer clears the ancestor's row, then restored so an inherited auto-title stays upgradeable. @@ -2832,7 +2780,6 @@ def _publish_rotated_compaction( old_session_id: str, compressed_user_turn_outcome: str, ) -> None: """Rotate the session: flush the parent, publish the child, re-point the agent. - Flushes current-turn msgs to the OLD session, passing the durable prefix (messages[:persist idx]) so preflight, which runs before rows are marker-stamped, can't re-append them.""" current_idx = getattr(agent, "_persist_user_message_idx", None) @@ -2923,7 +2870,6 @@ def _warn_summary_or_aux_fallback(agent: Any) -> None: def _reset_read_dedup_caches(task_id: str, *, skills: bool = True) -> None: """Clear the file-read (and skill_view) repeat-read dedup caches after a boundary. - Original read content was summarized away, so a re-read must return full content, not a "file unchanged" stub. """ with contextlib.suppress(Exception): @@ -2943,7 +2889,6 @@ def _finish_compaction_boundary( task_id: str, ) -> int: """Post-commit bookkeeping: notify engines/providers/hooks, re-arm usage tracking. - Returns the rough post-compression token estimate (diagnostics only).""" # old_session_id is bound only on rotation; _boundary_parent is the id the # boundary notifications attribute prior state to (old id, or same id in-place). @@ -3037,7 +2982,6 @@ def _candidate_rejected( attempt_generation: Any, attempt_started_at: float, ) -> bool: """Reject an unusable compression candidate before any session mutation. - Order matters: compressor-reported abort, no progress, empty transcript, superseded attempt. Each branch surfaces its own warning/telemetry; the caller releases the lease and returns the input unchanged when True. """ @@ -3243,7 +3187,6 @@ def _run_summary_phase( attempt: _Attempt, ) -> _SummaryPhase: """Adopt a grown durable parent, gather memory context and run the summarizer. - A hard cancel restores the compressor snapshot + live list, records a stall backoff while the lease is still held, and aborts; any other failure releases the lease and re-raises.""" pre_msg_count = len(messages) @@ -3337,7 +3280,6 @@ class _Attempt: def _begin_compression_attempt(agent: Any, *, force: bool, defer_notification: bool) -> _Attempt: """Snapshot + claim the compressor, reset per-attempt agent signals, seed telemetry. - The claim stops a late-unwinding sibling (stall-fallback overlap) from restoring its snapshot over ours or clearing our cancellation consult. Signals are cleared at the VERY TOP, before codex/breaker early-returns, so a stale value cannot make a later no-op look like lock contention; @@ -3405,7 +3347,6 @@ def compress_context( commit_fence: Optional[CompressionCommitFence] = None, ) -> Tuple[list, str]: """Compress conversation context and split the session in SQLite. - ``force`` (manual /compress) clears the summary-failure cooldown; ``bypass_cooldown`` (provider-proven overflow) skips it once, breakers still apply. ``commit_fence`` stops a timed-out worker mutating session state. Returns ``(messages, system_prompt)``; on abort input is unchanged, NOT split.""" @@ -3583,7 +3524,6 @@ def _codex_compaction_cooldown_remaining(agent: Any) -> float: def _record_codex_compaction_failure(agent: Any, error: str) -> None: """Arm the shared compression-failure cooldown after a failed codex compaction. - The codex path returns the transcript unchanged, so without a cooldown the still-over-threshold session would retry every turn.""" from agent.context_compressor import _SUMMARY_FAILURE_COOLDOWN_SECONDS @@ -3600,7 +3540,6 @@ def _compress_context_via_codex_app_server( task_id: str = "default", force: bool = False, ) -> Tuple[list, str]: """Route compaction to Codex app-server for Codex-owned threads. - Rewriting the local transcript would not shrink the Codex thread, so Codex compacts its own thread and Hermes' transcript is left unchanged.""" _sid = getattr(agent, "session_id", None) or "none" @@ -3701,7 +3640,6 @@ def _decode_pixels(data_url: str) -> Optional[tuple]: def _shrink_data_url(url: str, *, max_dimension: int, resize_fn: Any) -> tuple: """Return ``(resized_url, unshrinkable)`` for a data URL. - ``resized_url`` is None when no rewrite applied. ``unshrinkable`` is True only when the image violated a constraint and resizing failed to satisfy that same constraint, so the caller knows a retry is pointless. The accept gate MUST use the axis that triggered the shrink: a pixel downscale can re-encode to MORE bytes @@ -3766,7 +3704,6 @@ def _source_to_data_url(source: Any) -> Optional[str]: def _write_data_url_to_source(source: dict, data_url: str) -> dict: """Return a NEW source dict carrying the re-encoded payload. - Copy-on-write: parts may be shared with the persistent history, so mutating in place would store the degraded image; the caller replaces the part.""" header, _, data = data_url.partition(",") @@ -3775,7 +3712,6 @@ def _write_data_url_to_source(source: dict, data_url: str) -> dict: def try_shrink_image_parts_in_messages(api_messages: list, *, max_dimension: int = 8000) -> bool: """Re-encode oversized native image parts to recover from image-too-large errors. - Mutates ``api_messages`` in place. Returns True if any part was replaced, False if nothing to shrink or Pillow could not help. Targets data-URL parts over 4 MB or ``max_dimension`` (Anthropic's per-side pixel cap, parsed from the rejection by the caller); http(s) image URLs are left untouched."""