How it works

This page is for when you want to know what happens under the hood: how the source is organised, what one agent.run() call does step by step, and which interfaces you implement to plug in your own model, memory store or other component. You don’t need any of it to use loomflow.

Where things live

How the layers depend on each other

Dependencies point downward. Higher layers use lower ones, not the other way round:

agent/, workflow/, team.py              the public API
  ↓
architecture/                           loop styles
  ↓
governance/, observability/, security/  cross-cutting policies
  ↓
mcp/, tools/, skills/                   tool dispatch
  ↓
runtime/, memory/, model/               execution, storage, providers
  ↓
core/                                   types and protocols, depends on nothing

Every cross-layer call goes through an interface in core/protocols.py, which is what lets tests replace any layer with a fake. (Model wrappers are the one notable exception: RetryingModel and FallbackModel reuse the retry policy from governance/.)

Lifecycle: what happens during agent.run("hi")

1. Agent.run() checks its arguments (an unknown effort, a negative output_validation_retries or a non-string user_id fail here), applies the agent’s timeout, and calls the internal _loop(prompt, ...); stream() runs the same loop and forwards its events.
2. Open runtime session: async with self._runtime.session(session_id): sets a contextvar that journaled runtimes use to key their cache, scoped to the run’s user and the session’s current run epoch (a run inside another run keys under its parent). See Runs and epochs.
3. Open root span: async with self._telemetry.trace("loom.run"): (exported to OpenTelemetry as invoke_agent). The budget’s run clock starts here too (budget.running()), so wall-clock caps count only time spent running.
4. Audit: write run_started entry.
5. Seed context: pull working blocks, recall recent episodes, recall facts (when memory exposes .facts), build the messages: list[Message] to send to the model.
6. Loop (shown here for the default ReAct architecture; other architectures drive the same steps in a different order):
   • Stop if max_turns is reached. Check budget. Block / warn as needed.
   • With auto_compact_at_tokens set, compact the conversation if it has grown past the threshold, so the run’s own tool results are summarised while it runs.
   • Open loom.turn span.
   • Call the model inside a loom.model.complete span (run()), or stream it inside loom.model.stream (stream()), through the runtime so a journaled runtime can replay the step later. OpenTelemetry exports this span as chat {model} (kind CLIENT).
   • Aggregate text + tool_calls + usage from chunks. Emit each chunk as MODEL_CHUNK event.
   • If no tool calls, append assistant message and break.
   • Otherwise, dispatch tools in parallel inside an anyio.create_task_group. Each _run_single_tool opens its own loom.tool span (execute_tool {tool} in OpenTelemetry), runs hooks → permissions → sandboxed runtime.step("tool_call_<turn>_<slot>", tool_host.call, ...), journaled by that position plus the tool name and arguments. Audit tool_call and tool_result per call.
   • Append tool result messages to the conversation; loop again.
7. Persist episode: runtime.step("persist_episode_<turns>", memory.remember, episode). With persist_tool_transcripts=True the episode carries this run’s tool calls. When auto-extraction is on (the default for real models), the write also starts fact extraction as a background task that keeps running after run() returns; await agent.aclose() waits for it.
8. Compute session-duration metric.
9. Consolidate: run memory.consolidate() if Tuning(auto_consolidate=True). Failures surface as ERROR events but don’t break the run.
10. Audit: write run_completed entry with token / cost / elapsed payload.
11. Emit: Event.completed(...).

Every milestone hits four boundaries: events, telemetry, audit, runtime journal. Each can be independently configured; _noop_emit, NoTelemetry, audit_log=None, InProcRuntime all let you turn off anything you don’t need.

Streaming: how agent.stream() works

stream() runs the loop as a producer and yields what it sends:

async def _produce(send):
    try:
        await self._loop_guarded(prompt, emit=send, session_id=session_id, ...)
    except Exception as exc:
        # Shielded send so the consumer always sees ERROR before the error is raised
        with anyio.CancelScope(shield=True):
            await send(Event.error(session_id, exc))
        raise
 
events = stream_from(_produce, buffer_size=128)   # loomflow/core/streaming.py
try:
    async for event in events:
        yield event
finally:
    await events.aclose()

stream_from connects the two through a bounded memory stream, so backpressure is automatic: a slow consumer blocks send once 128 events are queued. On asyncio it runs the producer as a separate task, not inside a task group held open across yield. That matters when the consumer breaks out early: the generator is closed later, possibly from another task, and tearing down a task group’s cancel scope there would cancel the caller instead (its next await would raise CancelledError). Closing the stream cancels the producer, even mid-tool-call, and waits for it to unwind, so aclose() returns after the run’s cleanup has happened. If the producer raised, the events it already sent are yielded first and then its exception is raised as-is, not wrapped in an ExceptionGroup. trio has no unstructured spawn, so there a task group is still used; close the stream with contextlib.aclosing before abandoning it.

Extension points

Every cross-layer interface is a Protocol in core/protocols.py. To add a new backend, write a class with the methods the protocol lists; no base class is needed:

The framework only calls the methods the protocol defines, so any implementation works without forking loomflow.

Running the test suite

ruff check loomflow
mypy --strict loomflow
pytest tests/

CI runs all three on Python 3.11 and 3.12.

The live integration tests for the Postgres / Redis backends skip unless JEEVES_TEST_PG_DSN and JEEVES_TEST_REDIS_URL are set.

Test patterns by module:

• tests/test_smoke.py: Agent.run() end to end
• tests/test_streaming.py: backpressure and cancellation
• tests/test_tools.py: parallel dispatch, hook denials, max turns
• tests/test_journaled_runtime.py, tests/test_sqlite_runtime.py: replay
• tests/test_facts.py: supersession, valid_at, the consolidator
• tests/test_*_memory.py, tests/test_*_facts.py: one file per backend
• tests/test_telemetry.py: span hierarchy, metric routing
• tests/test_audit.py: HMAC verification, file persistence
• tests/test_sandbox.py: symlink escape detection
• tests/test_lineage.py: freshness and lineage policies
• tests/test_anthropic.py, tests/test_openai.py: chunk normalisation with fake clients
• tests/test_mcp.py: fake MCP sessions

Reading the source

A good order for reading the code:

1. core/types.py: the value objects everything else passes around.
2. core/protocols.py: the interfaces.
3. agent/api.py: the Agent class and its loop. It is long (about 3,700 lines); start at run() and _loop().
4. runtime/journaled.py: how replay works. Small but central.
5. memory/facts.py: how a new fact closes an older, contradicting one.
6. memory/consolidator.py: the fact-extraction prompt and parser.

Together that’s about 6,400 lines, and gives you a complete picture of how a run works.