Introduction to SolidState

SolidState is a high-reliability, lightweight agent state machine runtime. It is designed to serve as a "pedagogical scaffold"—a reference codebase that is small enough to read and comprehend in a single sitting, yet robust and comprehensive enough to run production-grade agentic workflows.

Why SolidState?

In the current ecosystem, most agent frameworks (such as LangChain, Semantic Kernel, or AutoGen) are massive, heavily abstracted, and often opaque. While they are useful for rapid prototyping, their abstractions hide the core underlying mechanics of how agents actually manage memory, interact with systems, handle transient failures, and maintain safety guardrails.

SolidState is built on the philosophy of "No Hidden Magic". It exposes the ten essential pillars of modern agent design directly, making the code highly legible and providing total control over execution.

Chapter 1: Philosophy & The State Machine Pattern

In classical software engineering, determinism is the baseline expectation. A program is a set of instructions that, given a specific input, will navigate a predictable set of execution branches to produce a predictable output. This is the foundation upon which databases, banking applications, and compiler designs are built.

The introduction of Large Language Models (LLMs) has shattered this baseline. LLMs are fundamentally probabilistic text completion engines. They do not guarantee the same response for identical inputs due to temperature-based sampling, and they are prone to subtle changes in formatting, hallucinated parameters, and runaway generation loops.

How do we build a highly reliable, deterministic system that relies on a probabilistic engine at its core?

The Agency vs. Control Paradox

When developers design autonomous agents, they often make the mistake of granting the LLM direct control over the execution loop. In such designs, the LLM is given access to a terminal, APIs, or database connections, and is left to choose its next step in a loose, self-directed loop.

This approach is highly fragile. If the model formats a tool argument incorrectly, hallucinates a database schema, or encounters a minor API error, the execution crashes. Worse, an unconstrained model can enter infinite loop states, calling the same API repeatedly, which leads to high cloud bills or service blocks.

SolidState resolves this tension through the separation of concerns:

The model reasons. The runtime controls execution.

In the SolidState architecture, the LLM is treated as an external, untrusted reasoning engine. It cannot execute code, directly touch network resources, or choose its own next node. Instead, the LLM only suggests transitions. The runtime is a strict Finite State Machine (FSM) that parses these suggestions, validates them against deterministic rules, and controls how the state transitions.

Mapping Code to State Machine Concepts

SolidState maps the core concepts of an FSM directly to Python code structures:

• State: The System Configuration: The current configuration of the machine is represented by the RuntimeState class. Rather than just tracking text, it maintains the absolute state of the execution: message queues, execution status flags ("running", "completed", "paused", "failed", or "rejected"), and pending tool configurations.
• Transitions: State Mutation Rules: The state machine can only transition between states based on strict, deterministic logic inside the AgentRuntime loop: pausing for interrupts, rejecting on policy blocks, or failing on runaway steps.

Chapter 2: The Core Loop Execution

In any software runtime, the execution engine determines how state changes are processed over time. In SolidState, this engine is represented by the AgentRuntime class. The core loop acts as a strict executor that controls the tick-by-turn lifecycle of an agent's reasoning process.

The Life of an Execution Turn

When a user calls await runtime.run(state), they are submitting a state machine to the runtime. The runtime executes a sequential processing loop. Each iteration represents a single reasoning turn, consisting of the following steps:

1. Context Window Preparation & Sliding: Before invoking the model, the runtime checks history sizes. If state.messages exceeds max_history, the window is trimmed while preserving the primary system prompt instruction at index 0.
2. Schema Harvesting: The runtime queries the ToolRegistry to retrieve JSON schemas of all tools currently available to the agent based on the current state.
3. Model Invocation with Retry Protection: The runtime invokes the model adapter. To handle transient failures, the invocation is wrapped in an exponential backoff retry handler (doubling delay times up to max_retries).
4. Decision Parsing & Action Routing: Resolves to either complete execution (if no tool is requested) or transitions to the policy verification phase.

import asyncio
from solidstate import AgentRuntime
from solidstate.checkpointing.file_checkpointer import FileCheckpointer
from solidstate.models.fake_model import FakeModelAdapter
from solidstate.policy.evaluator import PolicyEvaluator
from solidstate.tools.registry import ToolRegistry
from solidstate.tracing.tracer import JsonlTracer
from solidstate.types import RuntimeState

async def main():
    # 1. Initialize core system adapters
    registry = ToolRegistry()
    model = FakeModelAdapter()
    
    # 2. Wire up the AgentRuntime with checkpointer, policies, and tracer
    runtime = AgentRuntime(model, registry, PolicyEvaluator(), FileCheckpointer(), JsonlTracer())

    # 3. Create a state machine instance representing the query
    state = RuntimeState.new("Explain what an agent runtime does.")
    
    # 4. Execute the runtime loop
    result = await runtime.run(state)
    
    # 5. Output results
    print(result.status)
    print(result.messages[-1].content)

if __name__ == "__main__":
    asyncio.run(main())

Chapter 3: Message & State Representation

State modeling is the cornerstone of reliability. SolidState enforces a strict, typed state representation. This architecture decouples conversation content from execution metadata, providing a structured schema for the runtime.

The Anatomy of RuntimeState

The RuntimeState class is the single source of truth for an agent's execution lifecycle. It contains all the necessary data to serialize, inspect, or reconstruct a run:

• `run_id`: A unique UUID4 identifier generated at initialization.
• `status`: A strict execution status string ("running", "completed", "paused", "failed", or "rejected").
• `messages`: A list of Message objects capturing the conversation history.
• `tool_results`: A list of completed ToolResult objects.
• `pending_approvals`: A list of ToolCall objects currently waiting for human approval.
• `created_at`: A timezone-aware UTC ISO 8601 timestamp generated at creation time.

import asyncio
from solidstate import AgentRuntime
from solidstate.models.fake_model import FakeModelAdapter
from solidstate.policy.evaluator import PolicyEvaluator
from solidstate.tools.registry import ToolRegistry
from solidstate.checkpointing.file_checkpointer import FileCheckpointer
from solidstate.tracing.tracer import JsonlTracer
from solidstate.types import RuntimeState, Message, ModelResponse

async def main():
    class BalanceModel(FakeModelAdapter):
        async def invoke(self, messages, tools, state):
            balance_msg = next((m.content for m in messages if "balance is" in m.content), "unknown")
            return ModelResponse(content=f"Your balance is confirmed as: {balance_msg.split('is ')[-1]}")

    agent = AgentRuntime(BalanceModel(), ToolRegistry(), PolicyEvaluator(), FileCheckpointer(), JsonlTracer())
    state = RuntimeState.new("What is my balance?")
    
    # Inject context MANUALLY
    state.messages.append(Message(role="system", content="User balance is $5,400.20"))
    
    result = await agent.run(state)
    print(f"Status: {result.status}")
    print(f"Response: {result.messages[-1].content}")

if __name__ == "__main__":
    asyncio.run(main())

Chapter 4: The Tool Registry & Parameter Introspection

To interact with databases, filesystems, and external APIs, an agent needs tools. SolidState resolves the gap between natural language prompts and strict parameter requirements using automatic schema introspection and execution safety wrapping inside the ToolRegistry and ToolExecutor classes.

Python Introspection & Autoboxing

Instead of forcing developers to write redundant JSON schemas for their tools, SolidState leverages Python's type hints and Pydantic to generate schemas automatically. When you register a function, the registry inspects the function signature, parses docstrings, and compiles schemas. If the argument is a Pydantic model, it autoboxes inputs and validates them automatically before invoking the tool.

import asyncio
from pydantic import BaseModel
from solidstate import AgentRuntime
from solidstate.models.fake_model import FakeModelAdapter
from solidstate.policy.evaluator import PolicyEvaluator
from solidstate.tools.registry import ToolRegistry
from solidstate.checkpointing.file_checkpointer import FileCheckpointer
from solidstate.tracing.tracer import JsonlTracer
from solidstate.types import RuntimeState

class Person(BaseModel):
    name: str
    age: int
    city: str

def save_person(data: Person):
    return f"Saved {data.name} to DB"

async def main():
    registry = ToolRegistry()
    registry.register(save_person)

    model = FakeModelAdapter("save_person", {"data": {"name": "Alice", "age": 30, "city": "London"}})
    agent = AgentRuntime(model, registry, PolicyEvaluator(), FileCheckpointer(), JsonlTracer())

    state = RuntimeState.new("Extract person info: Alice is 30 years old.")
    result = await agent.run(state)
    print(f"Status: {result.status}")
    print(f"Result: {result.tool_results[0].data}")

if __name__ == "__main__":
    asyncio.run(main())

Chapter 5: Guardrails, Policies & Human-in-the-Loop

Executing model-generated tool calls blindly presents significant security risks. SolidState provides Tool Governance via policy gates. Before executing any tool call, the runtime sends the request to the PolicyEvaluator.

The evaluator executes rules and resolves to one of three decisions:

• Allow: The tool is deemed safe and executes immediately.
• Reject: The tool violates safety constraints. The run loop terminates immediately and shifts status to "rejected".
• Interrupt: The tool requires manual approval. The run loop is paused, and the status shifts to "paused". The pending tool call is added to the state's pending_approvals, and the checkpointer persists the paused state.

import asyncio
from solidstate import AgentRuntime
from solidstate.checkpointing.file_checkpointer import FileCheckpointer
from solidstate.models.fake_model import FakeModelAdapter
from solidstate.policy.evaluator import PolicyEvaluator
from solidstate.tools.registry import ToolRegistry
from solidstate.tracing.tracer import JsonlTracer
from solidstate.types import RuntimeState

def update_customer_limit(customer_id: str, new_limit: float) -> dict:
    return {"customer_id": customer_id, "new_limit": new_limit, "updated": True}

async def main():
    registry = ToolRegistry()
    registry.register(update_customer_limit, risk_level="high", requires_approval=True)

    model = FakeModelAdapter("update_customer_limit", {"customer_id": "C123", "new_limit": 25000.0})
    checkpointer = FileCheckpointer()
    runtime = AgentRuntime(model, registry, PolicyEvaluator(), checkpointer, JsonlTracer())

    state = RuntimeState.new("Update customer C123 limit to 25000.")
    paused = await runtime.run(state)
    print("Paused:", paused.status, paused.run_id)

    resumed = await runtime.resume_after_approval(paused.run_id, approved=True)
    print("Resumed:", resumed.status)

if __name__ == "__main__":
    asyncio.run(main())

Chapter 6: Multi-Agent Orchestration & The StateGraph

Single-agent systems often struggle when faced with complex tasks. As you add more tools, context grows, and the agent becomes less reliable. The solution is a division of labor: decomposing a complex problem into a network of specialized agents using the StateGraph engine.

A StateGraph represents a workflow as a directed graph where Nodes represent processing steps, Edges represent transitions, and Conditional Edges are routing functions that inspect state dynamically to determine transitions.

import asyncio
from solidstate import AgentRuntime
from solidstate.models.fake_model import FakeModelAdapter
from solidstate.policy.evaluator import PolicyEvaluator
from solidstate.tools.registry import ToolRegistry
from solidstate.checkpointing.file_checkpointer import FileCheckpointer
from solidstate.tracing.tracer import JsonlTracer
from solidstate.types import RuntimeState

async def main():
    policy = PolicyEvaluator()
    checkpointer = FileCheckpointer()
    tracer = JsonlTracer()

    support = AgentRuntime(FakeModelAdapter(None), ToolRegistry(), policy, checkpointer, tracer, name="support_agent")
    billing = AgentRuntime(FakeModelAdapter("refund_user", {"user_id": "U123"}), ToolRegistry(), policy, checkpointer, tracer, name="billing_agent")

    # Construct graph symbolically using >> operator
    workflow = support >> billing
    
    state = RuntimeState.new("Transfer me to billing for a refund.")
    final_state = await workflow.run(state)
    print(f"Final Status: {final_state.status}")

if __name__ == "__main__":
    asyncio.run(main())

Chapter 7: Concurrency & The Multi-threaded Actor Model

As agentic systems scale to handle complex workflows, running them sequentially becomes a performance bottleneck. To support concurrent execution, SolidState implements a thread-based **Actor Model** inside the `StateGraph` engine.

The Actor Model Design

In this model, "actors" do not share state, communicate exclusively by passing asynchronous messages, and process incoming messages sequentially from an inbox queue:

• Queue-Based Message Inbox: Every node in the StateGraph gets a thread-safe queue.Queue instance.
• Isolated Event Loops: Spawns nodes as background worker threads (ActorThread-{node_name}) running independent, thread-local event loops.
• State-Passing Communication: When an actor finishes, it pushes the updated RuntimeState to the recipient actor's queue.

Toggling Concurrency Mode

You have complete control over whether to run a workflow sequentially or concurrently using the multithreaded flag:

# 1. Sequential Mode (Default)
final_state = await workflow.run(state, multithreaded=False)

# 2. Multi-threaded Actor Mode
final_state = await workflow.run(state, multithreaded=True)

Chapter 8: Memory & State Persistence (Checkpointing)

State machines require persistent memory to allow pause/resume actions, time-travel debugging, and fault recovery. SolidState implements this through the Checkpointer interface.

The BaseCheckpointer defines the storage contract (save and load). Distributed caches like Redis provide shared state access with sub-millisecond latency.

import asyncio
from solidstate import AgentRuntime
from solidstate.checkpointing.redis_checkpointer import RedisCheckpointer
from solidstate.types import RuntimeState

async def main():
    try:
        checkpointer = RedisCheckpointer(host="localhost", port=6379)
        agent = AgentRuntime(FakeModelAdapter(), ToolRegistry(), PolicyEvaluator(), checkpointer, JsonlTracer())
        
        state = RuntimeState.new("Hello Redis!")
        await agent.run(state)
        
        loaded_state = await checkpointer.load(state.run_id)
        print(f"Loaded Run ID: {loaded_state.run_id}")
    except Exception as e:
        print(f"Redis Example skipped: {e}")

if __name__ == "__main__":
    asyncio.run(main())

Chapter 9: Observability, Observational Integrity & Tracing

In production, you must have complete visibility into LLM choices, safety policy evaluations, and tool actions. SolidState implements this through observational integrity.

Every internal transition, model response, policy decision, context trim, and tool output is recorded as a structured JSON event. The CloudWatchTracer streams events directly to standard output, enabling collectors to query metrics like latency, error rates, and blocks.

import asyncio
from solidstate import AgentRuntime
from solidstate.tracing.cloudwatch_tracer import CloudWatchTracer
from solidstate.types import RuntimeState

async def main():
    tracer = CloudWatchTracer(service_name="payment-agent")
    agent = AgentRuntime(FakeModelAdapter(), ToolRegistry(), PolicyEvaluator(), FileCheckpointer(), tracer)

    state = RuntimeState.new("Transfer $50")
    await agent.run(state)

if __name__ == "__main__":
    asyncio.run(main())

Chapter 10: Production Performance, Caching & Cloud Serverless

Deploying agents at scale requires managing model latency, reducing consumption costs, protecting host environments, and choosing infrastructure. SolidState decouples model providers via the ModelAdapter strategy pattern, allowing local Ollama configurations or cloud Bedrock setups.

To isolate execution security risks, we recommend executing code generation tools inside containers or WebAssembly (Wasm) sandboxes.

import asyncio
import time
from solidstate import AgentRuntime
from solidstate.models.cache import ResponseCache
from solidstate.types import RuntimeState

async def main():
    cache = ResponseCache()
    agent = AgentRuntime(FakeModelAdapter(), ToolRegistry(), PolicyEvaluator(), FileCheckpointer(), JsonlTracer(), cache=cache, max_history=4)

    state = RuntimeState.new("Hello")
    
    # Run 1 (Cold Cache)
    await agent.run(state)
    
    # Run 2 (Hot Cache)
    state = RuntimeState.new("Hello")
    await agent.run(state) # Resolves instantly

if __name__ == "__main__":
    asyncio.run(main())

Framework Unit Testing

The repository enforces tests checking parallel executions, state loaders, and actor concurrency:

1. Parallel Execution Verification

@pytest.mark.asyncio
async def test_runtime_parallel_tool_execution(tmp_path):
    # Mock slow tasks with 0.1s delay
    runtime = AgentRuntime(MultiToolModel(), registry, PolicyEvaluator(), checkpointer, tracer)
    
    start_time = asyncio.get_event_loop().time()
    result = await runtime.run(state)
    end_time = asyncio.get_event_loop().time()
    
    # Verify concurrent timing remains under 0.18s
    assert (end_time - start_time) < 0.18

2. Safety Policy Assertions

def test_policy_interrupts_for_approval():
    registry = ToolRegistry()
    registry.register(risky_tool, requires_approval=True)
    decision = PolicyEvaluator().evaluate(
        ToolCall(name="risky_tool", args={"amount": 100}),
        RuntimeState.new("test"),
        registry,
    )
    assert decision.action == "interrupt"

3. Actor-Thread Multithreaded Execution

Asserts that graphs execute correctly under the actor-thread model (spawning individual worker OS threads and running thread-local loop engines).

@pytest.mark.asyncio
async def test_state_graph_multithreaded_flow():
    graph = StateGraph()
    graph.add_node("A", node_a)
    graph.add_node("B", node_b)
    graph.set_entry_point("A")
    graph.add_edge("A", "B")
    
    state = RuntimeState.new("start")
    final_state = await graph.run(state, multithreaded=True)
    
    assert final_state.status == "completed"
    assert len(final_state.messages) == 3