How to Assemble Prompts from Granted Slices

This guide shows how to build an agent's prompt so that it can contain only what that agent is allowed to see, every piece of it records its sources, and a size budget drops the least important pieces first. The example builds Norway's prompt in a fisheries negotiation: Norway may read the offers made in the room and its own private beliefs, never the UK's.

Before you start

• Beliefs set up as in How to Keep Per-Agent Beliefs as a Fold.

• The vocabulary used below:

  • a slice reads one source (the log, a belief store, the chronicle, or a value you hand in) at one log position, and remembers everything it returned;
  • a grant is the set of slices a role may read: a frozen dataclass whose every field is a slice;
  • a part is one structured piece of the prompt, with a slot name, a register, a rank and its sources;
  • the register says how binding a part is: binding (must hold), considered (advisory, ranked), craft (technique) or generated (prose a model wrote);
  • the budget is the size limit, and decides which parts are shed when the prompt is too big.

  Prompt Assembly explains each of these.

Who decides what

The application decides	The library does
Each role's grant: which slices, which scopes, which events	Refusing reads outside the grant
The assembler: which parts, in which slots and registers	Checking every part cites only what was read
Rendering parts into messages	Refusing chronicle prose under any register but generated
The budget's size, its token counter and shed order	Shedding by register and rank, never a binding part

flowchart LR
    G["Grant<br/>(slices at one position)"] --> A["Assembler<br/>(your code)"]
    A --> P["Parts with<br/>sources"]
    P --> C["assemble():<br/>checks + budget"]
    C --> R["Your renderer"]
    R --> M["PinnedModel"]

1. Bring the projections to the position you will read at

Every slice in a grant reads at one log position. A belief slice refuses to read if a belief event lies between that position and the belief fold's own position, so run the projector first and read at the log's last position:

assemble_prompts.py

import uuid
from dataclasses import dataclass

from doxastica import InMemoryBackend, MemoryCore
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.assembly import (
    BeliefSlice,
    Budget,
    EventRef,
    Given,
    LogSlice,
    Part,
    Register,
    assemble,
    render_labelled,
)
from anagrapheus.beliefs import (
    BeliefEvents,
    BeliefFold,
    Predicate,
    PredicateKind,
    PredicateRegistry,
    SQLiteBeliefIndexStore,
    StateRef,
)
from anagrapheus.provider import PinnedModel
from anagrapheus.timeline import (
    DomainEvent,
    EventReader,
    EventTypeRegistry,
    PinLedger,
    Projector,
    SQLiteEventStore,
    UpcastRegistry,
)


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


types = EventTypeRegistry()
register_builtin_events(types)
types.event("fisheries", "OfferMade")(OfferMade)
predicates = PredicateRegistry(
    [Predicate("position", 2, PredicateKind.FUNCTIONAL)], namespace="fisheries"
)
store = SQLiteEventStore(":memory:")
reader = EventReader(store, UpcastRegistry(), types)
fold = BeliefFold(
    MemoryCore(InMemoryBackend()),
    predicates,
    SQLiteBeliefIndexStore(store.connection),
    types,
    durable=False,
)
projector = Projector(store, reader, [fold])

uk, mackerel = uuid.uuid7(), uuid.uuid7()
beliefs = BeliefEvents(types, predicates)
store.append(
    [
        beliefs.revise("party:norway:held", "position", (uk, mackerel), "UK will settle at 8%"),
        beliefs.revise("party:uk:held", "position", (uk, mackerel), "cut 5%"),
        types.pending(OfferMade(party="UK", text="A 12% cut to the mackerel quota.")),
    ]
)
projector.catch_up()

2. Define the role's grant

A grant is a frozen dataclass of slices. Build a fresh one for every assembly, since a slice's record of what it handed out belongs to one call:

assemble_prompts.py (continued)

# The grant: everything Norway's negotiator may read, and nothing else.
@dataclass(frozen=True)
class NorwayGrant:
    room: LogSlice
    own_beliefs: BeliefSlice
    ask: Given[str]


def norway_grant(upto_seq: int) -> NorwayGrant:
    return NorwayGrant(
        room=LogSlice(
            reader,
            upto_seq=upto_seq,
            admit=lambda env: (env.namespace, env.type) == ("fisheries", "OfferMade"),
        ),
        own_beliefs=BeliefSlice.over(
            fold,
            projector,
            reader,
            visible=lambda scope: scope == "party:norway:held",
            upto_seq=upto_seq,
        ),
        ask=Given("ask", "Respond to the latest offer in one sentence."),
    )

The slices used here:

• LogSlice reads events up to upto_seq that admit accepts. admit sees only the envelope, so a refused event is never even decoded. This is where a private event is kept out of a shared prompt.
• BeliefSlice reads the scopes visible accepts. BeliefSlice.over asks the projector how far the belief fold has got, which is right for either kind of fold: an in-memory fold (durable=False) keeps its position on itself, not in the file. The slice refuses to read if an event the fold claims lies between the fold's position and upto_seq. Events the fold ignores, such as offers or promoted pins, are read past without complaint.
• Given wraps a value your application computed, such as the instruction.

Two more slices exist: ChronicleSlice for chronicle sections, and EventsSlice for a fixed run of events, such as the range a summariser is given.

3. Write the assembler

The assembler is a function from the grant to parts. It must be pure: the same log, position and grant must give the same parts, or replay will miss its pins. Cite each part's sources with the reference the slice handed out:

assemble_prompts.py (continued)

# The assembler: parts with a register, a rank and their sources.
def norway_assembler(grant: NorwayGrant) -> list[Part]:
    parts = [Part("ask", Register.binding, grant.ask.value, sources=(grant.ask.ref,))]
    for state in grant.own_beliefs.scope("party:norway:held"):
        parts.append(
            Part(
                "belief",
                Register.considered,
                state.value,
                rank=2,
                sources=(StateRef.of(state),),
            )
        )
    for stored, offer in grant.room.events():
        parts.append(
            Part(
                "offer",
                Register.considered,
                f"{offer.party}: {offer.text}",
                rank=1,
                sources=(EventRef(stored.seq, stored.envelope.event_id),),
            )
        )
    return parts

A Part's content stays structured: a string here, but any JSON value is allowed. How it becomes text is your renderer's decision. rank orders parts within a register for the budget, and a higher rank survives longer.

4. Assemble within a budget

assemble runs the assembler over the grant, checks the result, and fits it to a Budget. The budget's count is your token counter; this example counts characters:

assemble_prompts.py (continued)

budget = Budget(total=120, count=lambda part: len(str(part.content)))
context = assemble("norway", norway_assembler, norway_grant(store.last_seq()), budget=budget)
prompt = render_labelled(context)
print(prompt)
print("shed:", [p.slot for p in context.shed])
print("read:", len(context.read))

MUST HOLD:
Respond to the latest offer in one sentence.

CONSIDER:
UK will settle at 8%
UK: A 12% cut to the mackerel quota.
shed: []
read: 3

With total=80 the parts no longer fit, and the budget sheds the lowest-ranked considered part, the offer, first. A binding part is never shed. The default shed order is considered, then generated, then craft; pass order= to change it.

render_labelled is a plain default that writes each register under a heading. Replace it with your own renderer; the wording is not part of any contract.

5. Send the prompt through the pinned model

Rendering is deterministic, so the same context gives the same bytes and a replay hits its pins. Send the result through PinnedModel as usual:

assemble_prompts.py (continued)

def norway_model(messages: list[ModelMessage], info: AgentInfo) -> ModelResponse:
    return ModelResponse(parts=[TextPart("Norway can accept 6%, no more.")])


ledger = PinLedger(store, types)
print(Agent(PinnedModel(FunctionModel(norway_model), ledger)).run_sync(prompt).output)
ledger.commit()

Any change to what the assembler builds changes the prompt, and therefore the key: an assembly change shows up on replay as a PinMiss, not as a silently different answer.

Verify it worked

context.read lists every source the grant's slices handed out during this assembly, whether or not a part cites it. It is filled by the slices themselves, so it does not depend on the assembler being honest. That makes it the thing to test when the question is whether a prompt leaked:

read = context.describe()["read"]
assert all(source.get("scope") != "party:uk:held" for source in read)

context.describe() returns the whole context (parts, shed parts and sources) as plain data, ready for a log line or for the golden test below.

Check the assembler against a golden

The library cannot check that your assembler is pure, and it ships no golden-test helper. Your own tests can do the check: build a context over a fixed log at a fixed position, and compare context.describe() with a file you committed. Comparing structure (parts, registers, ranks, sources) rather than rendered text keeps the test stable when only the wording of your renderer changes.

One thing gets in the way. The sources in describe() carry event ids, and a belief's id carries the ids of its entities. These are fresh uuids every time the fixture log is built, so a golden compared as it stands fails on the next run. There are two ways round it:

• Replace the ids with stable labels before comparing, as below. Each uuid becomes id-1, id-2 and so on, in the order it first appears, so the test still sees which sources are the same.
• Load the same log every time. Appending stored events keeps their event ids, and entity ids are in the payloads, so a log saved once and loaded into a fresh store gives identical ids and seqs.

This pytest test imports the fixture log, the grant and the assembler from assemble_prompts.py above. Importing that script runs it once, against its in-memory store: it prints the example prompt and commits a pin, which moves the end of the log, so the test reads at a fixed position. In your own project, keep the fixture log, the grants and the assemblers in a module that does nothing when imported. Run the test once with UPDATE_GOLDEN=1 to write the golden, review the file, and commit it:

test_norway_golden.py

import json
import os
import re
from pathlib import Path

from assemble_prompts import budget, norway_assembler, norway_grant

from anagrapheus.assembly import assemble

GOLDEN = Path(__file__).parent / "goldens" / "norway.json"
UUID = re.compile(r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}")


def stable_ids(text: str) -> str:
    """Replace each uuid with a label numbered in order of first appearance."""
    labels: dict[str, str] = {}
    return UUID.sub(lambda m: labels.setdefault(m[0], f"id-{len(labels) + 1}"), text)


def test_norway_context_matches_golden() -> None:
    # Read at a fixed position, not at the end of the log, which pins move.
    context = assemble("norway", norway_assembler, norway_grant(3), budget=budget)
    actual = stable_ids(json.dumps(context.describe(), indent=2)) + "\n"
    if os.environ.get("UPDATE_GOLDEN") == "1":
        GOLDEN.parent.mkdir(exist_ok=True)
        GOLDEN.write_text(actual)
    assert actual == GOLDEN.read_text()

The golden it writes, goldens/norway.json:

goldens/norway.json

{
  "role": "norway",
  "upto_seq": 3,
  "parts": [
    {
      "slot": "ask",
      "register": "binding",
      "rank": 0,
      "content": "Respond to the latest offer in one sentence.",
      "sources": [
        {
          "given": "ask",
          "event": null
        }
      ]
    },
    {
      "slot": "belief",
      "register": "considered",
      "rank": 2,
      "content": "UK will settle at 8%",
      "sources": [
        {
          "state": "v1:fisheries:position(id-1,id-2)",
          "scope": "party:norway:held",
          "event": "id-3"
        }
      ]
    },
    {
      "slot": "offer",
      "register": "considered",
      "rank": 1,
      "content": "UK: A 12% cut to the mackerel quota.",
      "sources": [
        {
          "event": "id-4",
          "seq": 3
        }
      ]
    }
  ],
  "shed": [],
  "read": [
    {
      "event": "id-4",
      "seq": 3
    },
    {
      "state": "v1:fisheries:position(id-1,id-2)",
      "scope": "party:norway:held",
      "event": "id-3"
    },
    {
      "given": "ask",
      "event": null
    }
  ]
}

When the test fails, the assembler built a different context from the same log. If the change was intended, rerun with UPDATE_GOLDEN=1 and review the diff of the golden before committing it. A change that alters the rendered prompt also shows up on replay, as a pin miss.

Put chronicle prose in a prompt

A long run cannot put its whole log into every prompt. The chronicle exists for this: older rounds are read as summarised sections (depth), and only the events since the last section are read verbatim (recency). In a prompt, both halves come through slices in one grant:

• a ChronicleSlice, built as ChronicleSlice(chronicle, upto_seq=...) over the chronicle store, returns the sections that end at or before its position;
• a LogSlice returns the recent events, read with events(since_seq=...) from the end of the last section, so nothing the chronicle already covers is repeated.

Any part built from a section must be Register.generated and must cite the section, so that summarised prose can never pass as canon.

This example continues chronicle.py from How to Consolidate History into Chronicle Sections, after step 3: round 1 (events 1 to 3) is consolidated as section 1-3, and the UK's offer at seq 4 is still recent. Norway's prompt reads both:

chronicle.py (continued)

from dataclasses import dataclass

from anagrapheus.assembly import (
    Budget,
    ChronicleSlice,
    EventRef,
    Given,
    LogSlice,
    Part,
    Register,
    SectionRef,
    assemble,
    render_labelled,
)


@dataclass(frozen=True)
class NegotiatorGrant:
    past: ChronicleSlice
    room: LogSlice
    ask: Given[str]


def negotiator_grant(upto_seq: int) -> NegotiatorGrant:
    return NegotiatorGrant(
        past=ChronicleSlice(chronicle, upto_seq=upto_seq),
        room=LogSlice(
            reader,
            upto_seq=upto_seq,
            admit=lambda env: (env.namespace, env.type) == ("fisheries", "OfferMade"),
        ),
        ask=Given("ask", "Respond to the latest offer in one sentence."),
    )


def negotiator(grant: NegotiatorGrant) -> list[Part]:
    parts = [Part("ask", Register.binding, grant.ask.value, sources=(grant.ask.ref,))]
    # Depth: the consolidated past, as generated prose citing its section.
    through = 0
    for section in grant.past.sections():
        anchor = section.anchor
        source = SectionRef(anchor.section_id, anchor.from_seq, anchor.to_seq)
        parts.append(Part("past", Register.generated, section.prose, sources=(source,)))
        through = max(through, anchor.to_seq)
    # Recency: only the offers after the last section, verbatim from the log.
    for stored, offer in grant.room.events(since_seq=through):
        parts.append(
            Part(
                "offer",
                Register.considered,
                f"{offer.party}: {offer.text}",
                sources=(EventRef(stored.seq, stored.envelope.event_id),),
            )
        )
    return parts


budget = Budget(total=200, count=lambda part: len(str(part.content)))
context = assemble("norway", negotiator, negotiator_grant(store.last_seq()), budget=budget)
prompt = render_labelled(context)
print(prompt)
print("read:", [type(source).__name__ for source in context.read])


def norway_model(messages: list[ModelMessage], info: AgentInfo) -> ModelResponse:
    return ModelResponse(parts=[TextPart("Norway can accept 6%, no more.")])


print(Agent(PinnedModel(FunctionModel(norway_model), ledger)).run_sync(prompt).output)
ledger.commit()

MUST HOLD:
Respond to the latest offer in one sentence.

CONSIDER:
UK: We could move to 10%.

SUMMARISED (advisory prose, not canon):
The UK opened at 12%; Norway held at 5%.
read: ['SectionRef', 'EventRef', 'GivenRef']
Norway can accept 6%, no more.

context.read shows one event read from the log: the two round 1 offers reached the prompt only as the section that summarises them. The split point comes from the sections the slice returned, not from a separate query, so the two halves are always read at the same position. If consolidation falls behind, the last section ends earlier and more of the log is read verbatim; nothing is lost or repeated.

render_labelled puts generated parts under SUMMARISED (advisory prose, not canon):. Under the default shed order, the budget sheds considered parts before generated ones; pass order= to Budget if old history should go first.

Troubleshooting

StaleRead: the belief fold is at 4 and this slice reads at 5

A belief event lies between the fold's position and the slice's, so the context would mix beliefs from either side of it. If the fold is behind, run projector.catch_up() before building the grant at store.last_seq(). If it is ahead, a belief event was appended and folded while the grant was being read; build a new grant. Appends that carry no belief do not cause this: committing a PinLedger appends PinPromoted events, which no belief fold claims, so you can assemble straight after it without catching up.

The same error, ending the log ends before 5, means the slice was built at a position the log has not reached.

NotGranted: scope 'party:uk:held' is not in this grant

The assembler asked a belief slice for a scope its visible function rejects. This is the firewall working. Fix the assembler, or widen the grant if the role really may read that scope.

UncitedSource: part 'made-up' cites EventRef(…), which this grant never handed out

A part cites something its slices did not return during this call. Cite only references built from what the slices returned.

MislabelledPart: part 'past' cites chronicle prose as considered

Chronicle prose must be Register.generated. The same error is raised when a section was read but no generated part cites it, so leaving the citation off is not a way round it.

BudgetExceeded: the binding parts alone come to 44 against a total of 10

The binding parts alone exceed the budget. Binding parts are never shed, so the call is refused rather than sent without its invariants. Raise the budget or make the binding parts smaller.

AssemblyError: this grant was already read from

A grant was reused. Build a new grant for every assembly. The same exception type, with a different message, is raised when the slices in one grant read at different positions.

Related

• How to Keep Per-Agent Beliefs as a Fold: the scopes a belief slice reads.
• How to Consolidate History into Chronicle Sections: the sections a chronicle slice reads.
• How to Pin Model and Tool Calls: the pinned model the rendered prompt goes through.
• Prompt Assembly: why a role reads only what it was granted.