"""
Context compressor — summarizes old messages to stay within the token limit.
Flow:
1. Partition session messages into "to_summarize" (old turns) and "to_keep" (recent turns).
2. Call the LLM to produce a structured summary of the old messages.
3. Replace the old turns with a structured summary message (role=user, is_summary=True).
A "turn" is one user message plus all following assistant/tool messages up to the
next user message. Tool call groups (assistant + tool results) are never split.
Existing summary messages are always folded into the next compression pass.
Compression is profile-aware: AgentProfile can provide compression_keep_recent,
compression_max_tokens, and a compression_prompt_file to specialize summaries.
"""
import json
import re
import structlog
from datetime import datetime, timezone
from pathlib import Path
from collections.abc import Callable
from typing import TYPE_CHECKING
from navi.llm.base import LLMBackend, Message
from navi.config import settings
from .events import ContextCompressed
from .token_calibration import calibration_for
from .context_builder import truncate_oversized
if TYPE_CHECKING:
from navi.profiles.base import AgentProfile
log = structlog.get_logger()
_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."
)
# Tools whose full output is often needed later and should not be aggressively truncated.
_CRITICAL_TOOL_NAMES = frozenset({
"filesystem",
"code_exec",
"terminal",
"ssh_exec",
})
# Content markers that make a turn worth preserving verbatim longer.
_CRITICAL_PATTERNS = [
re.compile(r"\b(error|exception|traceback|failed|failure)\b", re.IGNORECASE),
re.compile(r"\b(user\s+said|no,\s+|don't\s+|do\s+not\s+|never\s+|instead\s+|wrong\s+|incorrect\s+|fix\s+|correct\s+)\b", re.IGNORECASE),
re.compile(r"\b(edited|modified|created|deleted|wrote|added)\s+(file|function|class|method)\b", re.IGNORECASE),
]
def should_compress(context_tokens: int, max_context_tokens: int, threshold: float) -> bool:
return context_tokens >= int(max_context_tokens * threshold)
def _turn_importance(turn: list[Message]) -> int:
"""Score a turn for adaptive keep_recent. Higher = more important to keep."""
score = 0
text = "\n".join((m.content or "") for m in turn)
lowered = text.lower()
# Strong signals: user corrections and explicit negatives
if any(w in lowered for w in ("wrong", "incorrect", "fix", "don't use", "do not use", "instead use", "change to")):
score += 3
for pattern in _CRITICAL_PATTERNS:
score += len(pattern.findall(text))
for m in turn:
if getattr(m, "is_compression_critical", False):
score += 3
if m.role == "tool" and m.name in _CRITICAL_TOOL_NAMES:
score += 1
if m.role == "user" and len((m.content or "").strip()) <= 20:
# Very short user messages are usually social/filler; deprioritize
score -= 2
return max(0, score)
def _order_like(reference: list[Message], subset: list[Message]) -> list[Message]:
"""Reorder `subset` into the order its members appear in `reference`."""
wanted = {id(m) for m in subset}
return [m for m in reference if id(m) in wanted]
def partition_messages(
messages: list[Message],
keep_recent: int,
keep_recent_messages: int | None = None,
adaptive: bool = True,
) -> tuple[list[Message], list[Message]]:
"""
Returns (to_summarize, to_keep).
Keeps the system message and the last `keep_recent` conversational turns verbatim.
Everything older goes into to_summarize.
Tool call groups (assistant + tool results) always stay together.
When adaptive=True, important turns (user corrections, errors, critical tools)
are kept longer and social/filler turns are compressed sooner.
Summary messages (is_summary=True) are not part of any turn: they carry
already-compressed history, not a new user request. They always go into
to_summarize so the fresh summary replaces them (the single-summary
invariant) — an old summary can never act as a turn's user-request head
(which the midturn split would keep verbatim beside the fresh summary) or
be rescued by the adaptive swap. This is what used to make repeated
midturn compressions pile stale summary copies into the context.
"""
non_system = [m for m in messages if m.role != "system"]
summary_msgs = [m for m in non_system if m.is_summary]
convo = [m for m in non_system if not m.is_summary]
# Group into turns: each turn starts with a user message. A non-user
# message with no open turn (can only follow a stripped summary) trails
# the user request that preceded it, so it attaches to the previous turn
# instead of forming an orphan turn without a head.
turns: list[list[Message]] = []
for msg in convo:
if msg.role == "user" or not turns:
turns.append([msg])
else:
turns[-1].append(msg)
if len(turns) <= keep_recent:
if keep_recent_messages is not None:
intra_turn = partition_current_turn_messages(turns, keep_recent_messages)
if intra_turn is not None:
old, to_keep = intra_turn
# A compression pass replaces every old summary, never keeps
# one beside the fresh one (single-summary invariant).
return _order_like(non_system, old + summary_msgs), to_keep
if summary_msgs:
# Even when nothing else has aged out, summaries are never kept:
# two or more of them must consolidate (the meta-summary path in
# compress_context) instead of sitting in the context together.
return _order_like(non_system, summary_msgs), convo
return [], non_system # nothing old enough to compress
# Adaptive: pull important older turns into the kept region and push
# unimportant recent/filler turns out for summarization.
base_keep = keep_recent
recent_turns = turns[-base_keep:]
old_turns = turns[:-base_keep]
# Adaptive swap rescues important old turns in multi-turn conversations. It is
# disabled in midturn mode (keep_recent_messages set): there the priority is to
# compress the in-flight turn predictably (see the split below), and the swap
# could otherwise move a filler in-flight turn into old_turns and summarize it
# whole — losing the current request. No test exercises the swap with
# keep_recent_messages set, so this guard is safe.
if adaptive and keep_recent_messages is None:
# Identify important old turns that should not be lost.
important_old = [t for t in old_turns if _turn_importance(t) > 0]
# Identify filler turns in the recent window that can be swapped out.
filler_recent = [t for t in recent_turns if _turn_importance(t) == 0]
swaps = min(len(important_old), len(filler_recent))
for i in range(swaps):
# Replace the oldest filler in recent with the most important old turn.
recent_turns[recent_turns.index(filler_recent[i])] = important_old[-(i + 1)]
# Re-sort kept turns by original position so context order stays chronological.
kept_set = {id(t) for t in recent_turns}
recent_turns = [t for t in turns if id(t) in kept_set]
old_turns = [t for t in turns if id(t) not in kept_set]
to_summarize = [m for turn in old_turns for m in turn]
# Midturn mode (keep_recent_messages set): the in-flight (last) turn can be very
# long (many tool iterations). When there are already > keep_recent turns, the
# turn-based keep window holds the whole in-flight turn verbatim, so midturn
# compression barely shrinks the context. Split the in-flight turn too: keep its
# head (the user request) + the newest keep_recent_messages verbatim, summarize
# the middle — on top of compressing older turns. Mirrors the head/tail shape of
# partition_current_turn_messages. Without this, the in-flight turn stays verbatim
# whenever turns > keep_recent, which is the 150->141 shallow-compression symptom.
if keep_recent_messages is not None and recent_turns:
in_flight = recent_turns[-1]
if len(in_flight) > keep_recent_messages + 1:
head = [in_flight[0]] if in_flight and in_flight[0].role == "user" else []
tail_start = max(len(head), len(in_flight) - keep_recent_messages)
to_summarize = to_summarize + in_flight[len(head):tail_start]
to_keep = (
[m for turn in recent_turns[:-1] for m in turn]
+ head
+ in_flight[tail_start:]
)
else:
to_keep = [m for turn in recent_turns for m in turn]
else:
to_keep = [m for turn in recent_turns for m in turn]
# Fold every old summary into this pass; to_summarize is emitted in
# chronological order so the summarizer sees history in sequence.
to_summarize = _order_like(non_system, to_summarize + summary_msgs)
return to_summarize, to_keep
def partition_current_turn_messages(
turns: list[list[Message]],
keep_recent_messages: int,
) -> tuple[list[Message], list[Message]] | None:
"""
Mid-turn fallback for long autonomous tool loops.
A long chain of assistant/tool iterations after one user message is one
conversational turn, so turn-based compression may have nothing to compress.
Keep the current user request and the newest messages verbatim, then summarize
older messages from the same in-flight turn.
"""
if not turns:
return None
current_turn = turns[-1]
if len(current_turn) <= keep_recent_messages + 1:
return None
head = [current_turn[0]] if current_turn and current_turn[0].role == "user" else []
tail_start = max(len(head), len(current_turn) - keep_recent_messages)
to_summarize = [m for turn in turns[:-1] for m in turn] + current_turn[len(head):tail_start]
to_keep = head + current_turn[tail_start:]
if len(to_summarize) < 2:
return None
return to_summarize, to_keep
def _format_for_summary(messages: list[Message]) -> 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.
"""
lines: list[str] = []
images: list[str] = []
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:
images.extend(m.images)
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
# Hard-truncate fallbacks keep at most this fraction of the context window's
# tokens of recent messages (the rest is dropped without summarizing).
_HARD_TRUNCATE_TOKEN_FRAC = 0.5
# 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),
)
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
def _plan_compression(
context: list[Message],
keep_recent: int,
keep_recent_messages: int | None = None,
*,
window: int | None = None,
estimate: Callable[[list[Message]], int] | None = None,
) -> tuple[list[Message], list[Message], list[Message]] | None:
"""Decide what to summarize and what to keep — the single decision point
shared by ``compress_context`` (real compression) and
``ContextCompressor.would_compress`` (dry-run prediction), so the two can
never drift apart.
Applies, in order:
1. Turn-based partition (``partition_messages``).
2. Target hysteresis (turn-based mode only): shrink the verbatim
keep-recent window until the kept region fits the target fraction of
the context window. A fixed keep_recent leaves roughly the same context
size after compression (just below the 90% trigger), so the next few
messages re-trigger it — the "125 -> 104, then 104 again after a couple
of messages" cycle. Dropping the kept size to the 65% target leaves
real headroom and folds the extra turns into the summary (same single
LLM call, no extra round-trip). Midturn mode (keep_recent_messages set)
already splits the in-flight turn aggressively, so a post-check in
compress_session truncates if it is still over target.
3. Aggressive intra-turn fallback (midturn mode only): when the turn-based
partition found nothing, retry with keep_recent_messages=2.
Returns (system_msgs, to_summarize, to_keep) or None when there is nothing
substantial to compress.
"""
system_msgs = [m for m in context if m.role == "system"]
to_summarize, to_keep = partition_messages(
context, keep_recent, keep_recent_messages=keep_recent_messages
)
if keep_recent_messages is None:
est = estimate or ContextCompressor.estimate_context_tokens
target_tokens = int(
(window or settings.ollama_num_ctx) * settings.context_compression_target
)
kept_tokens = est(system_msgs + to_keep)
_KEEP_RECENT_FLOOR = 2
while kept_tokens > target_tokens and keep_recent > _KEEP_RECENT_FLOOR:
keep_recent -= 1
to_summarize, to_keep = partition_messages(
context, keep_recent, keep_recent_messages=keep_recent_messages
)
kept_tokens = est(system_msgs + to_keep)
if len(to_summarize) < 2 and keep_recent_messages is not None and keep_recent_messages > 2:
to_summarize, to_keep = partition_messages(
context,
keep_recent,
keep_recent_messages=2,
)
if len(to_summarize) < 2:
return None
return system_msgs, to_summarize, to_keep
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
summary_text_input, images = _format_for_summary(to_summarize)
# 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)"
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),
)
return system_msgs + [summary_msg] + to_keep, summary_text
class ContextCompressor:
"""High-level context compression with retry strategy and hard-truncate fallback.
Thin wrapper around `compress_context` that adds:
1. Retry with keep_recent + 4 on LLM failure.
2. Hard-truncate fallback (drop oldest messages without summarizing).
"""
@staticmethod
def context_chars(context: list[Message]) -> int:
"""Characters of message text — the input to every token estimate."""
return sum(len(m.content or "") for m in context)
@staticmethod
def estimate_context_tokens(context: list[Message]) -> int:
"""Raw chars // 3 heuristic, no calibration.
Kept as the reference unit for tests and for callers that want the
old, deliberately under-counting number. Window decisions must use
``estimate_tokens`` instead — on prod this heuristic landed at ~56k
where the real prompt was 101k, and targets expressed in it were met
while the real prompt sat above the window.
"""
chars = ContextCompressor.context_chars(context)
imgs = sum(500 * len(m.images) for m in context if m.images)
return chars // 3 + imgs
@staticmethod
def estimate_tokens(context: list[Message], model=None) -> int:
"""Token estimate in real-prompt units, used by every window decision.
``estimate_context_tokens`` converted to real prompt tokens by the
calibration fitted for ``model`` (see ``navi/core/token_calibration.py``
for why chars // 3 is not usable as-is). Without observations the
calibration falls back to ``DEFAULT_CHARS_PER_TOKEN``.
"""
chars = ContextCompressor.context_chars(context)
imgs = sum(500 * len(m.images) for m in context if m.images)
return calibration_for(model).tokens_for(chars) + imgs
@staticmethod
def estimate_tokens_delta(context: list[Message], model=None) -> int:
"""Price messages *appended since* a real baseline, same units.
Uses the calibration slope without its intercept: the fixed prompt
overhead is already part of the baseline count. Pricing a delta with
the raw ``chars // 3`` left the pre/mid-turn gates in one unit and the
window in another for exactly the tool-heavy turns they exist for.
"""
chars = ContextCompressor.context_chars(context)
imgs = sum(500 * len(m.images) for m in context if m.images)
return calibration_for(model).delta_for(chars) + imgs
def __init__(self) -> None:
self._profile: "AgentProfile | None" = None
# Real prompt-token baseline from the last LLM call:
# (len(session.context) at call time, real prompt_tokens returned).
# The next estimate reuses these real tokens for the bulk of the context
# and only heuristics the messages appended since, instead of the
# chars//3 estimate which undercounts code-heavy tool output and fires
# compression too late. Cleared after compression (context shrank).
self._real_baseline: tuple[int, int] | None = None
def set_profile(self, profile: "AgentProfile | None") -> None:
"""Tell the compressor which profile is active so it can use profile-specific settings."""
self._profile = profile
def record_real_baseline(
self,
context_len: int,
real_tokens: int | None,
*,
chars: int | None = None,
model=None,
) -> None:
"""Cache the real prompt-token count returned by the last LLM call.
``context_len`` is ``len(session.context)`` at the time of that call
(before the new assistant/tool messages are appended). The next
estimate reuses these real tokens for the bulk of the context and only
heuristics the new messages appended since.
``chars`` is the character count of the exact prompt that was sent
(the built context, not just session.context). Together with
``real_tokens`` it feeds the per-model calibration line, which is what
converts hypothetical contexts (the proposed kept region) into real
tokens.
"""
if real_tokens and real_tokens > 0:
self._real_baseline = (context_len, int(real_tokens))
if chars:
calibration_for(model).observe(chars, real_tokens)
def real_baseline_estimate(
self,
session_context: list[Message],
built_ctx: list[Message],
model=None,
) -> int:
"""Estimate context tokens using the real baseline when available.
Falls back to the calibrated ``estimate_tokens`` on ``built_ctx`` when
no baseline is recorded or it is stale (the session shrank, e.g. right
after compression — ``base_len`` exceeds the current length). Every
branch is in real prompt tokens, the unit the window is measured in.
"""
if self._real_baseline is not None:
base_len, base_tokens = self._real_baseline
if base_len <= len(session_context):
delta = self.estimate_tokens_delta(
session_context[base_len:], model
)
return base_tokens + delta
return self.estimate_tokens(built_ctx, model)
async def compress_session(
self,
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,
window: int | None = None,
) -> tuple[list[Message], str] | None:
"""Compress context with retry + hard-truncate fallback.
Returns (new_context, summary_text) or None if nothing changed.
Does NOT mutate the session — the caller is responsible for updating
session.context, session.context_token_count, and persisting.
"""
effective_keep_recent = getattr(self._profile, "compression_keep_recent", None) or keep_recent
effective_max_tokens = getattr(self._profile, "compression_max_tokens", None) or max_tokens
window_tokens = window or settings.ollama_num_ctx
# Every size decision in this pass — the keep-region target, the
# post-check, the truncate budgets — must be in real-token units, the
# same units the trigger fired in. Mixed units are what let a
# compression pass finish at "65% of the window" while the real prompt
# it produced was above the window.
est = lambda msgs: self.estimate_tokens(msgs, model)
# Attempt 1: normal compression
try:
result = await compress_context(
context=context,
llm=llm,
model=model,
temperature=temperature,
keep_recent=effective_keep_recent,
max_tokens=effective_max_tokens,
keep_recent_messages=keep_recent_messages,
profile=self._profile,
window=window_tokens,
estimate=est,
)
except Exception:
# Attempt 2: keep more recent turns verbatim
try:
result = await compress_context(
context=context,
llm=llm,
model=model,
temperature=temperature,
keep_recent=effective_keep_recent + 4,
max_tokens=effective_max_tokens,
keep_recent_messages=(keep_recent_messages + 4)
if keep_recent_messages is not None
else None,
profile=self._profile,
window=window_tokens,
estimate=est,
)
except Exception:
# Attempt 3: hard-truncate fallback
return self._hard_truncate(context, window=window_tokens, estimate=est)
if result is None:
# Partition found nothing to summarize (e.g. few very large messages
# in a single turn). If the context is nonetheless over the token
# threshold, fall back to a token-budget hard-truncate so the gate
# firing always results in shrinkage — otherwise the agent runs until
# the window overflows. (The per-message truncation in
# context_builder.build caps single huge messages for the LLM view;
# this is the stored-context safety net.)
if should_compress(
est(context),
window_tokens,
settings.context_compression_threshold,
):
return self._token_budget_truncate(
context, estimate=est, window=window_tokens
)
return None
# Target hysteresis safety net: if the compressed context is still over
# the target fraction (midturn mode kept a huge in-flight turn, or even
# the keep_recent floor could not fit the window), drop oldest kept
# turns by token budget so a compression firing always leaves real
# headroom below the trigger — otherwise the next few messages re-fire
# compression and the context yo-yos at the trigger line. The dropped
# turns were kept verbatim (not summarized) and are lost; this only
# triggers when the verbatim kept region alone exceeds the target.
new_context, summary_text = result
target_tokens = int(window_tokens * settings.context_compression_target)
if est(new_context) > target_tokens:
truncated = self._token_budget_truncate(
new_context, budget_tokens=target_tokens, estimate=est, window=window_tokens
)
if truncated is not None:
new_context = truncated[0]
return new_context, summary_text
def would_compress(
self,
context: list[Message],
keep_recent: int,
keep_recent_messages: int | None = None,
threshold: float | None = None,
max_context_tokens: int | None = None,
current_tokens: int | None = None,
model=None,
) -> bool:
"""True if ``compress_session`` would actually shrink the context.
Runs the same ``_plan_compression`` decision as ``compress_context``
(no LLM call). Also returns True when the context is over the token
threshold even though partition found nothing — the token-budget
fallback in ``compress_session`` handles that. Used to emit
``CompressionStarted`` only when compression will really happen
(no misleading "compression" status on a no-op).
"""
effective_keep_recent = getattr(self._profile, "compression_keep_recent", None) or keep_recent
window_tokens = max_context_tokens or settings.ollama_num_ctx
est = lambda msgs: self.estimate_tokens(msgs, model)
if _plan_compression(
context,
effective_keep_recent,
keep_recent_messages=keep_recent_messages,
window=window_tokens,
estimate=est,
) is not None:
return True
# Partition no-op: compression can still shrink via the token-budget
# fallback — but only if dropping oldest turns actually reduces the
# context (a single huge turn that alone exceeds the budget cannot be
# split by message-boundary truncation, and is handled instead by the
# per-message view truncation in context_builder.build). Mirror that
# decision so would_compress and compress_session agree exactly.
# The size compared against the threshold is the caller's real-token
# estimate (`current_tokens`), not a fresh heuristic: the trigger fired
# on real tokens, and judging it again in chars // 3 units made the two
# disagree by ~1.8x.
if threshold is not None and max_context_tokens is not None:
total = current_tokens if current_tokens is not None else est(context)
if should_compress(total, window_tokens, threshold):
return self._token_budget_truncate(
context, estimate=est, window=window_tokens
) is not None
return False
def _hard_truncate(
self,
context: list[Message],
window: int | None = None,
estimate: Callable[[list[Message]], int] | None = None,
) -> tuple[list[Message], str] | None:
"""Last-resort fallback: drop oldest non-system messages without summarizing.
Keeps whole conversational turns so tool call groups are never split.
Bounded by a token budget (half the context window) rather than a fixed
message count, so it still shrinks a context made of a few very large
messages — the fixed-count floor used to no-op on <= 6 messages even
when they were huge.
"""
budget = max(1, int((window or settings.ollama_num_ctx) * _HARD_TRUNCATE_TOKEN_FRAC))
return self._token_budget_truncate(
context, budget_tokens=budget, estimate=estimate, window=window
)
def _token_budget_truncate(
self,
context: list[Message],
budget_tokens: int | None = None,
estimate: Callable[[list[Message]], int] | None = None,
window: int | None = None,
) -> tuple[list[Message], str] | None:
"""Drop oldest non-system messages (by turn groups) until under budget.
Keeps system messages and the newest conversational turns whole (tool
call groups stay together). Returns None if the context already fits the
budget or there is nothing to drop. ``budget_tokens`` defaults to half
the context window.
"""
est = estimate or ContextCompressor.estimate_context_tokens
if budget_tokens is None:
budget_tokens = max(
1, int((window or settings.ollama_num_ctx) * _HARD_TRUNCATE_TOKEN_FRAC)
)
system_msgs = [m for m in context if m.role == "system"]
non_system = [m for m in context if m.role != "system"]
if not non_system:
return None
# Group non-system into turns (user message starts a turn)
turns: list[list[Message]] = []
current: list[Message] = []
for msg in non_system:
if msg.role == "user" and current:
turns.append(current)
current = [msg]
else:
current.append(msg)
if current:
turns.append(current)
# Keep newest turns (whole) until the token budget is exhausted; always
# keep at least the most recent turn so the agent retains the in-flight
# request + its latest tool results.
kept_turns: list[list[Message]] = []
used = 0
for turn in reversed(turns):
turn_tokens = est(turn)
if kept_turns and used + turn_tokens > budget_tokens:
break
kept_turns.insert(0, turn)
used += turn_tokens
to_keep = [m for turn in kept_turns for m in turn]
if len(to_keep) == len(non_system):
return None # already within budget
new_context = system_msgs + to_keep
# dropped > 0 is guaranteed here: to_keep == non_system returns None above.
summary_text = (
"[Context was too large to summarize. Old messages were truncated to prevent "
"the model from exceeding its context window. Some earlier details may have been lost.]"
)
return new_context, summary_text
def check_context_size(
self,
context: list[Message],
session_context: list[Message] | None = None,
model=None,
window: int | None = None,
) -> None:
"""Raise ContextTooLargeError before an LLM call if the context is dangerously large."""
# Local import: the window and the reserve are read at call time, so a
# runtime settings swap (tests, operator reload) reaches this check.
from navi.config import settings as _settings
from navi.exceptions import ContextTooLargeError
if not context:
return
window_tokens = window or _settings.ollama_num_ctx
# A reserve bigger than the window turns `available` negative and makes
# every check fail, however small the context — a 500-token model still
# has to be able to answer. Cap the reserve at a quarter of the window.
output_reserve = min(
_settings.output_reserve_tokens, max(1, window_tokens // 4)
)
est = lambda msgs: self.estimate_tokens(msgs, model)
# When the caller supplies the raw session context, prefer the real-token
# baseline (real bulk + calibrated delta) over the calibrated estimate;
# both are in the same units as the real counts the window is measured in.
total = (
self.real_baseline_estimate(session_context, context, model)
if session_context is not None
else est(context)
)
available = window_tokens - output_reserve
if total > available:
existing = est(context[:-1])
new = est(context[-1:])
remaining = available - existing
raise ContextTooLargeError(
f"Context too large: new content is ~{new:,} estimated tokens, "
f"but only ~{max(0, remaining):,} tokens are available "
f"(window {window_tokens:,}, already used ~{existing:,}, "
f"output_reserve {output_reserve:,}). "
"Split the file into smaller parts or delegate to a subagent."
)
async def compress_and_save_session(
self,
session,
session_store,
llm: LLMBackend,
model: str,
temperature: float,
session_id: str,
reason: str,
keep_recent: int,
max_tokens: int | None = None,
keep_recent_messages: int | None = None,
window: int | None = None,
) -> ContextCompressed | None:
"""Compresses the session context and persists the changes to the session store."""
count_before = len(session.context)
window_tokens = window or settings.ollama_num_ctx
result = await self.compress_session(
context=session.context,
llm=llm,
model=model,
temperature=temperature,
keep_recent=keep_recent,
max_tokens=max_tokens,
keep_recent_messages=keep_recent_messages,
window=window_tokens,
)
if result is None:
return None
new_context, summary_text = result
# Mark messages in session.messages as not context if they are no longer in new_context
# and are not system messages.
# Iterate session.context too: after a session reload, context-only rows
# (old summaries — is_display=False) are present in session.context but
# NOT in session.messages, so demoting only the latter left stale
# summaries with is_context=True in the DB — they piled up in the
# context at every reload (the half-the-context-is-summaries symptom).
new_context_ids = {id(m) for m in new_context}
demoted: list[Message] = []
for msg in [*session.messages, *session.context]:
if msg.role != "system" and id(msg) not in new_context_ids:
msg.is_context = False
demoted.append(msg)
# Rows that live only in session.context — a summary is is_display=False,
# so after a reload it is absent from session.messages — are invisible to
# save(), which updates flags for display history only. Their demotion
# used to stay in memory: every reload brought the stale summary back
# with is_context=True, and one prod session ended up carrying 15 live
# summaries, 80% of its context. Persist those flips explicitly.
in_messages = {id(m) for m in session.messages}
context_only = [
m.sequence_number
for m in demoted
if m.sequence_number >= 0 and id(m) not in in_messages
]
if context_only:
await session_store.set_context_flag(session_id, context_only, False)
# Add the summary message to session.messages if it's not already there
summary_msg = next((m for m in new_context if m.is_summary), None)
if summary_msg and summary_msg not in session.messages:
summary_msg.is_display = False
session.messages.append(summary_msg)
# Add a system message with is_compression=True and content=summary_text
session.messages.append(
Message(
role="system",
content=summary_text,
is_compression=True,
is_context=False,
created_at=datetime.now(timezone.utc),
)
)
session.context = new_context
# The size the model is actually sent. A megabyte-long tool result stays
# in stored history (the user still sees it) but build() truncates it per
# message before the call — counting the raw message reported 817k tokens
# against a 128k window the model never received.
session.context_token_count = self.estimate_tokens(
truncate_oversized(new_context), model
)
# The context just shrank, so the recorded real-token baseline is stale.
# Drop it — the next LLM call re-establishes a fresh one.
self._real_baseline = None
await session_store.save(session)
log.info(
"compressor.compressed",
session_id=session_id,
reason=reason,
messages_before=count_before,
messages_after=len(new_context),
context_tokens=session.context_token_count,
)
# Archive old messages if the hot table exceeds the configured window.
if settings.session_messages_window > 0 and session.db_next_sequence > settings.session_messages_window:
threshold = session.db_next_sequence - settings.session_messages_window
archived = await session_store.archive_old_messages(session_id, threshold)
if archived > 0:
session.messages = [m for m in session.messages if m.sequence_number >= threshold]
session.context = [m for m in session.context if m.sequence_number >= threshold]
session.archive_threshold = threshold
return ContextCompressed(
messages_before=count_before,
messages_after=len(new_context),
summary=summary_text,
context_tokens=session.context_token_count,
max_context_tokens=window_tokens,
)
