diff --git a/hermes_cli/partial_compress.py b/hermes_cli/partial_compress.py index d097fd0bc2..e523e71f12 100644 --- a/hermes_cli/partial_compress.py +++ b/hermes_cli/partial_compress.py @@ -3,43 +3,34 @@ * **Role alternation.** The compressed head ends with summary/handoff content (assistant- or user- role, possibly a trailing todo snapshot). The verbatim tail must begin with a ``user`` message so the rejoined history keeps the user↔assistant alternation that providers validate. - -* **No silent context mutation.** This is a manual, user-invoked action. It rotates the session -exactly like ``/compress`` does (via the caller), so the prompt-cache reset is explicit and -expected, not silent. +* **No silent context mutation.** Manual, user-invoked; rotates the session exactly like +``/compress`` (via the caller), so the prompt-cache reset is explicit and expected. """ from __future__ import annotations from typing import Any, Dict, List, Optional, Tuple -#: Default number of recent exchanges to preserve verbatim when the user -#: runs ``/compress here`` without an explicit count. +#: Recent exchanges preserved verbatim when ``/compress here`` has no explicit count. DEFAULT_KEEP_LAST = 2 -#: Hard ceiling so a fat-fingered ``/compress here 9999`` doesn't turn -#: into a no-op surprise — clamp instead. +#: Hard ceiling so a fat-fingered ``/compress here 9999`` clamps instead of no-op'ing. MAX_KEEP_LAST = 100 def parse_partial_compress_args( raw_args: str, ) -> Tuple[bool, int, Optional[str]]: - """Parse the argument string after ``/compress``. + """Parse the argument string after ``/compress`` into ``(partial, keep_last, focus_topic)``. - Anything else is treated as a focus topic for the existing full ``/compress `` behavior. - - * ``partial`` — True when a boundary-aware form was requested. * ``keep_last`` — exchanges to - preserve verbatim (only meaningful when ``partial`` is True). * ``focus_topic`` — focus string - for full compression, or None. + ``here [N]`` / ``up to here [N]`` / ``--keep N`` / ``-k N`` / ``--keep=N`` select the + boundary-aware form; anything else is a focus topic for full ``/compress ``. """ text = (raw_args or "").strip() if not text: return False, DEFAULT_KEEP_LAST, None lowered = text.lower() - - # Normalize the "up to here" alias to "here". if lowered.startswith("up to here"): lowered = lowered[len("up to ") :] text = text[len("up to ") :] @@ -47,30 +38,24 @@ def parse_partial_compress_args( tokens = lowered.split() head = tokens[0] if tokens else "" - # Form: here [N] if head == "here": keep = _coerce_keep(tokens[1]) if len(tokens) >= 2 else DEFAULT_KEEP_LAST return True, keep, None - - # Form: --keep N (or --keep=N) if head in ("--keep", "-k") and len(tokens) >= 2: return True, _coerce_keep(tokens[1]), None if head.startswith("--keep="): return True, _coerce_keep(head.split("=", 1)[1]), None - # Otherwise: full compression with this as the focus topic. return False, DEFAULT_KEEP_LAST, text or None def extract_compress_flags(raw_args: str) -> Tuple[str, bool, bool]: - """Strip ``--preview``/``--dry-run``/``--aggressive`` from the ``/compress`` argument string. + """Strip ``--preview``/``--dry-run``/``--aggressive`` (anywhere in the string) and return + ``(remaining_args, preview, aggressive_requested)``. - Flags may appear anywhere alongside the positional forms (``here [N]``, ``--keep N``, focus - topic); the remainder is what :func:`parse_partial_compress_args` should see. Returns - ``(remaining_args, preview, aggressive_requested)``. ``preview`` (``--preview``/``--dry- - run``) means report what WOULD be compressed and change nothing. No surface implements an - LLM-free hard-truncate path, so callers surface "not supported" for ``--aggressive`` instead - of treating it as a focus topic. + ``preview`` means report what WOULD be compressed and change nothing. No surface implements an + LLM-free hard-truncate path, so callers surface "not supported" for ``--aggressive`` instead of + treating it as a focus topic. """ preview = False aggressive = False @@ -95,20 +80,16 @@ def summarize_compress_preview( ) -> Dict[str, Any]: """Build the ``/compress --preview`` report — pure, no side effects. - Shared by the CLI and the gateway slash handler so both surfaces report the same numbers the - real run would use. Returns ``head_count``/``tail_count``/``lines`` (ready-to-print - strings). + Shared by the CLI and the gateway slash handler so both report the numbers the real run would + use. Returns ``head_count``/``tail_count``/``total``/``partial``/``lines`` (ready to print). """ total = len(history) - head = list(history) - tail: List[Dict[str, Any]] = [] - effective_partial = partial + head, tail = list(history), [] + effective_partial = False if partial: head, tail = split_history_for_partial_compress(history, keep_last) - if not tail: - # Same degenerate-split fallback the real run applies. - effective_partial = False - head, tail = list(history), [] + # Same degenerate-split fallback the real run applies. + effective_partial = bool(tail) lines = [ "Preview — no changes made.", @@ -151,47 +132,31 @@ def split_history_for_partial_compress( history: List[Dict[str, Any]], keep_last: int, ) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]]]: - """Split ``history`` into ``(head, tail)`` for partial compression. + """Split ``history`` into ``(head, tail)``: head is summarized, tail is the last ``keep_last`` + exchanges kept verbatim. - ``head`` is summarized; ``tail`` is the last ``keep_last`` exchanges kept verbatim. Exchanges are counted by ``user`` messages so the tail always starts on a user turn and - rejoining ``compressed_head + tail`` keeps user↔assistant alternation valid. Returns - ``(history, [])`` when the head would be empty, signaling the caller to fall back to full - compression or "nothing to do". + ``compressed_head + tail`` keeps alternation valid. Returns ``(history, [])`` when the head + would be empty (or there are no user turns), signaling the caller to fall back to full + compression rather than rotating the session for a no-op. """ - if keep_last < 1: - keep_last = 1 - - n = len(history) - if n == 0: + keep_last = max(keep_last, 1) + if not history: return [], [] - # Walk backwards collecting the indices of the most recent `keep_last` - # user-message starts. The tail begins at the earliest such index. - user_starts: List[int] = [] - for idx in range(n - 1, -1, -1): + # Walk backwards to the earliest of the most recent `keep_last` user-message starts. + boundary = None + seen = 0 + for idx in range(len(history) - 1, -1, -1): if history[idx].get("role") == "user": - user_starts.append(idx) - if len(user_starts) >= keep_last: + boundary = idx + seen += 1 + if seen >= keep_last: break - if not user_starts: - # No user turns at all (degenerate) — nothing sensible to keep - # as a "recent exchange"; treat as full compression. + if not boundary: # no user turns, or everything is in the tail return list(history), [] - - boundary = user_starts[-1] # earliest of the kept user starts - - head = history[:boundary] - tail = history[boundary:] - - # If everything is in the tail (nothing left to compress), signal the - # caller to fall back to full compression rather than producing a - # no-op that rotates the session for no benefit. - if not head: - return list(history), [] - - return head, tail + return history[:boundary], history[boundary:] def rejoin_compressed_head_and_tail( @@ -200,12 +165,11 @@ def rejoin_compressed_head_and_tail( ) -> List[Dict[str, Any]]: """Concatenate a compressed head with the verbatim tail, defending the seam's role alternation. - The compressed head normally ends on an assistant/tool turn, but the head compressor's - output shape isn't contractually guaranteed (a plugin engine could end on a user turn). If - the last head message and first tail message share a user/assistant role, the tail's first - content is folded onto the head's last message so provider role-alternation rules hold. - ``tool`` messages are left alone — consecutive tool entries are the one legal repetition - (parallel results). + The head compressor's output shape isn't contractually guaranteed (a plugin engine could end on + a user turn). If the last head message and first tail message share a user/assistant role, the + tail's first string content is folded onto the head's last message; non-string (multimodal) + content gets a minimal bridging turn instead so nothing is lost. ``tool`` messages are left + alone — consecutive tool entries are the one legal repetition (parallel results). """ if not tail: return list(compressed_head) @@ -214,30 +178,16 @@ def rejoin_compressed_head_and_tail( head = list(compressed_head) rest = list(tail) - - last = head[-1] - first = rest[0] - last_role = last.get("role") - first_role = first.get("role") + last, first = head[-1], rest[0] + last_role, first_role = last.get("role"), first.get("role") if last_role == first_role and last_role in ("user", "assistant"): - # Illegal adjacency. Merge the tail's first message text into the - # head's last message so alternation is preserved. Only string - # contents are merged inline; structured/multimodal contents fall - # back to dropping the redundant standalone (the content is - # preserved by concatenation when both are strings). last_content = last.get("content") first_content = first.get("content") if isinstance(last_content, str) and isinstance(first_content, str): - merged = dict(last) - merged["content"] = f"{last_content}\n\n{first_content}" - head[-1] = merged + head[-1] = {**last, "content": f"{last_content}\n\n{first_content}"} rest = rest[1:] else: - # Can't safely string-merge multimodal content. Insert a - # minimal bridging turn so the seam alternates rather than - # losing data. - bridge_role = "assistant" if first_role == "user" else "user" - head.append({"role": bridge_role, "content": ""}) + head.append({"role": "assistant" if first_role == "user" else "user", "content": ""}) return head + rest