Skip to content

agentkit.context

WorkingContext — the in-flight reasoning state (prefix, messages, scratchpad, journal) — plus TokenCounter and the small prefix helpers. (Scope lives in agentkit.kernel.types; see API › kernel.)

agentkit.context — in-flight reasoning state.

The context subpackage owns the data agents reason over. It is the third leg of agentkit's run-state tripod:

  • RunContext (agentkit.runtime) — execution wiring: identity, budgets, services, cooperative cancellation. The how a run runs.
  • WorkingContext (this package) — in-flight reasoning state. A unified state object with four orthogonal axes:
    • prefix — cache-stable system + grounding head
    • messages — per-turn LLM transcript tail
    • scratchpad — cross-step notes
    • journal — append-only authored history (MutationJournal)

Long-term recall lives on the agent — Agent.memory: MemorySource (see agentkit.memory). The cognition queries memory and grounds the prompt via the RequestBuilder seam.

This package is deliberately narrow: it owns the data shapes and the pure operations on them (slice, fork, merge, freeze, diff, record, journal-diff, token estimate). It does NOT compact, retrieve, or call the LLM — those policies stay with their owners (Compactor, MemorySource, Invoker).

The split mirrors the KV-cache discipline: the prefix is the cache-stable head, the messages tail is per-turn churn, the journal is structured authored history, and they never bleed into each other.

JournalEntryT module-attribute 

JournalEntryT = TypeVar('JournalEntryT')

The type of entries stored in the journal axis. Project code parameterises this with its own mutation/event union (typically a discriminated union of domain mutation kinds, each stamped with agent_id for attribution that survives merges).

ContextDiff dataclass 

ContextDiff(
    messages_added: tuple[Message, ...],
    messages_removed: tuple[Message, ...],
    scratchpad_changes: dict[str, Any],
    prefix_changed: bool,
)

A structural diff between two WorkingContext snapshots.

Returned by WorkingContext.diff(other). The shape is intentionally minimal — added / removed message tuples and a flat scratchpad-changes dict.

messages_added / messages_removed are tuples (frozen) so a ContextDiff is itself hashable / pickleable. Order is preserved from the source contexts.

scratchpad_changes is a dict mapping each changed key to its new value in self (the right-hand side). A key present in other but absent in self appears here with value None to signal removal; ambiguity with a None-valued live key is accepted for a debug helper.

prefix_changed is True when the two prefixes compare unequal.

FrozenContext dataclass 

FrozenContext(
    prefix: PrefixContext,
    messages: tuple[Message, ...],
    scratchpad: tuple[tuple[str, object], ...],
    journal_entries: tuple[Any, ...] = (),
)

Immutable snapshot of a WorkingContext at a point in time.

Use when handing a context across an agent boundary without sharing mutation: a parent that wants to brief a child but doesn't want the child's writes to leak back.

All fields are immutable: PrefixContext is frozen, messages is a tuple, scratchpad is a sorted tuple of (key, value) pairs (so equality + hash are deterministic), and journal_entries is a tuple. A snapshot is therefore safe to share across agent boundaries without locking and to use as a memoization-cache key.

Round-tripping: WorkingContext(prefix=f.prefix, messages=list(f.messages), scratchpad=dict(f.scratchpad)) is functionally equivalent to the source (modulo identity).

MutationJournal 

MutationJournal(
    initial: Iterable[JournalEntryT] | None = None,
)

Bases: Generic[JournalEntryT]

Append-only journal of typed entries + watermark for streaming.

Used by hierarchical agents to stream progress upward without re-shipping entries the parent has already absorbed. The watermark tracks how far the consumer (parent) has acknowledged:

  • record(entry) / apply(entries) — append; never rewrites.
  • diff() — uncommitted suffix (past the watermark).
  • mark_committed() — advance watermark; called after a parent has absorbed the diff.
  • view() — full read-only view (committed + uncommitted).

Attribution invariant: callers may stamp agent_id (or any other field) on entries before recording. The journal never rewrites entries — author attribution survives every merge into a parent's journal.

No locking — the journal is private to ONE agent. Cross-agent merging goes through signal protocols, not the journal directly.

Build a journal, optionally seeded with prior entries.

Seed entries are treated as already-committed (watermark starts past them) — typical use is re-hydrating a journal from a persisted snapshot where the entries have already been observed by the parent.

committed_index property 

committed_index: int

Diagnostic — how many entries have been ack'd.

record 

record(entry: JournalEntryT) -> None

Append one entry. Journal-semantics — never rewritten.

apply 

apply(entries: Iterable[JournalEntryT]) -> None

Merge a batch of entries into the journal in arrival order.

Used by the parent's merge loop to absorb a child's diff: each incoming entry keeps whatever attribution it already carries so the audit trail survives.

Idempotency: callers MUST not double-apply the same diff. The journal trusts the merge loop's ack-style protocol.

view 

view() -> tuple[JournalEntryT, ...]

All entries in record order (committed + uncommitted).

diff 

diff() -> list[JournalEntryT]

Uncommitted suffix — everything appended past the watermark.

has_uncommitted 

has_uncommitted() -> bool

True if there's anything past the watermark.

size 

size() -> int

Total entries recorded so far.

mark_committed 

mark_committed() -> None

Advance the watermark past every entry currently in the journal. Called after a parent has acknowledged absorbing the diff (or after the final-delta has been shipped).

WorkingContext dataclass 

WorkingContext(
    prefix: PrefixContext = PrefixContext(),
    messages: list[Message] = list(),
    scratchpad: dict[str, Any] = dict(),
    journal: MutationJournal[Any] = MutationJournal(),
    token_counter: TokenCounter = ApproxTokenCounter(),
    limit: int | None = None,
    shared: bool = False,
    _lock: Lock = Lock(),
)

An agent's working memory — four orthogonal axes.

Required

prefix — the cache-stable head (system + grounding + optional schema block). Empty PrefixContext() is the right default when none is set yet.

Axes (each optional, each independent): messages — per-turn LLM transcript tail. Mutating methods touch only this list. scratchpad — cross-step notes. Last-write-wins. journal — append-only authored history with watermark. Used by hierarchical agents.

Long-term recall is no longer a context slot — see Agent.memory: MemorySource | None (auto-wired into the RequestBuilder grounder).

Knobs

token_counter — pluggable TokenCounter for tokens(). Defaults to ApproxTokenCounter (chars/4). limit — hard token ceiling. None = no enforcement (caller's policy still owns the actual abort). shared — team-blackboard hint. False (default) → single-coroutine use; the lock is never touched. True → multi-coroutine blackboard. The simple sync mutators are still GIL-atomic and don't auto-lock; for MULTI-step atomicity (sequences with awaits between writes, read-modify-write) wrap in async with ctx.lock: or await ctx.apply_locked(fn).

A single-shot chat agent uses transcript + scratchpad only; the journal stays empty. A long-lived hierarchical agent uses all three axes.

lock property 

lock: Lock

The internal asyncio lock — exposed as a public primitive when shared=True so callers can wrap multi-step mutation sequences in async with ctx.lock: to keep cross-coroutine atomicity.

Acquire when EITHER of these holds: - A sequence of two or more mutations with an await between them must observe a consistent state to other coroutines. - A read-modify-write pattern (read scratchpad value, await some computation, write back) must not be lost to a concurrent writer.

Sync single-step mutators (append / note / etc.) do NOT need to be wrapped — they're GIL-atomic. The lock is for sequencing across awaits. shared=True does NOT auto-lock mutators; this property is the primitive callers use explicitly.

append 

append(*messages: Message) -> Self

Add messages to the transcript tail.

extend 

extend(messages: Iterable[Message]) -> Self

Add multiple messages to the transcript tail.

clear_messages 

clear_messages() -> Self

Clear the tail. Prefix, scratchpad, journal untouched.

note 

note(key: str, value: Any) -> Self

Set a scratchpad value (last-write-wins).

get 

get(key: str, default: Any = None) -> Any

Read a scratchpad value.

update_scratchpad 

update_scratchpad(data: dict[str, Any]) -> Self

Bulk-update the scratchpad.

apply_locked async 

apply_locked(fn: Callable[[Self], Any]) -> Self

Run a synchronous mutation closure under self.lock for cross-coroutine atomicity.

Example::

async def add_pair(ctx):
    ctx.append(user_msg)
    await some_async_validation(ctx)
    ctx.append(assistant_msg)

await ctx.apply_locked(add_pair)

Both appends and the validation in between are serialized against any other coroutine that also goes through apply_locked (or async with ctx.lock). Callers that DON'T need cross-coroutine safety should just call the mutators directly — this helper is opt-in.

Supports async closures too: if fn returns an awaitable, it is awaited inside the lock.

fork 

fork() -> WorkingContext

An independent copy — own tail, deep-copied scratchpad, new journal seeded from current entries.

The prefix is shared by reference (it's frozen — safe to share). The token counter / limit / shared flag are inherited; the lock is fresh.

merge 

merge(
    other: WorkingContext,
    *,
    mode: Literal["concat", "union"] = "concat",
) -> Self

Aggregate another context's knowledge into this one.

mode="concat" (default) appends the other's messages verbatim and bulk-updates the scratchpad (last-write-wins). Also applies the other's full journal entries — preserving each entry's attribution (the journal never rewrites).

mode="union" deduplicates messages by value before appending, so a parent merging two siblings doesn't end up with the same observation twice.

slice 

slice(scope: ContextScope) -> WorkingContext

Return a NEW WorkingContext containing only messages that match scope.

Prefix, scratchpad, and journal are all inherited unchanged — slicing is a view operation on the tail, not a re-shape of the other axes. Useful for parent→child briefings.

diff 

diff(other: WorkingContext) -> ContextDiff

Replay/debug helper: what changed between two snapshots.

Computes self - other semantics: messages in self but not other are messages_added; the reverse are messages_removed. Scratchpad entries follow the same logic with None signalling removal. prefix_changed is a structural equality check.

Journal is intentionally NOT included in the diff — the journal has its own watermark-based diff via journal.diff() which is the right semantic for streaming.

freeze 

freeze() -> FrozenContext

Immutable snapshot — safe to share across agent boundaries without locking. Equality is structural.

tokens async 

tokens() -> int

Current token usage estimate via the configured counter.

Includes prefix + messages. Async because the counter is async (it may call a remote counting endpoint or warm a lazy encoder); in the common ApproxTokenCounter path this is a single cheap await.

size 

size() -> int

Number of messages in the tail. Cheap — no token math.

assembled 

assembled() -> list[Message]

The full message list the provider sees: prefix.as_messages() + self.messages.

Returns a new list each call so a caller mutating it doesn't corrupt the live tail.

PrefixContext dataclass 

PrefixContext(
    system_prompt: str = "",
    grounding: tuple[Message, ...] = tuple(),
    schema_block: str | None = None,
)

The cache-stable head: system prompt + grounding chunks.

Why frozen: the per-turn churn is the messages tail; the prefix is the shared cacheable prefix every turn re-uses. Mutating it after construction would invalidate every provider's prompt cache from that token forward — exactly the opposite of what we want from a "context engineering" boundary.

schema_block is reserved for the structured-output prompt-injection track (Track B3). When set, it appears as a third pinned system message after the prompt and grounding.

Equality is structural (dataclass-generated). Two prefixes built from the same (prompt, grounding, schema) are equal and hash equal — useful when keying a recall cache.

as_messages 

as_messages() -> list[Message]

The prefix as a flat message list — pinned system prompt, followed by the grounding messages in order, followed (when present) by the structured-output schema block.

Returns a new list each call so a caller mutating the result can't accidentally aliasing-corrupt the frozen prefix.

AllOf dataclass 

AllOf(scopes: tuple[ContextScope, ...])

Conjunction — keep messages every scope keeps.

AnyOf dataclass 

AnyOf(scopes: tuple[ContextScope, ...])

Disjunction — keep messages any scope keeps.

ContextScope

Bases: Protocol

Predicate over messages — returns True when the message belongs in the sliced view. Composable via the obvious AllOf / AnyOf / Not combinators.

index is the message's index in the original message list (not the sliced output). This is what makes Since cheap and LastNTurns implementable.

LastNTurns dataclass 

LastNTurns(n: int)

Keeps the last n turns (a turn = consecutive user+assistant pair). System messages always survive — they're the cache-stable framing the loop relies on.

Implementation note: this scope can't decide membership purely from (message, index) — it needs the full list to count back n turns. WorkingContext.slice() pre-computes the set of surviving indices for any scope that exposes _surviving_indices, falling back to per-message matches() otherwise. The protocol surface stays simple; the windowing path is local to this class.

Not dataclass 

Not(scope: ContextScope)

Inversion — keep messages the inner scope drops.

RoleFilter dataclass 

RoleFilter(roles: frozenset[str])

Keeps only messages with role in roles.

Since dataclass 

Since(checkpoint_index: int)

Keeps messages with index ≥ checkpoint_index.

Tagged dataclass 

Tagged(tag: str)

Keeps messages whose name field matches the tag. Useful for per-agent slicing in a team.

ApproxTokenCounter dataclass 

ApproxTokenCounter(chars_per_token: float = 4.0)

Default — chars/4 approximation.

Free, no deps, ~ok within 30% of actual for English. Use TiktokenCounter when accuracy matters (right before a borderline context-limit decision). Matches the heuristic RequestBuilder uses for its approx_tokens field, so a RequestBuilder-side pre-check stays consistent with the WorkingContext-side view.

TiktokenCounter dataclass 

TiktokenCounter(encoding: str = 'cl100k_base')

Provider-accurate when tiktoken is installed.

Opt-in dep under arc-agentkit[fast]. Falls back to ApproxTokenCounter when tiktoken is absent or fails to load (no import error — just less accurate). The encoding name keys into tiktoken's registry; cl100k_base matches GPT-4/3.5 and is a reasonable default for cross-provider estimates.

TokenCounter

Bases: Protocol

Estimates token usage of a message list.

Pluggable so a cheap-but-fuzzy ApproxTokenCounter (chars/4) can be swapped for a provider-accurate TiktokenCounter without changing the WorkingContext API. Implementations MUST be pure-ish — same input, same output — so a caller can pre-check a budget against the same number the invoker will produce.