Extract shared compaction helpers from duplicated compact logic

Author: beacoder

python_agent_harness/agent.py

@@ -37,7 +37,7 @@
 
 from . import config
 from .models import Message, ToolCall
-from .prompts import read_prompt_file, user_prompt_texts
+from .prompts import compact_summary, compacted_messages, user_prompt_texts
 from .token_estimator import context_window_for, estimate_payload_tokens
 from .tools.base import PendingToolResult
 
@@ -281,26 +281,18 @@ def compact(self) -> bool:
         self.session.compacting = True
         try:
             conversation = "\n\n".join(f"{m.role}: {m.text()}" for m in self.messages if m.text())
-            system = read_prompt_file("compact.md")
-            resp, _ = self.session.client.chat_sync(
-                [Message(role="user", content=conversation)],
-                system=system,
-                cancel_check=self._is_cancelled,
+            summary = compact_summary(
+                self.session.client, conversation, cancel_check=self._is_cancelled
             )
-            summary = resp.text_without_reasoning()
             if not summary:
                 return False
-            frame = config.COMPACT_HEADER + summary + config.COMPACT_SEPARATOR
             # The summary replaces the whole conversation history EXCEPT
             # the system prompt (self.system is passed separately and
             # stays untouched): it is part of the user turn, never a
             # system message.  Every real user prompt (nudges and other
             # harness-injected messages excluded) is preserved verbatim
             # after the frame, so the model keeps the actual requests.
-            self.messages = [
-                Message(role="user", content=frame.strip()),
-                *[Message(role="user", content=p) for p in prompts],
-            ]
+            self.messages = compacted_messages(summary, prompts)
             # The shared conversation now is the compacted one: mirror it
             # onto session.last_messages so the TUI (renders from it) and
             # a later manual /compact start from the summary, not the old

python_agent_harness/agent_session.py

@@ -715,8 +715,7 @@ def compact_conversation(self) -> tuple[bool, str]:
         manual command just replaces the history and waits for the next
         user message.
         """
-        from .models import Message as Msg
-        from .prompts import read_prompt_file, user_prompt_texts
+        from .prompts import compact_summary, compacted_messages, user_prompt_texts
 
         # Replacing the conversation is a new generation: invalidate any
         # worker still winding down from a cancelled run, or its
@@ -730,12 +729,9 @@ def compact_conversation(self) -> tuple[bool, str]:
         self.compacting = True
         try:
             conversation = self._conversation_text(messages)
-            system = read_prompt_file("compact.md")
-            resp, _ = self.client.chat_sync([Msg(role="user", content=conversation)], system=system)
-            summary = resp.text_without_reasoning()
+            summary = compact_summary(self.client, conversation)
             if not summary:
                 return False, "Compaction failed: empty summary."
-            frame = config.COMPACT_HEADER + summary + config.COMPACT_SEPARATOR
             # The summary is part of the user turn (the original system
             # prompt is passed separately and stays untouched), so it
             # replaces the history as a user message — matching the
@@ -744,10 +740,7 @@ def compact_conversation(self) -> tuple[bool, str]:
             # (nudges and other harness-injected messages excluded) is
             # preserved verbatim after the frame, so the model keeps
             # the actual requests.
-            self.last_messages = [
-                Msg(role="user", content=frame.strip()),
-                *[Msg(role="user", content=p) for p in user_prompt_texts(messages)],
-            ]
+            self.last_messages = compacted_messages(summary, user_prompt_texts(messages))
             self.auto_save(self.last_messages, self.system_prompt)
             self.notify("compact")
             return True, "Buffer compacted successfully."

python_agent_harness/prompts.py

@@ -4,19 +4,23 @@
 (agent/subagent/commands), strips YAML frontmatter, discovers skills
 for the {{SKILLS}} placeholder, assembles the effective system prompt
 from project context files + task-completion rules + agent prompt, and
-provides user_prompt_texts() for the compaction flow (summarize the
-conversation and rebuild the history with every user prompt preserved
-verbatim).
+provides the compaction flow helpers (summarize the conversation with
+the compact prompt and rebuild the history with every user prompt
+preserved verbatim), shared by the in-loop compaction and the manual
+/compact command.
 """
 
 from __future__ import annotations
 
 import os
 import re
 import subprocess
+from collections.abc import Callable
 from pathlib import Path
+from typing import Any
 
 from . import config
+from .models import Message
 
 
 def read_prompt_file(name: str) -> str:
@@ -450,3 +454,45 @@ def user_prompt_texts(messages: list) -> list[str]:
             continue
         prompts.append(text)
     return prompts
+
+
+def compact_summary(
+    client: Any,
+    conversation: str,
+    cancel_check: Callable[[], bool] | None = None,
+) -> str | None:
+    """Ask the model to summarize *conversation* using the compact prompt.
+
+    Shared by the in-loop compaction (``AgentLoop.compact``) and the
+    manual /compact command (``AgentSession.compact_conversation``).
+    Returns the summary text with the reasoning preamble stripped, or
+    None when the response carries no text.  Client exceptions
+    propagate to the caller, which owns the failure handling
+    (log/notify/status message).
+    """
+    system = read_prompt_file("compact.md")
+    kwargs: dict[str, Any] = {}
+    if cancel_check is not None:
+        kwargs["cancel_check"] = cancel_check
+    resp, _ = client.chat_sync(
+        [Message(role="user", content=conversation)],
+        system=system,
+        **kwargs,
+    )
+    summary = resp.text_without_reasoning()
+    return summary or None
+
+
+def compacted_messages(summary: str, prompts: list[str]) -> list[Message]:
+    """The post-compaction history: the summary frame as a user message
+    followed by every preserved user prompt, oldest first.
+
+    The summary lives in the user turn (the system prompt is passed
+    separately and stays untouched); *prompts* are the real user
+    requests (see ``user_prompt_texts``) that must survive compaction.
+    """
+    frame = (config.COMPACT_HEADER + summary + config.COMPACT_SEPARATOR).strip()
+    return [
+        Message(role="user", content=frame),
+        *[Message(role="user", content=p) for p in prompts],
+    ]