Context compaction

Before each live model call in the tool loop (ctx.llm.run and ctx.llm.stream), Pyrula estimates the prompt size. When the estimate crosses a budget below the model’s context window, the history is compacted first. If it still can’t fit the window, the turn ends with context_window_exceeded instead of sending a request the provider would reject. ctx.llm.complete and ctx.llm.structured send their messages as given and are not compacted.

When it runs

Each setting is read from AgentLimits, then from the same-named attribute on the LLM client, then from a built-in default:

Setting	Default	Meaning
context_window	the model’s window, if Pyrula knows the model	Tokens the model accepts.
reserved_output_tokens	4096	Room kept free for the answer.
compaction_fraction	0.75	Share of the remaining room the prompt may use before compaction.

The budget is (context_window - reserved_output_tokens) * compaction_fraction, so a 200,000 token model compacts once the prompt passes 146,928 tokens. Pyrula knows the windows of the current Claude, GPT and Gemini models. For any other model set context_window; without it there is no budget and no guard, and Pyrula logs a warning once per model.

The estimate counts the system prompt, every message and every tool definition. GPT-family models are counted with tiktoken when it is installed (pip install tiktoken; it downloads its encoding on first use); everything else uses a character heuristic that over-counts. Pass estimate_tokens= in AgentLimits to use your own counter, called as estimate_tokens(history, system, tools, model=...) and returning an int.

Compaction changes only the history sent to the provider. The list the agent passed in is not modified.

The default compactor

drop_oldest keeps the first message (usually the task) and the 6 most recent, and drops about half of the messages in between on each pass. It never separates a tool call from its result and makes no model call. make_drop_oldest(keep_recent=10) builds one that keeps more.

Set the knobs per agent with AgentLimits:

from pyrula.agents import AgentLimits, agent
from pyrula.agents.llm.context import make_drop_oldest

@agent(limits=AgentLimits(
    context_window=128_000,
    compaction_fraction=0.6,
    history_compactor=make_drop_oldest(keep_recent=10),
))
async def support(ctx, message: str) -> str:
    ...

history_compactor=None turns compaction off for that agent but keeps the window guard.

Writing a compactor

A compactor is an async function that takes a CompactionContext (history, system, tools, token_count, budget, window, model) and returns the new history as a list of messages.

ctx.summarize(messages, instructions=...) asks the agent’s own model for a summary and returns the text. Without instructions it asks for a concise summary that keeps facts, decisions and open tasks. The summary is a journaled model call: it counts toward max_llm_calls, max_total_tokens and max_cost_usd, and replay serves it from the journal.

Pyrula re-estimates the returned history. If it fits in context_window - reserved_output_tokens the call goes ahead, even when it is still over the budget; otherwise the turn ends with context_window_exceeded. Returning the list you were given (return ctx.history) means “leave it”, and nothing is recorded.

Keep a compactor deterministic apart from ctx.summarize. Replay runs it again, so a compactor that reads the clock or a random number rebuilds a different history than the model saw live.

The following example summarizes the middle of a ticket history, crashes the turn before the answer call, and resumes it:

import asyncio

from pyrula.agents import AgentContext, AgentLimits, agent
from pyrula.agents.llm.context import CompactionContext
from pyrula.agents.testing.fake_llm import FakeLLM, FakeResponse
from pyrula.agents.testing.runner import AgentRunner

async def summarize_middle(ctx: CompactionContext) -> list[dict]:
    head, middle, tail = ctx.history[:1], ctx.history[1:-2], ctx.history[-2:]
    summary = await ctx.summarize(middle, instructions="Summarize the ticket history.")
    return [*head, {"role": "user", "content": f"Earlier: {summary}"}, *tail]

@agent(limits=AgentLimits(
    context_window=2_000,
    reserved_output_tokens=200,
    compaction_fraction=0.5,
    history_compactor=summarize_middle,
))
async def ticket(ctx: AgentContext, notes: list[str]) -> str:
    history = [{"role": "user", "content": "You handle ticket T-9."}]
    for note in notes:
        history += [{"role": "user", "content": note}, {"role": "assistant", "content": "Noted."}]
    history.append({"role": "user", "content": "What should we do next?"})
    return await ctx.llm.run(history)

async def main() -> None:
    llm = FakeLLM([
        FakeResponse(text="Customer was double-charged; refund approved."),   # the summary
        FakeResponse(text="Issue the refund."),                               # the answer
    ])
    runner = AgentRunner(ticket, llm=llm)
    notes = [f"Note {i}: " + "details " * 60 for i in range(8)]

    runner.inject_crash_at("llm", nth=2)   # the process dies before the answer call
    await runner.run(notes=notes)
    assert len(llm.calls) == 1

    events = await runner.run(notes=notes)   # resume
    assert len(llm.calls) == 2                # the summary was not asked for again
    assert len(llm.calls[-1]["messages"]) == 4
    assert events[-1].payload["result"] == "Issue the refund."

    journal = runner.store.load_run("ticket", runner.turn_id)
    compacted = next(e.payload for e in journal if e.kind == "context:compacted")
    assert (compacted["messages_before"], compacted["messages_after"]) == (18, 4)
    assert compacted["strategy"] == "summarize_middle"

asyncio.run(main())

What is recorded

Each compaction that changes the history writes a context:compacted event to the run stream with call_index (the model call it ran before), tokens_before, tokens_after, window, messages_before, messages_after and strategy (the compactor’s function name). A resumed attempt reruns the compactor but writes no second event for a call the journal already serves.

Each ctx.summarize call is recorded as a step:done with step_kind: "llm", its own call_index, and compaction_for naming the call it compacted for. The compacted history itself is not stored.

A turn that can’t fit ends with a run:error whose code is context_window_exceeded, carrying the estimated tokens and the context_window. No provider call is made.

Replay

Replay runs the compactor at the same point in the loop and answers its ctx.summarize calls from the journal in order, so every later call keeps the index it was recorded at.

A model call’s input fingerprint is taken from the history the agent built, before compaction, so changing the compactor, the window or the fraction doesn’t make a recorded call look changed. The exception is a compactor that no longer makes the summarize calls the journal recorded: replay verification stops at that call, and a saved case stops with “recorded context compaction was not consumed”.

In Pyrula Cloud

The context:compacted event is sent to Cloud at every capture level, none included. It holds counts and the compactor’s function name, never message content. A field a custom emitter adds is held back below full. Each summarize call appears in the timeline as a model call, with its tokens and cost counted in the run’s usage.