Skip to content

How to Consolidate History into Chronicle Sections

This guide shows how to keep a long run's history inside a context window: each time a unit of work closes (a negotiating round, a scene), the events since the last one are summarised into a chronicle section, and an agent reads the summarised past next to the recent events verbatim.

Before you start

  • A working pinned model, as in How to Pin Model and Tool Calls. Summaries are model calls, and they should be pinned like any other.

  • The vocabulary used below:

    • a section is prose a model wrote over a range of the log, citing the range it covers;
    • consolidation is writing a section when a unit closes; it never deletes or edits the events it summarises;
    • recency vs depth: recent events are read verbatim from the log, and the older past is read as sections;
    • a section's anchor (its range, its unit and what it supersedes) is derived from the log and replays exactly, while its prose is generated and differs if regenerated.

    The Chronicle explains each of these.

Who decides what

The application decides The library does
What closes a unit (the boundary event) Detecting each boundary and the range since the last
The summariser: its prompt, its model, what it renders Recording each section with its range and pins
When the consolidation projector runs Keeping consolidation off the write lock
How to retrieve sections for a query Separating replayable anchors from generated prose

1. Write the summariser

A Summarizer is any callable that takes the range's stored events and returns a Consolidation: the prose, plus the pin keys its model calls resolved under. Draw through a PinnedModel and collect the keys with on_key:

chronicle.py
from collections.abc import Sequence

from pydantic_ai import Agent
from pydantic_ai.messages import ModelMessage, ModelResponse, TextPart
from pydantic_ai.models.function import AgentInfo, FunctionModel

from anagrapheus import register_builtin_events
from anagrapheus.chronicle import (
    Brief,
    ChronicleFold,
    Consolidation,
    EventTypes,
    SQLiteChronicleStore,
    substring_retriever,
)
from anagrapheus.provider import PinnedModel
from anagrapheus.timeline import (
    DomainEvent,
    EventReader,
    EventTypeRegistry,
    PinLedger,
    Projector,
    SQLiteEventStore,
    StoredEvent,
    UpcastRegistry,
    is_bookkeeping,
)


class OfferMade(DomainEvent, frozen=True):
    party: str
    text: str


class RoundClosed(DomainEvent, frozen=True):
    round_number: int


def summariser_model(messages: list[ModelMessage], info: AgentInfo) -> ModelResponse:
    return ModelResponse(parts=[TextPart("The UK opened at 12%; Norway held at 5%.")])


class RoundSummary:
    """The application's summariser: renders a round, draws through a pinned model."""

    def __init__(self, resolver) -> None:
        self._resolver = resolver

    def __call__(self, events: Sequence[StoredEvent]) -> Consolidation:
        lines = [
            f"{e.payload['party']}: {e.payload['text']}"
            for e in events
            if e.envelope.type == "OfferMade"
        ]
        keys: list[str] = []
        model = PinnedModel(FunctionModel(summariser_model), self._resolver, on_key=keys.append)
        result = Agent(model).run_sync(
            "Summarise this round of a fisheries negotiation in one sentence:\n" + "\n".join(lines)
        )
        return Consolidation(prose=result.output, pins=tuple(keys))

The summariser receives every event in the range, including library events such as committed pins. Render only your own domain content, and never sequence numbers, event ids or timestamps: those differ between runs, and putting them in the prompt makes every replayed summary a pin miss.

2. Install the fold on its own projector

A ChronicleFold watches for your boundary event. EventTypes names boundary types outright; implement UnitBoundary for anything more involved.

chronicle.py (continued)
types = EventTypeRegistry()
register_builtin_events(types)
types.event("fisheries", "OfferMade")(OfferMade)
types.event("fisheries", "RoundClosed")(RoundClosed)

store = SQLiteEventStore(":memory:")
reader = EventReader(store, UpcastRegistry(), types)
chronicle = SQLiteChronicleStore(store.connection)
ledger = PinLedger(store, types)
fold = ChronicleFold(
    chronicle,
    reader,
    boundary=EventTypes(("fisheries", "RoundClosed")),
    summarize=RoundSummary(ledger),
)
# A projector of its own, so consolidation runs off the turn's hot path.
consolidate = Projector(store, reader, [fold])

Each projector keeps an independent position per fold, so you can run belief and identity folds after every append and run consolidate.catch_up() on its own schedule, for example at the end of each round. From async code, await consolidate.catch_up_async() instead; see Consolidate from async code. The chronicle fold is generative, meaning its content comes from a model call. The fold makes that call in its preparation step, prepare or prepare_async, before the projector takes the database's write lock; apply then only writes the summary it was given, so a slow summary never blocks appends.

3. Close units and consolidate

Append events as the run goes, including the boundary, then run the consolidation projector and commit the summariser's pins:

chronicle.py (continued)
store.append(
    [
        types.pending(OfferMade(party="UK", text="A 12% cut to the mackerel quota.")),
        types.pending(OfferMade(party="Norway", text="No more than 5%.")),
        types.pending(RoundClosed(round_number=1)),
        types.pending(OfferMade(party="UK", text="We could move to 10%.")),
    ]
)
consolidate.catch_up()
ledger.commit()

Round 1 (events 1 to 3) becomes one section, with id "1-3". The UK's new offer at seq 4 belongs to round 2, which is still open, so it is not consolidated yet.

4. Read recent events and the summarised past

A Brief splits recency from depth. Read the consolidated past as sections, and everything after it verbatim from the log:

chronicle.py (continued)
brief = Brief(chronicle, reader, admit=lambda e: not is_bookkeeping(e))
through = brief.anchors.consolidated_through()
print("consolidated through", through)
for section in brief.depth():
    print(section.anchor.section_id, section.prose, section.pins)
for event in brief.tail(since_seq=through).events:
    print("recent:", event.seq, event.envelope.type, event.payload)
print([s.anchor.section_id for s in substring_retriever(brief.depth(), "norway")])

consolidated through 3
1-3 The UK opened at 12%; Norway held at 5%. ('p1:model_call:1904ca75…',)
recent: 4 OfferMade {'party': 'UK', 'text': 'We could move to 10%.'}
['1-3']

The tail always comes from the log, never from the chronicle. If consolidation falls behind, the tail simply starts earlier and nothing is missing.

admit sees each event's envelope before anything is decoded, and the tail keeps only what it accepts. is_bookkeeping matches the events the pin ledger appends about itself, such as the PinPromoted that ledger.commit() wrote at seq 5, so this tail holds only what happened in the negotiation. Without admit, the tail is every event in the log. Pass upto_seq= to tail to stop it at a position instead of at the end of the log.

There are two read surfaces, at two grades:

  • brief.anchors (Anchors) has ranges, coverage and consolidated_through(), but no prose. Use it for anything that decides something: gating, validation, what to consolidate next.
  • brief.prose (Prose) and brief.depth() return sections with their prose. Prose is advisory: show it to a model, but never let it settle anything.

substring_retriever is a deliberately naive case-insensitive search, most recent first. For real retrieval, write your own Retriever: any callable taking (sections, query) and returning sections.

Brief and Brief.tail are for reading the split yourself: inspecting, logging, testing. They are not slices, so nothing they return can go into an assembled prompt. A prompt reads the same split through two slices in one grant: a ChronicleSlice for the sections, labelled as generated, and a LogSlice read from the end of the last section. Put chronicle prose in a prompt continues this script with that grant, budgeted and sent through a pinned model.

Verify it worked

Check the anchors rather than the prose: brief.anchors.all() lists one anchor per closed unit, and brief.anchors.covers(seq) returns the anchor containing a given event. Each section's pins should be non-empty when the summariser drew through a ledger, which shows that the summary can be replayed.

Carry the chronicle to a new store

Every other projection can be dropped and rebuilt from the log for free. The chronicle cannot: rebuilding it re-runs every summary call and, unless every pin still hits, returns different prose. When you move to a new database, carry the sections across with export and load:

from pathlib import Path

new_store = SQLiteEventStore(Path("new-backend.db"))
carried = SQLiteChronicleStore(new_store.connection)
print(carried.load(chronicle.export()))  # sections written: 1

load skips section ids already present, so an interrupted transfer can be repeated. It is a transfer of one chronicle, not a merge of two: two chronicles over the same log share section ids, and load keeps what is already there.

Rebuild only when you mean it

Projector.rebuild refuses a generative fold:

GenerativeRebuild: 'chronicle' is a generative fold: rebuilding re-runs every model call it made and returns the same anchors with different content. Carry it across instead, or pass discard_generated=True to accept both.

Rebuilding discards every section's prose

discard_generated=True drops every section and regenerates it. Anything that read the old prose read something that no longer exists, and every summary is paid for again. The anchors come back identical; the prose comes back identical only where the summariser's pins still hit.

To rebuild with identical prose, give the summariser a ReplayPins over the log, so every summary is answered from its pin:

from anagrapheus.chronicle import FOLD_NAME
from anagrapheus.timeline import ReplayPins

replay = ReplayPins.from_events(store.read(), types)
replayed_fold = ChronicleFold(
    chronicle,
    reader,
    boundary=EventTypes(("fisheries", "RoundClosed")),
    summarize=RoundSummary(replay),
)
Projector(store, reader, [replayed_fold]).rebuild(
    FOLD_NAME,
    reset=chronicle.reset,
    reset_in_transaction=True,
    discard_generated=True,
)

Consolidate from async code

consolidate.catch_up() cannot run inside an event loop when the summariser calls Agent.run_sync, which refuses to start while a loop is already running. Moving the call into asyncio.to_thread does not help either, because a SQLite connection may only be used by the thread that opened it. Instead, write the summariser as a coroutine, an AsyncSummarizer, and drive the projector with catch_up_async():

chronicle.py (continued)
import asyncio


def round_two_model(messages: list[ModelMessage], info: AgentInfo) -> ModelResponse:
    return ModelResponse(parts=[TextPart("The UK moved to 10%; Norway met it at 8%.")])


class AwaitedRoundSummary:
    """The same summariser, awaited: it draws with `Agent.run` rather than `run_sync`."""

    def __init__(self, resolver) -> None:
        self._resolver = resolver

    async def __call__(self, events: Sequence[StoredEvent]) -> Consolidation:
        lines = [
            f"{e.payload['party']}: {e.payload['text']}"
            for e in events
            if e.envelope.type == "OfferMade"
        ]
        keys: list[str] = []
        model = PinnedModel(FunctionModel(round_two_model), self._resolver, on_key=keys.append)
        result = await Agent(model).run(
            "Summarise this round of a fisheries negotiation in one sentence:\n" + "\n".join(lines)
        )
        return Consolidation(prose=result.output, pins=tuple(keys))


async def close_round_two() -> None:
    store.append(
        [
            types.pending(OfferMade(party="Norway", text="We could go to 8%.")),
            types.pending(RoundClosed(round_number=2)),
        ]
    )
    ledger = PinLedger(store, types)
    fold = ChronicleFold(
        chronicle,
        reader,
        boundary=EventTypes(("fisheries", "RoundClosed")),
        summarize=AwaitedRoundSummary(ledger),
    )
    await Projector(store, reader, [fold]).catch_up_async()
    ledger.commit()


asyncio.run(close_round_two())
for section in brief.depth():
    print(section.anchor.section_id, section.prose)

1-3 The UK opened at 12%; Norway held at 5%.
4-7 The UK moved to 10%; Norway met it at 8%.

Round 2 covers events 4 to 7: the UK offer that was still open in step 3, the pin that ledger.commit() promoted into the log, and the two events appended here.

Only the summaries are awaited. The projector awaits each one with no transaction open, then writes the batch and its position synchronously on the calling thread, so the store is only ever used from the thread that opened it. The pinned model resolves its pins through the ledger's awaited path on that same thread.

Some things to know about the awaited path:

  • It takes synchronous folds too. Belief and identity folds, and a generative fold with only a synchronous prepare, run under catch_up_async() as they do under catch_up(), so one projector can hold both kinds.
  • A synchronous summariser blocks the loop. Under catch_up_async() it is called on the event loop's thread and holds the loop until it returns. It cannot call Agent.run_sync, which raises inside a running loop, so a summariser that calls a model from async code should be a coroutine like AwaitedRoundSummary.
  • catch_up() and rebuild() refuse an async summariser. They raise TypeError before folding or resetting anything, so the model is not called and existing sections are kept. The fold recognises an async def function or a class whose __call__ is async def.
  • Rebuild with rebuild_async(). It takes the same arguments as rebuild() and refuses a generative fold in the same way.
  • Each fold's position is unchanged by a failed batch. If a summary raises, nothing from that batch is written, and the next catch_up_async() starts the batch again.
  • Do not run one projector from two tasks at once. Each would fold the same batch.

To write a generative fold of your own that awaits its draws, implement AsyncGenerativeFold's prepare_async(event) beside apply.

Re-consolidate a section

The fold consolidates each range once. To write a better account of a range the chronicle already covers, for example after improving the summariser's prompt, call reconsolidate with the id of the current section. It reads the same range again from the log, calls your summariser on it, and appends the result as the next generation, superseding the old section:

chronicle.py (continued)
import logging

from anagrapheus.chronicle import ChronicleFork, reconsolidate

logging.basicConfig(format="%(levelname)s %(name)s: %(message)s")


def revised_model(messages: list[ModelMessage], info: AgentInfo) -> ModelResponse:
    return ModelResponse(
        parts=[TextPart("Round 1: the UK asked for a 12% mackerel cut; Norway offered 5%.")]
    )


class RevisedRoundSummary:
    """The round summariser with a revised prompt that names the stock."""

    def __init__(self, resolver) -> None:
        self._resolver = resolver

    def __call__(self, events: Sequence[StoredEvent]) -> Consolidation:
        lines = [
            f"{e.payload['party']}: {e.payload['text']}"
            for e in events
            if e.envelope.type == "OfferMade"
        ]
        keys: list[str] = []
        model = PinnedModel(FunctionModel(revised_model), self._resolver, on_key=keys.append)
        result = Agent(model).run_sync(
            "Summarise this round of a fisheries negotiation in one sentence, naming the stock:\n"
            + "\n".join(lines)
        )
        return Consolidation(prose=result.output, pins=tuple(keys))


ledger = PinLedger(store, types)
revised = reconsolidate(chronicle, reader, "1-3", RevisedRoundSummary(ledger))
ledger.commit()
print(revised.anchor)
for section in brief.depth():
    print(section.anchor.section_id, section.prose)
print("still readable:", brief.prose.section("1-3").prose)

try:
    reconsolidate(chronicle, reader, "1-3", RevisedRoundSummary(ledger))
except ChronicleFork as refused:
    print("refused:", refused)

WARNING anagrapheus.chronicle.reconsolidation: re-consolidated 1-3 as 1-3#2 (seq 1..3): runs that read the old prose will diverge on replay, and a rebuild of the chronicle discards the new section
Anchor(section_id='1-3#2', from_seq=1, to_seq=3, unit=3, supersedes='1-3')
1-3#2 Round 1: the UK asked for a 12% mackerel cut; Norway offered 5%.
4-7 The UK moved to 10%; Norway met it at 8%.
still readable: The UK opened at 12%; Norway held at 5%.
refused: section '1-3' is already superseded; the current one is '1-3#2'

The new section keeps the old one's range and unit, and its id carries the generation: "1-3#2", then "1-3#3". The superseded prose stays readable by id, so you can still see what an earlier run read.

Some things to know about re-consolidating:

  • Pass the current section. A section that is already superseded is refused with ChronicleFork before the summariser is called, because superseding it again would leave two current sections over one range. The message names the section to use instead.
  • Every call logs a warning. Later prompts read the new prose, so a run that read the old prose no longer replays against it. The warning is logged through logging, on the anagrapheus.chronicle.reconsolidation logger, so that this is never silent.
  • It is outside the replay guarantee. Nothing in the log records a re-consolidation, so rebuilding the chronicle discards it and returns generation 1. Carry the chronicle to keep it.
  • From async code, await reconsolidate_async. It takes the same arguments and accepts either kind of summariser; reconsolidate raises TypeError for an async one.

To write a section by hand instead, use the store's append_new, which raises SectionCollision if the id is already stored. Its append keeps the existing row on a collision, because that is what a replayed fold batch needs. Both refuse a section that would fork the chronicle: one that supersedes an already-superseded section, one that supersedes a section over a different range, or one that overlaps a current section without superseding it.

Troubleshooting

No sections appear

The boundary never matched. EventTypes compares (namespace, type) exactly, so check both against your registration. Also check that something calls catch_up() on the projector holding the chronicle fold: a projector holding only your deterministic folds will never consolidate.

A replayed summary misses its pin

The summary prompt changed between recording and replay. The usual cause is rendering something that differs per run, such as seq, event_id or created_at. Render domain payload fields only.

RuntimeError: This event loop is already running from async code

The summariser calls Agent.run_sync from inside an event loop, either because something called catch_up() there or because the projector was driven with catch_up_async(), which calls a synchronous summariser on the loop's thread. Make the summariser a coroutine that calls Agent.run, and await catch_up_async() instead, as in Consolidate from async code.

TypeError: 'chronicle' only prepares asynchronously

The fold's summariser is a coroutine, and its projector was driven with catch_up() or rebuild(). Nothing was folded or reset. Use catch_up_async() or rebuild_async(). A summariser that is a plain function returning an awaitable is only found out once called, and raises TypeError: 'chronicle' has an async summarizer instead; write it as an async def.

ChronicleFork: section '…' is already superseded

You passed reconsolidate an older generation. Pass the id the message names as current, or find it with brief.prose.covering(seq).

ChronicleFork from append or load

The section would leave two current sections over one range, or a consolidated range that no current section fully covers. A hand-written section must supersede the current section for its range and cover exactly that range, from the same from_seq to the same to_seq, and load refuses the whole transfer if any incoming section overlaps a current one it does not supersede.

The tail contains PinPromoted events

The Brief was built without admit, so Brief.tail returns every event above the position, the pin ledger's included. Pass admit=lambda e: not is_bookkeeping(e), as in step 4. If you promote pins into your own vocabulary with promote_as, is_bookkeeping does not know those events: write your own admit that refuses them too.