Context Management: Budgets, Pruning, and Compaction

agents
reliability

The loop from Chapter 5 re-sends a growing message history on every model turn. That history is not free memory: it is an input-token budget with a hard ceiling, a latency cost, and an attention-allocation problem. This chapter makes the budget visible, then compares the two ways the current agent_harness package manages pressure: mechanical pruning and LLM-backed compaction. The important reliability question is not whether fewer tokens fit. It is whether the facts needed for the next correction still survive.

The demonstrations are offline and deterministic. The compactor receives a fake client whose response is fixed, so no provider call or API key is needed.

Context budget

A context window limits the prompt sent to one model call. Let n(h) be the estimated number of input tokens for history h, and let B be the model’s context window. A useful controller needs more than the boolean test n(h) > B: it needs an early warning threshold, a hard threshold, and a record of which usage field was measured.

ContextManager consumes TokenUsage.prompt_tokens from the client. Its cheap per-message estimator is deliberately simple, approximately one token per four characters plus ten message tokens. That estimate is a steering signal, not a billing ledger. The provider’s usage report remains the authority when available.

from pathlib import Path

from agent_harness.config import Config, ModelConfig
from agent_harness.context import ContextManager
from agent_harness.events import TokenUsage

config06 = Config(
    cwd=Path.cwd(),
    model=ModelConfig(name="offline-demo", context_window=200),
)
manager06 = ContextManager(
    config06,
    compaction_threshold=0.8,
    pruning_threshold=0.9,
    keep_last=2,
)
messages06 = [
    {"role": "system", "content": "You are a careful coding agent."},
    {"role": "user", "content": "Inspect the parser and preserve its public API."},
    {"role": "assistant", "content": "I will inspect the parser before proposing an edit."},
]

estimates06 = [manager06.estimate_message_tokens(message) for message in messages06]
print("per-message estimates:", estimates06)
for prompt_tokens in (120, 170, 185):
    manager06.update_token_count(TokenUsage(prompt_tokens=prompt_tokens))
    stats = manager06.get_context_stats()
    print(
        prompt_tokens,
        "=>",
        f"{stats['usage_percent']:.1f}%",
        "compact=",
        stats["needs_compaction"],
        "prune=",
        stats["needs_pruning"],
    )

The three observations separate policy from measurement. At 60% the manager is below both thresholds; at 85% it recommends compaction; at 92.5% it also reports hard-pruning pressure. The last call to update_token_count wins because the value describes the current prompt, not cumulative conversation spend. Cumulative usage belongs to the session ledger from Chapter 5, while this manager answers a narrower question: can the next prompt fit?

The estimate is intentionally cheap enough to run before every turn. Its error must be calibrated against provider reports before choosing a threshold. An underestimate turns a soft warning into a late failure; an overestimate spends context conservatively but may reduce task success.

History pruning

The simplest intervention is to remove old observations. The package’s pruner keeps the system message and the last keep_last messages verbatim. It first elides old tool messages, then removes old assistant messages until the estimated middle is below the compaction target. This is a useful baseline because it is local, cheap, and explainable.

It is also lossy. A tool result can contain the only copy of a test failure, and an old assistant message can contain the reason a later edit was chosen. Treat the protected tail as an explicit invariant rather than assuming that recency is the same as relevance.

from agent_harness.context import ContextManager

prune_manager06 = ContextManager(
    Config(cwd=Path.cwd(), model=ModelConfig(context_window=100)),
    compaction_threshold=0.5,
    keep_last=2,
)
messages_to_prune06 = [
    {"role": "system", "content": "system contract"},
    {"role": "user", "content": "old request " + "x" * 90},
    {"role": "tool", "tool_call_id": "old-call", "content": "old failure " + "x" * 160},
    {"role": "assistant", "content": "old diagnosis " + "x" * 90},
    {"role": "user", "content": "protected request"},
    {"role": "assistant", "content": "protected answer"},
]
pruned06 = prune_manager06.prune_messages(messages_to_prune06)

print("before:", [message["role"] for message in messages_to_prune06])
print("after: ", [message["role"] for message in pruned06])
print("protected tail:", [message["content"] for message in pruned06[-2:]])
print("old tool kept:", any(message.get("tool_call_id") == "old-call" for message in pruned06))
assert pruned06[0] == messages_to_prune06[0]
assert pruned06[-2:] == messages_to_prune06[-2:]
assert all(message.get("tool_call_id") != "old-call" for message in pruned06)
assert messages_to_prune06[1]["content"].startswith("old request")

<1> applies the policy to a copy, so the caller’s history is not mutated. The printed roles show the intended order of preference: the old tool observation disappears before the old assistant turn, while the system message and protected tail remain. <2> checks the less visible contract, namely that pruning did not mutate the original list behind the caller’s back.

This implementation has no tombstone message. The model therefore cannot tell whether an old tool result was deliberately elided or was never produced. A production pruner can replace the result with a compact marker containing the call id and reason, but that marker itself consumes budget and must not impersonate an observation. Also, if the protected tail alone exceeds the target, the current code cannot make the prompt fit without breaking its retention guarantee; it stops rather than deleting protected messages.

Context compaction

Pruning deletes history mechanically. Compaction replaces an eligible span with a new message generated by a summarizer. The package’s ChatCompactor keeps the system message and recent tail verbatim, formats the middle as a summarization prompt, and inserts one [Context Summary] message. This preserves more semantic content at the same budget, but introduces a new model call and a new failure mode: the summary may omit a correction, path, or pending task.

The first experiment uses a deterministic fake client. It exercises the real ChatCompactor protocol without relying on a remote model.

import asyncio

from agent_harness.client import LLMClient
from agent_harness.compaction import ChatCompactor
from agent_harness.events import StreamEvent, StreamEventType, TextDelta


class OfflineSummaryClient06(LLMClient):
    def __init__(self, summary: str):
        self.summary = summary
        self.calls = 0

    async def chat_completion(self, messages, stream=False):
        assert stream is False
        self.calls += 1
        yield StreamEvent(
            type=StreamEventType.MESSAGE_COMPLETE,
            text_delta=TextDelta(self.summary),
        )


summary_client06 = OfflineSummaryClient06(
    "- changed parser.py\n- tests still need to run\n- keep the public API stable"
)
compactor06 = ChatCompactor(summary_client06, config06)
conversation06 = [
    {"role": "system", "content": "system contract"},
    {"role": "user", "content": "Inspect parser.py."},
    {"role": "assistant", "content": "I found a tokenization edge case."},
    {"role": "tool", "tool_call_id": "call-1", "content": "pytest: 1 failed"},
    {"role": "user", "content": "Please fix only the parser."},
    {"role": "assistant", "content": "I will preserve the public API."},
]
compacted06 = asyncio.run(compactor06.compact(conversation06, keep_last=2))
print("roles:", [message["role"] for message in compacted06])
print("summary:", compacted06[1]["content"])
print("tail unchanged:", compacted06[-2:] == conversation06[-2:])
print("summarizer calls:", summary_client06.calls)
assert compacted06[0] == conversation06[0]
assert compacted06[-2:] == conversation06[-2:]
assert compacted06[1]["content"].startswith("[Context Summary]")

<1> is an asynchronous operation because a real summarizer is an LLM call. The resulting history has the same system message, one summary in the eligible span, and two verbatim recent messages. Notice the contract boundary: ChatCompactor knows how to preserve positions and detect a summary, but it cannot prove that the summary retained every fact that matters to the task.

The summary is inserted with role user. That is an intentional prompt-level convention in this library, not a privileged memory channel. The summary is still model-visible text and should be treated as a lossy observation when designing correction and safety policies.

compacted_again06 = asyncio.run(compactor06.compact(compacted06, keep_last=2))
print("first length:", len(compacted06))
print("second length:", len(compacted_again06))
print("byte-stable fixed point:", compacted_again06 == compacted06)
print("summarizer calls:", summary_client06.calls)
assert compacted_again06 == compacted06
assert summary_client06.calls == 1

The second pass is a termination test, not just a convenience check. ChatCompactor recognizes its own [Context Summary] marker and does not summarize that span again. The equality assertion establishes idempotence for this representation, and the call count proves that the fixed point required no hidden provider request. Without this guard, repeated threshold checks could recursively summarize summaries and slowly distort the history even when its size no longer changes.

Compaction loss

Idempotence does not imply faithfulness. A compactor can reach a stable fixed point while the first summary has already lost the only useful correction. Probe that risk with planted facts: place a correction in the eligible middle, use a deterministic summary that omits it, and inspect the result. The point is to measure the loss rather than argue from the summary’s tone.

lossy_client06 = OfflineSummaryClient06("- inspected app.py\n- pending work remains")
lossy_compactor06 = ChatCompactor(lossy_client06, config06)
loss_probe06 = [
    {"role": "system", "content": "system contract"},
    {"role": "user", "content": "Correction: never edit the test fixtures."},
    {"role": "assistant", "content": "I will keep that correction in mind."},
    {"role": "tool", "content": "The failing test is test_parser.py::test_fixture."},
    {"role": "user", "content": "Continue with the smallest safe change."},
]
lossy_result06 = asyncio.run(lossy_compactor06.compact(loss_probe06, keep_last=1))
summary_text06 = lossy_result06[1]["content"]
correction06 = "never edit the test fixtures"
print("correction present after compaction:", correction06 in summary_text06.lower())
print("summary under test:", summary_text06)
assert correction06 not in summary_text06.lower()

faithful_client06 = OfflineSummaryClient06(
    "- Correction: never edit the test fixtures.\n- failing test: test_parser.py::test_fixture"
)
faithful_compactor06 = ChatCompactor(faithful_client06, config06)
faithful_result06 = asyncio.run(faithful_compactor06.compact(loss_probe06, keep_last=1))
print("faithful summary preserves correction:", correction06 in faithful_result06[1]["content"].lower())
assert correction06 in faithful_result06[1]["content"].lower()

The lossy run reaches the same structural shape as the faithful run, yet only the second preserves the planted correction. This is self-induced distribution shift: the model’s next decision is conditioned on a harness-written message rather than the original trajectory. A useful survival set is therefore explicit: keep the system prompt, recent action/result pairs, unresolved errors, and safety-relevant corrections verbatim; summarize only what the task-specific policy has declared eligible. The survival set is a correctness constraint, not a prompt-writing preference.

Prompt assembly

A final context boundary is the system-prompt assembly order. build_system_prompt composes identity, environment, tool guidance, security guidance, developer instructions, user instructions, and operational guidance. The current function assembles only the system string; Session supplies it as message zero and appends conversation history separately. That separation makes it possible to test the invariant that pruning and compaction never rewrite the system contract.

from agent_harness.prompts import build_system_prompt
from agent_harness.tools.base import ToolRegistry

prompt_config06 = Config(
    cwd=Path.cwd(),
    developer_instructions="Project rule: run the offline regression suite before reporting success.",
    user_instructions="User request: explain any remaining uncertainty.",
)
prompt06 = build_system_prompt(prompt_config06, ToolRegistry(prompt_config06).get_tools())
section_names06 = [
    "# Identity",
    "# Environment",
    "# Security Guidelines",
    "# Project Instructions",
    "# User Instructions",
    "# Operational Guidelines",
]
positions06 = {name: prompt06.index(name) for name in section_names06}
print("section order:", [name for _, name in sorted((position, name) for name, position in positions06.items())])
print("developer before user:", positions06["# Project Instructions"] < positions06["# User Instructions"])
assert positions06["# Security Guidelines"] < positions06["# Project Instructions"]
assert positions06["# Project Instructions"] < positions06["# User Instructions"]
assert prompt06.startswith("# Identity")

The section-order assertion protects a real interface boundary: configuration and instructions are inserted after the built-in identity and security sections. It does not prove that a model will obey a later instruction. It also exposes two current implementation limits worth carrying into Chapter 7: the environment section includes the current date and platform, so byte-for-byte prompt purity is not yet guaranteed across machines, and history is not part of this helper’s input. A reliable assembler should make both sources of variation explicit.

Context invariants

The chapter’s controller can now be evaluated with four independent checks:

  1. The measured field is prompt tokens, and thresholds are visible.
  2. Pruning preserves the system message and protected tail without mutating the input.
  3. Compaction preserves its protected messages and reaches an idempotent fixed point.
  4. A planted correction is tested for survival instead of being assumed to survive.

The next chapter moves the same discipline to configuration and durable memory. Instructions change the system prompt; memory changes what can be removed from the transient context. Both need explicit precedence and provenance.

Back to top