State Management
Protolink's State system is a sophisticated, modular orchestration layer that manages the persistence of an agent's internal data. It bridges the gap between the high-level Agent logic and the low-level Storage backends, providing a unified API for session-based memory.
Why use the State system?
In a distributed agentic system, maintaining context is critical. Without state management, every interaction is a "cold start." The State system allows agents to:
- Resume conversations: Remember what was said minutes or days ago.
- Persist tool data: Allow tools to keep track of their own history or configuration.
- Track task progress: Monitor long-running tasks across multiple execution cycles.
- Coordinate flows: Manage checkpoints and state transitions in complex workflows.
Core Architecture
The State class acts as a central hub (orchestrator). When an agent is initialized, it creates a State instance which, in turn, initializes one or more State Modules. All enabled modules receive the exact same Storage object. The current state layer does not create a separate storage namespace per module; isolation is by module API and data convention, while the Storage instance's own namespace remains the physical persistence boundary.
State Modules
Protolink exposes four state module names through StateMode. You can enable them individually or in combination via the state parameter in the Agent constructor.
conversation is the fully integrated automatic runtime path today: agents load LLM history before inference and save it after normal completion. A failed turn is normally isolated, but history is retained when a new Artifact(kind="action_result") proves that a tool or delegation side effect completed and its observation must survive a retry. tools, task, and flow are available as typed module slots for persistent extension work; they share the same storage backend, but only flow currently exposes a small to_dict() helper and tool/task modules intentionally stay minimal.
1. Conversation State (conversation)
This is the most common module. It manages the ConversationHistory object used by LLMs.
- Data Saved: All messages (user, assistant, system, tool) in a session.
- Key Factor: Uses the
session_idprovided in task metadata to partition history. - Automatic Sync: The
Agentloads history before inference and saves it after normal completion, or after a failed turn that contains a completed-action receipt.
2. Tool State (tools)
Provides a dedicated module slot for tool-specific persistence.
- Usage: Useful for custom tools that need a shared storage handle for caches, counters, credentials, or external synchronization metadata.
- Current behavior: The module is initialized with the agent storage backend. Tool authors decide what APIs or conventions to add on top.
3. Task State (task)
Provides a dedicated module slot for task metadata persistence.
- Usage: Useful for applications that want to index, replay, or resume task-related metadata outside the in-memory
Taskobject. - Current behavior: Runtime task lifecycle transitions are managed on
Task.stateand recorded intask.metadata["state_history"]; the state module is a storage-backed extension point.
4. Flow State (flow)
Provides a storage-backed module for the Structured Flows architecture.
- Usage: Intended for checkpointing flow progress or storing workflow context across runs.
- Current behavior:
FlowState.to_dict()returns the serialized storage contents. Flow orchestration also usestask.flow_statefor per-task semantic context injection.
Activation and Configuration
Enabling state persistence requires two steps: providing a Storage instance and specifying the Enabled Modules.
Basic Setup (Conversation Only)
from protolink.agents import Agent
from protolink.storage import SQLiteStorage
# 1. Setup persistent storage
storage = SQLiteStorage(db_path="agent.db", namespace="support_bot")
# 2. Enable conversation module
agent = Agent(
card=card,
storage=storage,
state=["conversation"],
)
Advanced Setup (Multi-Module)
# Enable everything
agent = Agent(
card=card,
storage=storage,
state=["conversation", "tools", "task", "flow"],
)
Session Management
The State system relies on a session_id to know which data to load. Protolink handles this through Task Metadata.
Providing a Session ID
When sending a task, include a session_id in the metadata:
task = Task.create_infer("Hello, I'm Alice.")
task.metadata["session_id"] = "user_42_convo_A"
await agent.execute_task(task)
Default Behavior
If no session_id is provided:
invoke()/sync.invoke(): These methods use a default ID ("invocation_session_id"), ensuring that sequential calls to the same agent instance share history by default.- External Tasks: The agent falls back to using the
task.id. This effectively makes the task stateless across different task IDs, but persistent if the same task is updated and re-processed.
The State Object API
Construct State directly when application code needs manual module access, then retain that object while also passing it to Agent(state=state). The current Agent implementation stores it internally and exposes state operations through describe_state(), reset_state(), and compact_state(); it does not define a public agent.state property.
The optional state container that gives agents durable conversation, tool, task, and flow memory while keeping persistence explicit and inspectable.
protolink.statestate=[...]conversationstoragedescribe_state()State
State(
storage: Storage,
enabled: list[StateMode],
)Create the state orchestrator and instantiate the requested built-in modules in list order. Duplicate names replace the same dictionary entry rather than creating multiple stores.
Parameters
storageStoragerequired- Shared backend passed unchanged to every enabled module. State performs no runtime type validation.
enabledlist[StateMode]required- Any combination of
"conversation","tools","task", and"flow". An empty list creates a valid stateless orchestrator.
Raises
ValueError- An enabled name is not registered in
STATE_REGISTRY. module constructor error- Errors raised while binding a module to storage propagate.
State module properties
conversation: ConversationState | None
tools: ToolState | None
task: TaskState | None
flow: FlowState | None
storage: Storage
enabled_modes: tuple[StateMode, ...]Access or replace enabled module objects and inspect their shared backend.
Properties
conversationConversationState | None- Enabled conversation module, or
None. The setter accepts a replacement object without runtime validation; the getter returns it only when it is actually ConversationState. toolsToolState | None- Enabled tool extension slot, with the same setter/getter type behavior.
taskTaskState | None- Enabled task extension slot.
flowFlowState | None- Enabled flow extension slot.
storageStorage- Orchestrator backend reference.
enabled_modestuple[StateMode, ...]- Enabled names in fixed registry order: conversation, tools, task, then flow, not necessarily constructor-list order.
state.storage changes only the orchestrator's _storage reference. Existing ConversationState, ToolState, TaskState, and FlowState instances retain the Storage object they received at construction.State.describe
describe(
request: StateOperationRequest | None = None,
) -> StateOperationResultInspect requested stores without mutation. Omission creates a default request and reports every enabled module in deterministic registry order.
Parameters
requestStateOperationRequest | Nonedefault: None- Optional store selection, session scope, data-inclusion flag, and application metadata. Request metadata is not copied into the result by the current orchestrator.
Returns
resultStateOperationResult- One report per requested store plus disabled names in
missing. Conversation reports become session-scoped when a session ID is supplied.
to_dict() methods return {} for empty storage, so exists is true even when item_count is zero. Session-scoped conversation reports test the selected session directly.State.reset
reset(
request: StateOperationRequest | None = None,
) -> StateOperationResultClear one conversation session or delete the entire shared storage namespace, depending on request scope.
Parameters
requestStateOperationRequest | Nonedefault: None- A session ID defaults selection to conversation. Without a session, an empty store selection means every enabled mode.
Returns
resultStateOperationResult- Structured cleared, missing, and error reports. Unsupported partial resets are reported rather than raised.
storage.delete() once. A non-session subset that differs from all enabled modes is rejected because deleting the shared namespace would clear more than requested.State.to_dict
to_dict() -> dict[str, Any]Call to_dict() on each enabled module that implements it and return results keyed by module name. ToolState and TaskState are omitted because they currently expose no serializer. Because ConversationState and FlowState share storage and both serialize the whole payload, enabling both may duplicate the same data under two keys.
Manual State Interaction
from protolink.state import State
state = State(storage=storage, enabled=["conversation"])
agent = Agent(card=card, storage=storage, state=state)
# Get history manually
if state.conversation:
history = state.conversation.get_history("session_123")
# Clear a session
if state.conversation:
state.conversation.clear_session("session_123")
# View everything as a dict
all_data = state.to_dict()
Enabled Store APIs
ConversationState
ConversationState(
storage: Storage,
)Manage a dictionary of serialized ConversationHistory lists keyed by session ID.
Parameters
storageStoragerequired- Backend whose entire loaded payload is expected to be a mapping from session IDs to message lists.
storage.load() or and then mapping operations. A truthy non-dictionary payload raises at runtime.ConversationState.get_history
get_history(
session_id: str,
default_system_prompt: str | None = None,
) -> ConversationHistoryLoad and deserialize one session, or create a fresh history when the key is missing or its stored value is empty.
Parameters
session_idstrrequired- Exact dictionary key; no normalization or validation is applied.
default_system_promptstr | Nonedefault: None- System prompt used only for a newly created history.
Returns
historyConversationHistory- A reconstructed or new mutable history. Reading does not write it back.
Raises
storage or history error- Backend failures, incompatible payload shapes, and malformed serialized messages propagate.
ConversationState.save_history
save_history(
session_id: str,
history: ConversationHistory,
)Load the complete session mapping, replace one key with history.to_list(), and save the complete mapping.
Parameters
session_idstrrequired- Session key to create or replace.
historyConversationHistoryrequired- History serialized into its provider-neutral message-list representation.
Returns
NoneNone- The implementation has no explicit return annotation and returns
None.
ConversationState.clear_session
clear_session(
session_id: str,
)Delete one session key and save the remaining mapping. If the key is absent, return without calling storage.save().
Parameters
session_idstrrequired- Exact stored conversation-session key to remove.
ConversationState.to_dict
to_dict() -> dictReturn storage.load() directly, falling back to a new empty dictionary for falsey values. The returned mapping is not defensively copied.
ToolState / TaskState
ToolState(storage: Storage)
TaskState(storage: Storage)Bind the shared storage object as _storage. These are intentionally minimal extension slots: they currently expose no public persistence, retrieval, reset, or serialization methods of their own. TaskState is linked above; ToolState has the same constructor shape.
Parameters
storageStoragerequired- Backend retained for application-defined extensions.
FlowState
FlowState(
storage: Storage,
)Bind a storage-backed flow extension slot.
Parameters
storageStoragerequired- Shared backend.
FlowState.to_dict
to_dict() -> dictReturn the entire loaded storage payload or an empty dictionary. This helper does not serialize the transient Task.flow_state prompt used by structured-flow execution unless the application explicitly wrote that data to Storage.
State Control Plane
Agents expose typed state inspection and mutation operations for applications
that need to prove what state exists without reading private storage directly.
These methods are available locally on Agent and remotely through
AgentClient request specs.
from protolink import StateOperationRequest
report = await agent.describe_state("customer-42")
assert report.stores[0].name == "conversation"
reset = await agent.reset_state("customer-42")
assert "conversation" in reset.cleared
compacted = await agent.compact_state(
"customer-42",
strategy="tokens",
max_tokens=8_000,
)
The request and result models are immutable dataclasses designed to cross local, HTTP, WebSocket, runtime, and other transport boundaries.
StateOperationRequest
StateOperationRequest(
session_id: str | None = None,
stores: tuple[str, ...] = (),
include_data: bool = False,
strategy: HistoryCompactionStrategy = "tokens",
max_messages: int = 20,
max_tokens: int = 4000,
preserve_recent: int = 6,
summary_max_tokens: int = 512,
metadata: dict[str, Any] = field(default_factory=dict),
)Describe the scope and compaction settings for one state control-plane operation. The same type is reused for describe, reset, and compact; the receiving operation decides which fields apply.
Fields
session_idstr | Nonedefault: None- Optional session scope. Conversation is the only built-in session-keyed store.
storestuple[str, ...]default: ()- Requested stores. Empty delegates selection to the operation: enabled modes for describe/full reset and conversation for compact.
include_databooldefault: False- Include inspected payloads in describe reports.
strategyLiteral["recent", "tokens", "summary"]default: "tokens"- History compaction strategy.
max_messagesintdefault: 20- Positive message limit.
max_tokensintdefault: 4000- Positive estimated token ceiling.
preserve_recentintdefault: 6- Non-negative number of newest messages protected during token/summary compaction.
summary_max_tokensintdefault: 512- Positive requested summary length.
metadatadict[str, Any]default: {}- Application-owned request context, created with a per-instance default factory.
Raises
ValueError- Invalid strategy, limits below one, or negative
preserve_recent.
StateOperationRequest.to_dict / from_dict
to_dict() -> dict[str, Any]
StateOperationRequest.from_dict(
data: dict[str, Any] | None,
) -> StateOperationRequestSerialize tuples as lists or coerce a decoded mapping back into a validated request. from_dict(None) creates defaults; a string stores value becomes a one-element tuple, numeric limits pass through int(), and metadata is copied.
Parameters
datadict[str, Any] | Nonerequired- Decoded request mapping passed to
from_dict(); explicitNoneselects all request defaults.
StateStoreReport
StateStoreReport(
name: str,
enabled: bool,
exists: bool = False,
item_count: int | None = None,
message_count: int | None = None,
cleared: bool = False,
compacted: bool = False,
data: Any | None = None,
metadata: dict[str, Any] = field(default_factory=dict),
error: str | None = None,
)Report the observation or mutation outcome for one store.
Fields
namestrrequired- Requested store name.
enabledboolrequired- Whether the State orchestrator has that module.
existsbooldefault: False- Whether relevant store or session data exists under the orchestrator's reporting semantics.
item_countint | Nonedefault: None- Length for mapping, list, tuple, or set payloads.
message_countint | Nonedefault: None- Conversation-session message count when known.
clearedbooldefault: False- Reset mutation succeeded for this store.
compactedbooldefault: False- Compaction succeeded for this store.
dataAny | Nonedefault: None- Optional inspected payload when requested.
metadatadict[str, Any]default: {}- Operation-specific before/after or session-scope details.
errorstr | Nonedefault: None- Store-scoped non-exception failure.
from_dict() applies basic string, bool, integer, and dictionary coercion but likewise permits combinations such as enabled=False with cleared=True.StateStoreReport.to_dict / from_dict
to_dict() -> dict[str, Any]
StateStoreReport.from_dict(
data: dict[str, Any],
) -> StateStoreReportSerialize recursively with dataclasses.asdict() or reconstruct a report. Missing names become "unknown"; optional counts are converted with int().
Parameters
datadict[str, Any]required- Decoded store-report mapping passed to
from_dict().
StateOperationResult
StateOperationResult(
operation: Literal["describe", "reset", "compact"],
session_id: str | None = None,
stores: tuple[StateStoreReport, ...] = (),
cleared: tuple[str, ...] = (),
compacted: tuple[str, ...] = (),
missing: tuple[str, ...] = (),
errors: tuple[dict[str, str], ...] = (),
metadata: dict[str, Any] = field(default_factory=dict),
)Aggregate all per-store reports and operation-level outcome lists.
Fields
operationLiteral["describe", "reset", "compact"]required- Logical operation represented by the result.
session_idstr | Nonedefault: None- Target session when supplied.
storestuple[StateStoreReport, ...]default: ()- Per-store results.
clearedtuple[str, ...]default: ()- Stores cleared by reset.
compactedtuple[str, ...]default: ()- Stores compacted.
missingtuple[str, ...]default: ()- Requested stores not enabled or data not found, depending on the producing operation.
errorstuple[dict[str, str], ...]default: ()- Structured store/message failures.
metadatadict[str, Any]default: {}- Application or operation metadata.
from_dict() explicitly rejects operations outside describe, reset, and compact.StateOperationResult.to_dict / from_dict
to_dict() -> dict[str, Any]
StateOperationResult.from_dict(
data: dict[str, Any],
) -> StateOperationResultConvert tuple fields into transport-friendly lists and nested report dictionaries, or reconstruct the immutable result. Parsing copies error and metadata mappings so they are not shared with the decoded payload.
Parameters
datadict[str, Any]required- Decoded operation-result mapping passed to
from_dict().
Each StateStoreReport includes the store name, whether it is enabled, whether
state exists, item/message counts when known, and operation metadata. Passing
include_data=True to describe_state() includes the inspected payload in the
report for debugging or export workflows.
Remote State Operations
AgentClient uses the same control-plane pattern as cancellation and history
compaction:
report = await client.describe_state(agent_url, session_id="customer-42")
reset = await client.reset_state(agent_url, session_id="customer-42")
compacted = await client.compact_state(
agent_url,
session_id="customer-42",
strategy="recent",
max_messages=20,
)
The remote endpoints are:
| Operation | Endpoint | Capability |
|---|---|---|
describe_state() | POST /state/describe | state.describe |
reset_state() | POST /state/reset | state.reset |
compact_state() | POST /state/compact | state.compact and llm.history.compact |
Reset Semantics
Conversation state is session-keyed, so reset_state("customer-42") precisely
clears that conversation session. Calling reset_state() without a session ID
performs a full reset of the agent storage namespace for all enabled stores.
Partial full-store resets are rejected because the current storage abstraction
is namespace-based; ProtoLink reports that limitation instead of clearing more
state than requested.
compact_state() currently targets conversation state. It loads the persisted
session, runs the LLM-owned HistoryCompactor, saves the compacted history, and
returns before/after counts in the report metadata. The operation is still a
control-plane request and is never shown to the model as a tool.
Comparison: Manual vs. Automated State
Persistence Beyond A2A
A2A provides the task exchange model but does not prescribe application state storage. Without ProtoLink State, you load and save data inside handle_task yourself.
async def handle_task(self, task):
data = self.storage.load()
# ... logic ...
self.storage.save(data)
Automated Persistence (Protolink State)
Protolink handles the lifecycle for you.
# Just enable it in the constructor
agent = Agent(..., state=["conversation"])
# History is loaded and saved automatically in Agent.execute_task()
Design Philosophy
The State system is built on three pillars:
- Selective modules: Module APIs are decoupled and enabled independently. They currently share one Storage payload, so applications adding tool/task/flow persistence must still establish non-colliding keys or separate namespaces.
- Transparency: All data is eventually serialized into the same storage backend, making it easy to backup or migrate.
- Implicit Context: By using
session_idas a first-class citizen in metadata, Protolink creates sticky conversation context across workers when those workers use the same durable storage namespace.