agentkit.state.context¶
WorkingContext — the in-flight reasoning state (prefix, messages,
scratchpad, journal) — plus TokenCounter and the small prefix
helpers. (Scope lives in agentkit.core.contracts.types; see
API › kernel.)
agentkit.state.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.core.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 headmessages— per-turn LLM transcript tailscratchpad— cross-step notesjournal— append-only authored history (MutationJournal)
Long-term recall lives on the agent — Agent.memory: MemorySource
(see agentkit.state.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.
module-attribute
¶
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).
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.
scratchpad_changes is FROZEN at construction (see __post_init__),
which is what makes the "a ContextDiff is itself hashable /
pickleable" promise above true of the WHOLE value rather than of its
message tuples only. It is still a dict for every consumer; writing
into it raises.
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 is deterministic) whose VALUES are
deep-frozen (see __post_init__), 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 — see __hash__ for what "hashable" can
and cannot mean when two of those tuples hold arbitrary
user objects.
The one deliberate exception is the CONTENTS of journal_entries:
the tuple is frozen, the entries inside it are passed through by
identity. __post_init__ says why, with the measurement.
Round-tripping: WorkingContext(prefix=f.prefix,
messages=list(f.messages), scratchpad=dict(f.scratchpad)) is
functionally equivalent to the source (modulo identity). The
rehydrated context is a fully live one — dict(...) rebuilds a
plain, writable top level, so note / update_scratchpad work
as always; what it inherits is frozen NESTED values, which is the
point of taking a snapshot rather than a copy. A child that wants to
edit one builds a new value: ctx.note("plan", {**ctx.get("plan"),
"steps": [...]}).
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.
record(entry: JournalEntryT) -> None
Append one entry. Journal-semantics — never rewritten.
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.
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).
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.
property
¶
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.
async
¶
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() -> 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(
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. "By value" means __eq__:
the set below only uses Message.__hash__ to pick a
bucket, so two tool-call messages that share a call id but
differ in arguments collide and both survive, as they must.
That hash exists only because ToolCall defines one — this
line raised TypeError: unhashable type: 'dict' for every
message carrying tool calls, i.e. on every real fan-in.
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(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() -> FrozenContext
Immutable snapshot — safe to share across agent boundaries without locking. Equality is structural.
Scratchpad values are deep-frozen and un-aliased by
FrozenContext.__post_init__; read that docstring for what
"immutable" covers and what it deliberately does not.
This is the only one of the four composition seams that returns
something IMMUTABLE, and that is what decides how each treats
scratchpad values. fork() returns an independent live copy, so it
deepcopys them. merge() aggregates into a live context and
hands values over by reference — last-write-wins means the receiver
owns the object and keeps writing to it. slice() returns a live
view with a fresh top-level dict and shared values, as its docstring
says ("scratchpad ... inherited unchanged"). freeze() is the one
whose result nobody may write to, so it is the one that freezes.
async
¶
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.
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() -> 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.
dataclass
¶
AllOf(scopes: tuple[ContextScope, ...])
Conjunction — keep messages every scope keeps.
dataclass
¶
AnyOf(scopes: tuple[ContextScope, ...])
Disjunction — keep messages any scope keeps.
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.
A scope that genuinely cannot answer from (message, index)
alone — one whose answer depends on where the END of the list is —
should implement _surviving_indices as well; see
PositionalScope.
dataclass
¶
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.
dataclass
¶
Keeps messages with index ≥ checkpoint_index.
Index-dependent but NOT positional: the checkpoint is an absolute
coordinate the caller already knows, so matches() is the whole
truth and this scope composes correctly through matches()
alone. Only a scope that measures from the END of the list needs
the _surviving_indices hook.
dataclass
¶
Keeps messages whose name field matches the tag. Useful
for per-agent slicing in a team.
dataclass
¶
Default — chars/4 approximation over the whole wire payload.
Free, no deps, ~ok within 30% of actual for English. Use
TiktokenCounter when accuracy matters (right before a borderline
context-limit decision).
Shares :func:estimate_message_tokens with
capabilities.compaction.base._approx_tokens — the two were
separate copies of chars/4 and both silently ignored tool_calls,
which is how a pre-check here could pass while the compactor over
there refused to fire on the same transcript.
RequestBuilder.approx_tokens now delegates to the same function
too, so a pre-check, a built request and a compaction decision all
report the SAME number on a tool-heavy transcript.
dataclass
¶
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.
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.