Compact Long Context¶
SQLite makes the project conversation durable. It also preserves every result forever. One large document can dominate the next request, and a long session eventually reaches the model’s context limit.
Persistence controls lifetime. It does not control size.
Outcome¶
Tool handlers return bounded text, and AutoCompactStore summarizes older
messages when actual provider usage crosses a configured input threshold.
Recent messages remain verbatim.
Fast Track¶
Bound tool output before the handler returns it.
Keep the
SQLiteContextStoreas the durable inner store.Wrap it with
AutoCompactStoreand pass the agent transport.Choose
keep_recentas a message count.Observe stored usage instead of estimating tokens from characters.
Hands-on delta¶
1. Bound tool results first¶
The renderer controls terminal output only. Axio stores the handler’s complete return value and sends it on later model requests.
Add one application-level bound. The adapter also covers third-party text tools whose handlers you do not control:
TOOL_OUTPUT_MAX_CHARS = 4_000
TRUNCATION_MARKER = "\n[tool output truncated]"
def bound_tool_output(
text: str,
max_chars: int = TOOL_OUTPUT_MAX_CHARS,
) -> str:
if max_chars < len(TRUNCATION_MARKER):
raise ValueError("max_chars is too small for the truncation marker")
if len(text) <= max_chars:
return text
prefix_length = max_chars - len(TRUNCATION_MARKER)
return text[:prefix_length] + TRUNCATION_MARKER
def bound_text_tool(
tool: Tool[Any],
max_chars: int = TOOL_OUTPUT_MAX_CHARS,
) -> Tool[Any]:
async def bounded_handler(**kwargs: Any) -> Any:
result = await tool(**kwargs)
if isinstance(result, str):
return bound_tool_output(result, max_chars)
return result
return Tool(
name=tool.name,
handler=bounded_handler,
description=tool.description,
schema=tool.schema,
)
Keep the existing access guard and change only the return path in
read_document:
async def read_document(
path: Annotated[
str, Field(description="Exact document path, for example README.md")
],
) -> str:
"""Return one document visible to the current user."""
document = DOCUMENTS.get(path)
if document is None:
return f"Document {path} was not found."
rendered = f"{path}: {document['summary']}"
return bound_tool_output(rendered)
Apply bound_tool_output inside handlers you control. Use bound_text_tool
for an existing text tool. The wrapper keeps the original tool contract and
bounds its result before Axio stores it.
Bound structured fields before encoding JSON. Use separate size and dimension limits for images or other binary results.
2. Compact accumulated history¶
Keep connection ownership from the previous lesson. Wrap the session store
before passing it to render_turn:
DATA_DIRECTORY = Path(os.environ.get("AXIO_TUTORIAL_DATA_DIR", "data"))
DATABASE_PATH = DATA_DIRECTORY / "harness.db"
PROJECT_ID = "project-workbench"
SESSION_ID = os.environ.get("AXIO_SESSION_ID", "local-project-demo")
async def run_compacting_turn(prompt: str) -> SessionEndEvent:
connection = await connect(DATABASE_PATH)
base_context = SQLiteContextStore(
connection,
session_id=SESSION_ID,
project=PROJECT_ID,
)
context = AutoCompactStore(
base_context,
transport,
keep_recent=6,
threshold=0.75,
)
try:
return await render_turn(agent, prompt, context)
finally:
await context.close()
await connection.close()
The wrapper keeps SQLite as the durable store. It adds a compaction decision when Axio records each transport iteration.
Use two rules:
Trigger from the current iteration’s real
input_tokens, not cumulative session usage.Treat
keep_recent=6as sixMessageobjects, not six turns or tokens.
Axio keeps tool calls paired with their results. It also leaves the original history intact when summarization cannot complete.
See Context & Messages for the full compaction algorithm and Stream Events for provider usage events.
Try It¶
Use an explicit low threshold to trigger compaction without a live provider. The scripted response is the summary produced by the compaction agent:
async def demonstrate_compaction() -> None:
base_context = MemoryContextStore()
original = []
for index in range(10):
role = "user" if index % 2 == 0 else "assistant"
message = Message(
role=role,
content=[TextBlock(text=f"project documentation message {index}")],
)
original.append(message)
await base_context.append(message)
summary_transport = StubTransport(
[
make_text_response(
"Earlier discussion concerned README.md.",
usage=Usage(input_tokens=20, output_tokens=4),
)
]
)
context = AutoCompactStore(
base_context,
summary_transport,
keep_recent=4,
max_tokens=100,
)
await context.add_context_tokens(input_tokens=101, output_tokens=7)
history = await context.get_history()
assert len(history) == 6
assert history[0].content == [
TextBlock(text="Earlier discussion concerned README.md.")
]
assert history[-4:] == original[-4:]
assert await context.get_context_tokens() == (101, 7)
Run uv run python examples/tutorial/compact_long_context.py. The complete
example also checks both output adapters before it runs the persistent agent
turn.
Done when¶
[ ] Every potentially large tool has an explicit output bound.
[ ]
AutoCompactStorewraps the durable SQLite store.[ ] Compaction uses real
IterationEndinput usage.[ ]
keep_recentis chosen as a message count.[ ] Cumulative input and output totals survive compaction.
[ ] A failed summary leaves the original history available.
Next failure¶
The project assistant can now preserve a long documentation conversation. A developer next asks it to inspect repository files and run commands. Local handlers would inherit the host process’s permissions.
The next lesson adds those capabilities through bounded
DockerSandbox.tools wrappers and keeps the agent loop unchanged.
Continue with Isolate Execution.