Agent runtime (module)

ThreadRunner and related runtime helpers. Narrative: Agent Runtime.

Agent runtime for Agency.

Public API (this module — not yet split into subpackages):

• ThreadRunner — invoke / resume a Thread (creates LangGraph graph + Run)

• RunMiddleware — run lifecycle, token accounting, Monitor / channel events

• CheckpointSaver — DB-backed LangGraph checkpointer

• TokenUsage — token counters attached to a Run

Supporting types: State, Configurable.

Experimental / not product policy: agency.agent.policy (structured-output demos). Prefer ThreadRunner over ad-hoc graph scripts.

Built on LangChain / LangGraph (create_agent, optional deep agents).

class agency.agent.State

Bases: TypedDict

question: str

answer: str

docs: list[str]

class agency.agent.Configurable(config)

Bases: object

A class to wrap the configurable dictionary of the RunnableConfig.

__init__(config)

checkpoint_id: str | None

checkpoint_ns: str

class agency.agent.CheckpointSaver(*, serde=None)

Bases: BaseCheckpointSaver

Database-backed checkpoint saver for LangGraph executions.

This class persists LangGraph checkpoints and pending writes in the database so that graph execution can be paused, resumed, retried, and audited across process restarts.

It implements LangGraph’s checkpointing contract by storing: - the serialized graph state (checkpoint) - associated metadata - parent checkpoint linkage - pending writes (side effects not yet applied) - messages from the checkpoint state

This persistence layer is the foundation for: - human-in-the-loop interruptions - long-running or asynchronous agent workflows - deterministic resume via thread_id - full execution auditability

Conceptually: - Thread = identity of a long-lived execution context - Run = a concrete technical execution attempt - Checkpoint = snapshot of graph state at a point in time - CheckpointWrite = pending side effects (e.g. interrupts)

The saver is intentionally stateless; all execution state lives in the DB.

__init__(*, serde=None)

put(config, checkpoint, metadata, new_versions)

Persist a checkpoint produced by LangGraph.

This method is called by LangGraph whenever the graph reaches a point where its state should be snapshotted.

Responsibilities: - Serialize the checkpoint value using JsonPlusSerializer - Associate the checkpoint with a Thread (via thread_id) - Optionally associate it with a Run (via run_id, if provided) - Record metadata and parent checkpoint linkage - Return an updated config containing the checkpoint identifiers

The returned config may include: - checkpoint_id - checkpoint namespace which LangGraph can use internally for subsequent operations.

put_writes(config, writes, task_id, task_path='')

Persist pending writes generated during graph execution.

Pending writes represent side effects that were produced by a node but not yet applied to the graph state. Examples include: - interrupts (__interrupt__) - tool outputs - internal control signals

LangGraph uses a two-phase model: - First, nodes emit writes - Later, writes are committed and applied

This method stores those writes in the database so they can be: - replayed on resume - inspected for auditing/debugging - used to detect interruptions (e.g. human-in-the-loop)

Each write is stored with: - thread_id - checkpoint_id - channel name - serialized value - task metadata (task_id, task_path)

get_tuple(config)

Load a checkpoint tuple for LangGraph execution or resume.

This method reconstructs the full execution context expected by LangGraph when continuing or resuming a graph. It returns a CheckpointTuple containing: - the checkpointed graph state - metadata - parent checkpoint reference - pending writes that were not yet applied

The pending writes are critical: they allow LangGraph to correctly resume interrupted executions (e.g. human-in-the-loop) by replaying side effects such as ‘__interrupt__’.

The checkpoint is resolved using: - thread_id (required) - optional checkpoint namespace

If multiple checkpoints exist, the most recent one is selected.

This method is the authoritative source of truth for reconstructing graph execution state from persistent storage.

class agency.agent.TokenUsage(msg=None)

Bases: object

💰 Tokens usage helper class.

TODO: Check if we are counting multiple times the same messages.

__init__(msg=None)

total: int = 0

input: int = 0

output: int = 0

class agency.agent.RunMiddleware(run, interrupt_on=None)

Bases: AgentMiddleware

Middleware to wrap agent’s tools and model calls.

Manages the run lifecycle and token usage.

Publishes event updates to channels.

__init__(run, interrupt_on=None)

run

saved_message_ids: list[str]

presence

before_agent(state, runtime)

Execution starts.

check_new_message(message)

Check if a message is new (not yet seen in this run).

This is used to determine if a message should be sent to the user channel. The actual persistence is handled by CheckpointSaver when checkpoints are saved.

Returns:

True if message is new, False if already seen

wrap_model_call(request, handler)

Model call wrapper.

wrap_tool_call(request, handler)

Tool call wrapper.

after_agent(state, runtime)

Run ends.

class agency.agent.ThreadRunner(user=None, model=None, thread=None, deep=False)

Bases: object

ThreadRunner controls the execution of a LangGraph agent. Each invocation or resume advances the associated Thread forward.

Creates the executable graph instance (LangGraph agent) with:

• The Agent model and tools

• The CheckpointSaver to persist checkpoints and messages

• The RunMiddleware: Supervises the run lifecycle and token usage

Operational overview:

• Invoke: invokes the graph and checks for interruptions

• Run: Initialize the runner and create a new Run

• Resume: resumes the interrupted execution of the Thread

__init__(user=None, model=None, thread=None, deep=False)

presence: AgentPresence | None

Presence of the running Agent, owned by the middleware once a graph exists.

Interruption and failure are decided out here, not inside the graph, so both sides publish through the same object or they would fight over

resolve_model(agent)

Model for this run: explicit override, else the agent’s configured model.

create_graph(agent, run)

Compile a LangGraph agent graph.

run(agent, input, run=None, images=None)

Execute the mission: Invoke the graph with the given input.

This method performs a single execution attempt (“Run”) of the agent within the associated Thread. It:

1. Creates a Run record to track the technical execution attempt

2. Invokes the LangGraph agent with the current thread configuration

3. Persists checkpoints automatically via the CheckpointSaver

Handles three possible outcomes:

• SUCCESS:

  The graph completes normally and returns a final output.

• INTERRUPTION (human-in-the-loop):

  A tool calls interrupt(…), causing the graph to pause. The current graph state is already persisted as a Checkpoint. An Interruption record is created and execution is suspended until a human provides input.

• ERROR:

  An unexpected exception occurs during execution.

The checkpoint associated with an interruption is resolved by selecting the most recent checkpoint written for the current Run. This guarantees that execution can later be resumed from the exact point where the interruption occurred.

Returns:

the graph output (dict) - On interruption: a dict containing the interruption identifier - On error: raises the exception after persisting Run state

Return type:

• On success

resume(interruption)

Resume a previously interrupted execution.

Resuming continues the same Thread (same thread_id / timeline) from the latest persisted checkpoint, providing the human answer back to the graph via Command(resume=…).

This method creates a new Run to track the technical resume attempt. This keeps auditing clear: the original run may be INTERRUPTED, while the resume run may succeed, error, or interrupt again.

Parameters:

interruption (Interruption) – The interruption record containing the persisted checkpoint reference and the human answer payload.

Returns:

The graph output after resuming, which may complete successfully or produce new interruptions.

Return type:

dict

continue_turn(message, run=None, images=None)

Advance an existing Thread with a new user message (post-mission chat).

Uses the same thread_id / checkpointer so LangGraph appends to the prior timeline. Does not re-inject system instructions (already in checkpoint). Creates a new Run for audit.

invoke(input, config, agent, run)

Invoke the LangGraph agent and normalize outcomes into Run + Interruption persistence.

This is the shared execution primitive used by both run() and resume(). It: - Marks the run as RUNNING - Calls graph.invoke(…) - Detects interruptions via the ‘__interrupt__’ key in the output - Persists Interruption rows linked to the latest checkpoint for this run - Updates run status to SUCCESS / INTERRUPTED / ERROR

Parameters:

• input – Graph input (e.g. {“messages”: […]}) or a Command for resume.

• config (dict) – LangGraph config containing thread_id, checkpoint namespace, and run_id.

• run (Run) – The Run instance tracking this technical attempt.

Returns:

The raw graph output dict. If interrupted, the output includes ‘__interrupt__’ entries and corresponding Interruption rows are created.

Raises:

Propagates unexpected exceptions after persisting run error details. –

Return type:

dict

check_interrupt(run, output)

Check if the output contains an interrupt