"""Summary production: format history for the summarizer, call it, parse the KEYS line."""
import json
from collections.abc import Callable
from datetime import datetime, timezone
from pathlib import Path
from typing import TYPE_CHECKING
from navi.config import settings
from navi.llm.base import LLMBackend, Message
from navi.memory.breadcrumbs import split_keys_line
from ..image_tokens import estimate_image_tokens
from .partition import _CRITICAL_TOOL_NAMES, _plan_compression
if TYPE_CHECKING:
from navi.profiles.base import AgentProfile
def _format_for_summary(
messages: list[Message], image_budget: int | None = None
) -> tuple[str, list[str]]:
"""
Render messages as plain text for the summarization prompt.
Returns (text, images) where images is a flat list of base64 strings
collected from all user messages. Vision-capable models will receive
the images alongside the text; non-vision models silently ignore them.
``image_budget`` caps the total cost of the collected images in tokens.
Every image in the history used to be forwarded in one user message with
only the text capped, so a session with a dozen screenshots produced a
summarization prompt that could not fit its own window — the call was
silently truncated by the backend and the summary was written as if the
history had been read. Images over the budget stay out of the call; their
per-message "[+ N image(s)]" note still tells the summarizer they existed.
"""
lines: list[str] = []
images: list[str] = []
remaining = image_budget if image_budget is None else max(0, image_budget)
i = 0
while i < len(messages):
m = messages[i]
if m.is_summary:
# Existing summary — include as-is (already compressed)
lines.append(m.content or "")
i += 1
elif m.role == "user":
if m.images:
for img in m.images:
if remaining is None:
images.append(img)
continue
cost = estimate_image_tokens(img)
if cost <= remaining:
images.append(img)
remaining -= cost
img_note = f" [+ {len(m.images)} image(s)]"
else:
img_note = ""
if m.content:
lines.append(f"User: {m.content}{img_note}")
elif img_note:
lines.append(f"User:{img_note}")
i += 1
elif m.role == "assistant" and m.tool_calls:
# Render tool calls + their results as a compact block
if m.content:
lines.append(f"Assistant: {m.content}")
for tc in m.tool_calls:
args_preview = json.dumps(tc.arguments)[:120]
lines.append(f"[Tool call: {tc.name}; arguments preview: {args_preview}]")
i += 1
while i < len(messages) and messages[i].role == "tool":
tool_msg = messages[i]
result = tool_msg.content or ""
# Critical tool results and explicit critical flag survive verbatim
# up to a larger budget, so exact errors/file contents are preserved.
critical = (
getattr(tool_msg, "is_compression_critical", False)
or tool_msg.name in _CRITICAL_TOOL_NAMES
)
if critical and len(result) <= 4000:
preview = result
elif critical:
# Over the critical budget: keep head + tail halves rather
# than collapsing to the non-critical 800-char preview —
# the all-or-nothing cliff used to throw away 98% of a
# large file read (the common navi_code shape) while a
# 3999-char result survived whole.
half = 2000
preview = result[:half] + "\n…[middle elided]…\n" + result[-half:]
else:
# Give the summarizer enough of the result to preserve
# diagnostic detail (the exact error line, a grep match,
# a key value), not just the first 300 chars — the summary
# is only as detailed as what the summarizer can see.
preview = result[:800] + ("…" if len(result) > 800 else "")
lines.append(f"[Tool result: {tool_msg.name}; preview: {preview}]")
i += 1
elif m.role == "assistant" and m.content:
lines.append(f"Assistant: {m.content}")
i += 1
else:
i += 1 # skip orphan tool messages
return "\n".join(lines), images
# Safety limit: truncate formatted input to this many characters before sending to LLM.
# Prevents the summarizer from receiving near-context-sized input it can't fit alongside output.
# Typical ollama_num_ctx is 65k–128k, so 32k of input leaves ample room for a 6k-token summary.
_MAX_SUMMARY_INPUT_CHARS = 32_000
# Share of the window the historical images may take in a summarization call
# (see _format_for_summary). The text side is capped separately, in chars.
_SUMMARY_IMAGE_BUDGET_FRAC = 0.25
# When existing summaries in to_summarize exceed this many chars combined,
# run a quick meta-summary to consolidate them before the main compression pass.
_META_SUMMARY_THRESHOLD = _MAX_SUMMARY_INPUT_CHARS // 3 # 10_666
_META_SUMMARY_SYSTEM = (
"You are condensing multiple conversation summaries into a single compact summary. "
"Preserve all key facts, decisions, file paths, config values, errors, and user preferences. "
"Eliminate redundancy between overlapping summaries. "
"Write tight bullet points. Do not include filler or meta-commentary."
)
async def _meta_summarize(
summaries: list[Message],
llm: LLMBackend,
model,
temperature: float,
) -> Message:
"""Consolidate multiple existing summary messages into one compact meta-summary."""
combined = "\n\n---\n\n".join(m.content for m in summaries)
prompt = [
Message(role="system", content=_META_SUMMARY_SYSTEM),
Message(role="user", content=combined),
]
response = await llm.complete(
prompt,
tools=None,
temperature=temperature,
model=model,
think=False,
max_tokens=1500,
)
text = (response.content or "").strip() or "(consolidated summary unavailable)"
return Message(
role="user",
content=f"[Consolidated Context Summary]\n{text}",
is_summary=True,
is_display=False,
is_context=True,
created_at=datetime.now(timezone.utc),
)
# The section list and the instructions built from it define the summary
# format every compression pass produces. `KEYS` is parsed out by
# `split_keys_line` and becomes the summary's breadcrumbs.
_SUMMARY_SECTIONS = [
("Goal", "One clear sentence describing what the user is trying to accomplish in this session. Include deadlines or constraints if stated."),
("Active Files", "Every file or directory the assistant touched, with absolute or project-relative path, status (created / modified / read / deleted), the purpose of the change, and the key function/class/symbol affected."),
("Decisions & User Preferences", "Explicit choices, architecture decisions, style preferences, or corrections stated by the user — each with a one-line rationale and what triggered it. Things the user said NOT to do."),
("Completed Work", "Concrete finished steps — step + file/function name + verification outcome (pass/fail). Be specific and complete; prefer listing over generalizing."),
("Intermediate Findings", "Key results from read/grep/log/terminal inspection that informed decisions: the fact discovered and the file/section it came from. Drop verbose bulk, keep the diagnostic line that changed understanding."),
("Pending Work / Todo", "Open tasks, in-progress items, or follow-ups that still need action."),
("Errors & Blockers", "Failures, exceptions, or unresolved issues — each with the exact short error snippet, what was tried, and the outcome. Preserve the diagnostic line verbatim."),
("Key Values", "Exact constants the assistant should remember: ports, config keys and values, versions, dependency names, important paths, IDs, exact CLI flags."),
]
_SUMMARY_TEMPLATE_INSTRUCTIONS = (
"You are summarizing a conversation history to free up context space. "
"The assistant will continue working using ONLY this summary — it will have no access "
"to the original messages. Be thorough and precise; prefer specifics and complete lists "
"over generalities — the cost of losing a fact is higher than the cost of a longer summary. "
"This summary is historical context, not a new user request.\n\n"
"Use EXACTLY the Markdown structure below. Every section must be present. "
"If a section has no relevant information, write its header and the literal word NONE. "
"Keep bullets information-dense, but do not generalize away specifics to save space. "
"Do not include greetings, filler, transitions, or meta-commentary.\n\n"
+ "\n\n".join(f"## {title}\n{desc}" for title, desc in _SUMMARY_SECTIONS)
+ "\n\n"
"Output rules:\n"
"- Preserve exact file paths, function/class/symbol names, config keys and values, "
"exact CLI flags, and short error/diagnostic snippets verbatim.\n"
"- Preserve short critical code the agent must reproduce exactly: final function/class "
"signatures, one-line fixes the user approved, key config lines. Do NOT paste long blocks, "
"patches, or full command output — keep only the short line that matters.\n"
"- Do not paraphrase values that must stay precise.\n"
"- Use Markdown headers exactly as shown.\n"
"- End with ONE line `KEYS: item, item, …` — the breadcrumbs of this history: "
"names, commands, file paths, config keys, amounts, ticket numbers, decisions and "
"links that a later turn may ask about by name. Plain items, no explanations, at "
"most 30, most important first. This line is an index, not prose; write "
"`KEYS: NONE` when nothing is worth indexing."
)
def _build_summary_system_prompt(profile: "AgentProfile | None") -> str:
"""Build the system prompt used by the summarization LLM.
Uses the profile-specific compression prompt file if configured.
"""
base = _SUMMARY_TEMPLATE_INSTRUCTIONS
extra = ""
if profile is not None and getattr(profile, "compression_prompt_file", None):
profile_dir = Path("navi/profiles") / profile.id
prompt_path = profile_dir / profile.compression_prompt_file
if prompt_path.exists():
try:
extra = "\n\n---\n\n[Profile-specific compression instructions]\n\n" + prompt_path.read_text(encoding="utf-8").strip()
except Exception:
pass
return base + extra
async def compress_context(
context: list[Message],
llm: LLMBackend,
model: "list[str] | str | None",
temperature: float,
keep_recent: int,
max_tokens: int | None = None,
keep_recent_messages: int | None = None,
profile: "AgentProfile | None" = None,
window: int | None = None,
estimate: Callable[[list[Message]], int] | None = None,
) -> tuple[list[Message], str] | None:
"""
Summarize old messages in the LLM context and return a shorter context list.
Only operates on `context` — the full display history (session.messages) is never touched.
Returns None if there is nothing to compress.
Images from old user messages are passed to the summarization model.
Vision-capable models will incorporate image descriptions into the summary;
non-vision models silently ignore the images field.
Uses the same model already loaded in memory (profile.model passed via WorkerContext) —
no model swap, no extra loading overhead.
Profile settings override global defaults when provided:
- compression_keep_recent -> keep_recent
- compression_max_tokens -> max_tokens
- compression_prompt_file -> appended to summary system prompt
Exceptions propagate to the caller (CompressionWorker catches them).
"""
effective_keep_recent = getattr(profile, "compression_keep_recent", None) or keep_recent
effective_max_tokens = getattr(profile, "compression_max_tokens", None) or max_tokens
planned = _plan_compression(
context,
effective_keep_recent,
keep_recent_messages=keep_recent_messages,
window=window,
estimate=estimate,
)
if planned is None:
return None # nothing substantial to compress
system_msgs, to_summarize, to_keep = planned
# Meta-summary: if to_summarize contains multiple existing summary messages
# that are long enough to crowd the summarizer input, consolidate them first.
summary_msgs = [m for m in to_summarize if m.is_summary]
if len(summary_msgs) > 1:
total_summary_len = sum(len(m.content or "") for m in summary_msgs)
if total_summary_len > _META_SUMMARY_THRESHOLD:
try:
meta = await _meta_summarize(
summary_msgs, llm, model, temperature
)
to_summarize = [m for m in to_summarize if not m.is_summary]
to_summarize.insert(0, meta)
except Exception:
# If meta-summary fails, continue with raw summaries as-is
pass
# Images forwarded to the summarizer share one user message with the
# formatted text. Give them a quarter of the window at most: the summary
# itself needs the rest, and a summarization call that overflows is
# silently truncated by Ollama (to half the window), which would produce a
# summary written from an arbitrary slice of the history.
image_budget = int((window or settings.ollama_num_ctx) * _SUMMARY_IMAGE_BUDGET_FRAC)
summary_text_input, images = _format_for_summary(
to_summarize, image_budget=image_budget
)
# Truncate oversized input so the summarizer LLM has room to generate output.
# Keep head + tail, not just the head: to_summarize is chronological, so a
# head-only cut silently dropped the NEWEST summarized messages — the ones
# closest to the kept window, which bridge old context and the recent turn.
if len(summary_text_input) > _MAX_SUMMARY_INPUT_CHARS:
elision = "\n…[middle elided]…\n"
head = _MAX_SUMMARY_INPUT_CHARS * 3 // 4
tail = _MAX_SUMMARY_INPUT_CHARS - head - len(elision)
summary_text_input = summary_text_input[:head] + elision + summary_text_input[-tail:]
system_prompt = _build_summary_system_prompt(profile)
prompt = [
Message(role="system", content=system_prompt),
Message(role="user", content=summary_text_input, images=images or None),
]
# think=False: compression must be fast — extended reasoning wastes context and hangs
response = await llm.complete(
prompt, tools=None, temperature=temperature, model=model, think=False, max_tokens=effective_max_tokens
)
summary_text = (response.content or "").strip() or "(summary unavailable)"
# The summarizer's trailing `KEYS:` line is the breadcrumb index for the
# messages this pass drops. It is stripped from the summary the agent reads
# (it is an index, not prose) and carried on the summary message so the
# caller can write it next to the raw chunks.
summary_text, summary_keys = split_keys_line(summary_text)
if not summary_text.strip():
summary_text = "(summary unavailable)"
summary_msg = Message(
role="user",
content=f"[Context Summary - historical context only, not a new user request]\n{summary_text}",
is_summary=True,
is_display=False,
created_at=datetime.now(timezone.utc),
metadata={"breadcrumbs": summary_keys} if summary_keys else {},
)
return system_msgs + [summary_msg] + to_keep, summary_text
