Storage

ProtoLink provides pluggable storage for Agent state, process-local caches and durable execution records. Persistence depends on the selected backend: SQLiteStorage survives restarts, while InMemoryStorage intentionally does not.

from protolink import Agent
from protolink.storage import SQLiteStorage

agent = Agent(
    name="helper",
    llm="mock",
    storage=SQLiteStorage("memory.sqlite", namespace="helper"),
    state=["conversation"],
)
print(agent.sync.invoke("Hello", session_id="customer-42"))

SQLite persists this session's model history in a local file. InMemoryStorage keeps data in the current process. Pass your configured backend through storage=; state modules decide what to load and save.

Need	Attach or use	What it preserves
Conversation memory or application key/value data	`storage=SQLiteStorage(...)`	Data used by enabled state modules and application code
Inspect and compare past runs	`run_store=SQLiteRunStore(...)`	Task snapshots and diagnostic run reports
Restore a filesystem change	`StorageCheckpointStore` with filesystem tools	Original resource bytes and recovery records
Resume unfinished execution	`durability="runs.sqlite"` or `SQLiteDurableStore`	Pending actions, responses, committed results and execution progress

See Execution, approvals and recovery for the workflows using these stores. A saved conversation or report alone does not make an interrupted operation resumable.

Storage Types

Protolink currently supports the following storage implementations:

SQLiteStorage - namespaced JSON key/value persistence in a local SQLite database.
InMemoryStorage - process-local object storage with optional sliding time-to-live expiration.
SQLiteRunStore - indexed task snapshots and run reports for replay, audit and regression workflows.
StorageCheckpointStore - recovery records and filtered inventory over a dedicated Storage namespace, imported from protolink or protolink.core.resources.
SQLiteDurableStore - fenced execution checkpoints for restarting paused or interrupted runs, imported from protolink or protolink.storage.

You can implement a custom state backend by subclassing Storage, or implement the structural RunStore protocol when diagnostic execution records belong in an application database. Implement DurableStore when your application supplies its own execution checkpoint backend; its ownership and commit requirements are explained in durable execution.

Configuration

Using storage with an agent is straightforward:

Instantiate the Storage implementation:

from protolink.storage import SQLiteStorage

storage = SQLiteStorage(db_path="agent_memory.db", namespace="my_agent")

Pass the Storage instance to your Agent:

from protolink.agents import Agent
from protolink.models import AgentCard

agent_card = AgentCard(url="http://localhost:8020", name="memory_agent", description="Agent with long-term memory")

agent = Agent(
    card=agent_card,
    transport="http",
    storage=storage,
    state=["conversation"],  # Enables persistence for specific modules
)

Namespacing

The namespace parameter in SQLiteStorage allows you to isolate data for different agents or contexts within the same database file.

Storage API Reference

This section provides a detailed API reference for the Storage module.

Unified Storage Interface

Protolink provides a consistent CRUD interface for all storage backends. Whether you are using SQLite, a cloud database, or a simple JSON file, you interact with them through the same standard methods: save(), load(), update() and delete().

Persistence moduleStorage

The storage surfaces used for namespaced Agent state, process-local TTL values, registry persistence and indexed execution records.

protolink.storage

CRUD interfaceSQLite implementationIn-memory TTLNamespaced dataDurable run recordsCustom backends

Base classDefines the common save, load, update and delete contract for storage backends.Storage

SQLitePersists JSON-serialized data under a namespace inside a local SQLite database.SQLiteStorage

MemoryKeeps arbitrary Python objects in-process with optional sliding expiration.InMemoryStorage

Run recordsIndexes task snapshots and normalized run reports separately from agent state.SQLiteRunStore

Agent stateAutomatically persists conversation history and gives the other state modules a shared storage extension point.state=[...]

Generic storage contract

Storage

abstract classprotolink.storage.Storage

source

class Storage

Define the synchronous, single-value persistence contract used by ProtoLink state modules. A storage instance represents one logical namespace: callers save or replace its complete value, load it, or delete it.

Abstract methods

save(data: Any) -> None: Persist a complete namespace value.
load() -> Any: Return the stored value, normally None when absent.
update(data: Any) -> None: Replace or otherwise update the namespace according to backend semantics.
delete() -> None: Remove the namespace value.

Synchronous interface

These methods are ordinary blocking functions. A database or network-backed custom implementation should manage blocking I/O appropriately when called from an asynchronous Agent.

Storage.save

abstract methodprotolink.storage.Storage.save

source

save(
  data: Any,
) -> None

Persist the supplied value as the current contents of this storage namespace.

Parameters

dataAnyrequired: Backend-specific value. Implementations decide whether it must be serializable, whether it is copied and whether saving replaces existing data.

Returns

NoneNone: Persistence is performed for its side effect.

Storage.load

abstract methodprotolink.storage.Storage.load

source

load() -> Any

Load the current namespace value.

Returns

dataAny: Backend-specific value, conventionally None when no value is stored. The interface cannot distinguish an absent value from an explicitly stored None.

Storage.update

abstract methodprotolink.storage.Storage.update

source

update(
  data: Any,
) -> None

Update the namespace using backend-specific semantics. Both built-in implementations make this exactly equivalent to save(); it is not a partial dictionary merge.

Parameters

dataAnyrequired: New complete value for the built-in backends.

Returns

NoneNone: Updating is performed for its side effect.

Storage.delete

abstract methodprotolink.storage.Storage.delete

source

delete() -> None

Remove the current namespace value.

Returns

NoneNone: Built-in deletion is idempotent: deleting a missing namespace is not an error.

SQLite key/value storage

SQLiteStorage

classprotolink.storage.SQLiteStorage

source

class SQLiteStorage(
  db_path: str = "storage.db",
  table_name: str = "storage",
  namespace: str = "default",
)

Persist one JSON-serializable value per namespace in a small SQLite table. Construction validates the table identifier, opens or creates the database and creates the table when it is missing.

Parameters

db_pathstrdefault: "storage.db": SQLite database path. SQLite creates the file when possible; the parent directory itself is not created by this class.
table_namestrdefault: "storage": Table containing key and JSON value columns. It must satisfy Python's str.isidentifier() check before it is interpolated into SQL.
namespacestrdefault: "default": Primary key for this storage instance's value. Multiple instances can share a database and table while using different namespaces.

Attributes

db_pathstr: Configured database path.
table_namestr: Validated table identifier.
namespacestr: Active row key.

Raises

ValueError: Raised for a table name that is not a valid Python identifier.
sqlite3.Error: Database creation, connection, schema, permission and locking errors propagate.

Connection model

Construction and every CRUD call open a short-lived SQLite connection. ProtoLink does not configure busy timeouts, WAL mode, migrations, encryption, or cross-process coordination for this minimal adapter.

SQLiteStorage.save

methodprotolink.storage.SQLiteStorage.save

source

save(
  data: Any,
) -> None

JSON-encode a value and insert or replace the row for the active namespace.

Parameters

dataAnyrequired: Value accepted by Python's json.dumps(). Tuple and other JSON conversions follow standard-library behavior.

Returns

NoneNone: The transaction is committed before the method returns.

Raises

TypeError / ValueError: JSON serialization errors propagate before the database write.
sqlite3.Error: Connection, locking, statement and commit errors propagate.

SQLiteStorage.load

methodprotolink.storage.SQLiteStorage.load

source

load() -> Any

Read and JSON-decode the current namespace row.

Returns

dataAny: Deserialized JSON value, or None when the namespace has no row. An explicitly saved JSON null also loads as None.

Raises

json.JSONDecodeError: Raised if another writer or manual edit stored invalid JSON.
sqlite3.Error: Database connection and query errors propagate.

SQLiteStorage.update

methodprotolink.storage.SQLiteStorage.update

source

update(
  data: Any,
) -> None

Replace the namespace value by calling save(data). The operation is an upsert, so it also creates a row that does not already exist.

Parameters

dataAnyrequired: Complete JSON-serializable replacement value.

Returns

NoneNone: Returns after the delegated save transaction commits.

SQLiteStorage.delete

methodprotolink.storage.SQLiteStorage.delete

source

delete() -> None

Delete the row for the active namespace and commit the transaction.

Returns

NoneNone: Missing rows are ignored, making deletion idempotent.

Raises

sqlite3.Error: Connection, locking, statement and commit errors propagate.

In-memory storage

InMemoryStorage

classprotolink.storage.InMemoryStorage

source

class InMemoryStorage(
  namespace: str = "default",
  ttl: int | None = None,
  store: dict[str, tuple[Any, float]] | None = None,
  ttl_heap: list[tuple[float, str]] | None = None,
)

Keep arbitrary Python objects in a dictionary-backed namespace. Loads use sliding expiration: a successful access refreshes the entry timestamp and pushes a new expiration marker when TTL is enabled.

Parameters

namespacestrdefault: "default": Dictionary key owned by this wrapper.
ttlint | Nonedefault: None: Sliding lifetime in seconds. None disables expiration. Values are stored without validation, so zero or negative values expire on the next sufficiently later access.
storedict[str, tuple[Any, float]] | Nonedefault: None: Optional backing dictionary of values and last-touch timestamps. When omitted, all default instances share a class-level process store.
ttl_heaplist[tuple[float, str]] | Nonedefault: None: Optional min-heap of expiration timestamps and namespaces. Pair a custom store with its own heap whenever TTL is enabled; otherwise save/load markers for the custom store enter the unrelated class-global heap.

Attributes

namespacestr: Active dictionary key.
ttlint | None: Sliding expiration configured on this wrapper.

Shared objects and TTLs

Values are stored and returned by reference, not copied. The default backing dictionary and heap are shared across instances and are not synchronized for concurrent threads. cleanup_expired() applies the calling instance's TTL while inspecting the shared heap, so instances sharing a backing store should use one consistent TTL policy.

InMemoryStorage.save

methodprotolink.storage.InMemoryStorage.save

source

save(
  data: Any,
) -> None

Store the object reference with the current wall-clock timestamp and, when TTL is enabled, push its calculated expiration onto the heap.

Parameters

dataAnyrequired: Any Python object. No serialization or copy is performed.

Returns

NoneNone: Existing namespace values are replaced.

InMemoryStorage.load

methodprotolink.storage.InMemoryStorage.load

source

load() -> Any

Return the current object when present and unexpired. Successful reads refresh the timestamp, making the configured TTL idle-based rather than a fixed time-since-save limit.

Returns

dataAny: The exact stored object reference, or None when missing or expired. An expired entry is removed lazily.

Touch semantics

Every successful load rewrites the entry timestamp. With TTL enabled it also adds a new heap marker; stale older markers are discarded later by cleanup_expired().

InMemoryStorage.update

methodprotolink.storage.InMemoryStorage.update

source

update(
  data: Any,
) -> None

Replace or create the namespace value through save(data), resetting its last touch and expiration marker.

Parameters

dataAnyrequired: Complete replacement object.

Returns

NoneNone: The method has the same side effects as save().

InMemoryStorage.delete

methodprotolink.storage.InMemoryStorage.delete

source

delete() -> None

Remove the active namespace from the backing dictionary.

Returns

NoneNone: Missing namespaces are ignored. Existing heap markers are left in place and discarded as stale during future cleanup.

InMemoryStorage.cleanup_expired

methodprotolink.storage.InMemoryStorage.cleanup_expired

source

cleanup_expired() -> int

Pop elapsed heap markers and proactively remove entries whose latest stored timestamp is also older than the calling instance's TTL.

Returns

removedint: Number of backing-store entries deleted. Returns zero when the heap is empty, TTL is disabled on the caller, or only stale markers were found.

Complexity

timeO(M log N): M is the number of elapsed heap markers and N is heap size.
spaceO(N): Repeated touches can temporarily create multiple markers for one namespace.

Durable execution records

Generic Storage holds the current value for one state namespace. RunStore solves a different problem: it preserves indexed task snapshots and normalized run reports so an application can retrieve, replay, compare, or audit past executions.

View runs in Devtools

Open a local run store with protolink dashboard --store runs.db --open.

TaskRecord

frozen dataclassprotolink.storage.TaskRecord

source

class TaskRecord(
  task_id: str,
  state: str,
  run_id: str | None = None,
  session_id: str | None = None,
  trace_id: str | None = None,
  agent_name: str | None = None,
  task: dict[str, Any] = field(default_factory=dict),
  metadata: dict[str, Any] = field(default_factory=dict),
  created_at: str | None = None,
  updated_at: str = field(default_factory=utc_now),
)

Represent the searchable index fields and serialized payload for one persisted task snapshot.

Parameters

task_idstrrequired: Stable task identifier and SQLite primary key.
statestrrequired: Serialized task lifecycle state.
run_idstr | Nonedefault: None: Correlated logical execution run.
session_idstr | Nonedefault: None: Correlated application or conversation session.
trace_idstr | Nonedefault: None: Correlated observability trace.
agent_namestr | Nonedefault: None: Agent name supplied when the snapshot was saved.
taskdict[str, Any]default: {}: Serialized Task.to_dict() payload.
metadatadict[str, Any]default: {}: Caller-owned index-record metadata, separate from fields inside the task.
created_atstr | Nonedefault: None: Timestamp copied from the task when available.
updated_atstrdefault: utc_now(): Timestamp of this persisted snapshot.

Shallow immutability

Field assignment is blocked, but the nested task and metadata dictionaries remain mutable.

TaskRecord.to_dict

methodprotolink.storage.TaskRecord.to_dict

source

to_dict() -> dict[str, Any]

Return all task-record fields in a serialization-friendly mapping.

Returns

recorddict[str, Any]: New outer dictionary containing the ten record fields. Nested task and metadata dictionaries are reused rather than deep-copied.

RunReportRecord

frozen dataclassprotolink.storage.RunReportRecord

source

class RunReportRecord(
  run_id: str,
  session_id: str | None = None,
  trace_id: str | None = None,
  agent_name: str | None = None,
  report: dict[str, Any] = field(default_factory=dict),
  metadata: dict[str, Any] = field(default_factory=dict),
  created_at: str = field(default_factory=utc_now),
)

Represent the indexed identity and serialized payload for one persisted RunReport.

Parameters

run_idstrrequired: Logical run identifier and SQLite primary key.
session_idstr | Nonedefault: None: Session copied from the report context when available.
trace_idstr | Nonedefault: None: Trace copied from the report context when available.
agent_namestr | Nonedefault: None: Agent name supplied by the saving application.
reportdict[str, Any]default: {}: Serialized RunReport.to_dict() payload.
metadatadict[str, Any]default: {}: Caller-owned record metadata.
created_atstrdefault: utc_now(): Timestamp at which the record was persisted.

Shallow immutability

The record is frozen, while its nested report and metadata dictionaries remain mutable references.

RunReportRecord.to_dict

methodprotolink.storage.RunReportRecord.to_dict

source

to_dict() -> dict[str, Any]

Return all run-report record fields in a new outer mapping.

Returns

recorddict[str, Any]: Mapping containing run/session/trace/agent identity, report payload, metadata and creation time. Nested mappings are not deep-copied.

RunStore

protocolprotolink.storage.RunStore

source

class RunStore(Protocol)

Define the structural interface an Agent can use for durable task and report records. Implementations do not need to inherit from this protocol and the protocol itself cannot be instantiated.

Task methods

save_task(Task, *, context=None, agent_name=None, metadata=None) -> TaskRecord: Persist or replace a task snapshot.
get_task(task_id: str) -> Task | None: Reconstruct a task payload by ID.
get_task_record(task_id: str) -> TaskRecord | None: Load the serialized record and index metadata by ID.
list_task_records(*, limit=100, session_id=None, run_id=None, state=None, agent_name=None) -> list[TaskRecord]: Query recent task records using optional indexed filters.

Report methods

save_report(RunReport, *, run_id=None, agent_name=None, metadata=None) -> RunReportRecord: Persist or replace a normalized run report.
get_report(run_id: str) -> RunReport | None: Reconstruct a report by run ID.
get_report_record(run_id: str) -> RunReportRecord | None: Load the serialized report record by ID.
list_report_records(*, limit=100, session_id=None, agent_name=None) -> list[RunReportRecord]: Query recent report records using optional indexed filters.

Protocol scope

Deletion is not required by RunStore. SQLiteRunStore adds delete_task() and delete_report() as concrete administrative extensions.

SQLiteRunStore

classprotolink.storage.SQLiteRunStore

source

class SQLiteRunStore(
  db_path: str | Path = "runs.db",
  *,
  table_prefix: str = "protolink",
  read_only: bool = False,
  redaction_policy: RedactionPolicy | None = None,
)

Implement RunStore with two SQLite tables: one for task snapshots and one for run reports. JSON payload columns retain serialized objects after the optional redaction policy is applied, while relational columns index common lookup fields.

Parameters

read_onlybooldefault: False: Open an existing database for inspection without creating tables or permitting writes.
redaction_policyRedactionPolicy | Nonedefault: None: Apply this policy to every saved task/report payload and caller metadata before writing. Live objects, index columns and existing rows remain unchanged. Reads return the stored representation.
db_pathstr | Pathdefault: "runs.db": SQLite database path, converted to str. The database file and schema are created when missing; parent directories are not created.
table_prefixstrdefault: "protolink": Prefix used to create <prefix>_tasks and <prefix>_run_reports. It must satisfy str.isidentifier().

Attributes

db_pathstr: Normalized database path.
table_prefixstr: Validated table prefix.
tasks_tablestr: Resolved task table name.
reports_tablestr: Resolved run-report table name.

Raises

ValueError: Raised for an invalid table-prefix identifier.
sqlite3.Error: Connection, schema creation, index creation, permission and locking errors propagate.

Replacement semantics

Tasks are keyed by task_id and reports by run_id. Saves use SQLite INSERT OR REPLACE, so a repeated identifier replaces the full prior row.

SQLiteRunStore.save_task

methodprotolink.storage.SQLiteRunStore.save_task

source

save_task(
  task: Task,
  *,
  context: RunContext | None = None,
  agent_name: str | None = None,
  metadata: dict[str, Any] | None = None,
) -> TaskRecord

Serialize and upsert one task snapshot together with indexed run correlation.

Parameters

taskTaskrequired: Task whose ID, state, creation time and complete to_dict() payload are persisted.
contextRunContext | Nonedefault: None: Explicit run/session/trace source. When omitted, RunContext.from_task(task) derives one from task metadata.
agent_namestr | Nonedefault: None: Optional indexed agent identity.
metadatadict[str, Any] | Nonedefault: None: Optional record metadata. The store copies it and applies its configured redaction policy before serialization.

Returns

recordTaskRecord: Frozen record matching the row that was committed.

Raises

serialization error: Errors from task serialization or json.dumps() propagate.
sqlite3.Error: Connection, statement, locking and commit errors propagate.

SQLiteRunStore.get_task

methodprotolink.storage.SQLiteRunStore.get_task

source

get_task(
  task_id: str,
) -> Task | None

Load a task record and reconstruct its domain model.

Parameters

task_idstrrequired: Primary-key identifier.

Returns

taskTask | None: Task.from_dict(record.task), or None when no row exists.

Raises

deserialization or sqlite error: Invalid stored JSON/task payloads and database failures propagate.

SQLiteRunStore.get_task_record

methodprotolink.storage.SQLiteRunStore.get_task_record

source

get_task_record(
  task_id: str,
) -> TaskRecord | None

Load the indexed record without reconstructing a Task.

Parameters

task_idstrrequired: Primary-key identifier.

Returns

recordTaskRecord | None: Parsed task record, or None when absent.

SQLiteRunStore.list_task_records

methodprotolink.storage.SQLiteRunStore.list_task_records

source

list_task_records(
  *,
  limit: int = 100,
  session_id: str | None = None,
  run_id: str | None = None,
  state: str | TaskState | None = None,
  agent_name: str | None = None,
) -> list[TaskRecord]

Query task records with conjunctive optional filters, ordered by newest updated_at first.

Parameters

limitintdefault: 100: SQLite result limit. Values are not validated; zero returns no rows and SQLite treats a negative limit as unbounded.
session_idstr | Nonedefault: None: Match one session exactly.
run_idstr | Nonedefault: None: Match one run exactly.
statestr | TaskState | Nonedefault: None: Match the enum's .value or the string representation of the supplied value.
agent_namestr | Nonedefault: None: Match one stored agent name exactly.

Returns

recordslist[TaskRecord]: Matching records in descending lexicographic ISO timestamp order.

SQLiteRunStore.save_report

methodprotolink.storage.SQLiteRunStore.save_report

source

save_report(
  report: RunReport,
  *,
  run_id: str | None = None,
  agent_name: str | None = None,
  metadata: dict[str, Any] | None = None,
) -> RunReportRecord

Serialize and upsert one complete run report.

Parameters

reportRunReportrequired: Report serialized through to_dict(). Its context supplies session and trace index fields when present.
run_idstr | Nonedefault: None: Optional primary-key override. When omitted, the report context's run ID is used.
agent_namestr | Nonedefault: None: Optional indexed agent identity.
metadatadict[str, Any] | Nonedefault: None: Optional record metadata, shallow-copied before serialization.

Returns

recordRunReportRecord: Frozen record matching the committed row.

Raises

ValueError: Raised when neither an explicit run ID nor a report-context run ID is available.
serialization or sqlite error: Report serialization, JSON encoding, database and commit errors propagate.

Run ID override

An explicit run_id changes the record key; it does not rewrite the run ID already serialized inside the report payload.

SQLiteRunStore.get_report

methodprotolink.storage.SQLiteRunStore.get_report

source

get_report(
  run_id: str,
) -> RunReport | None

Load a report record and reconstruct the RunReport.

Parameters

run_idstrrequired: Primary-key identifier.

Returns

reportRunReport | None: RunReport.from_dict(record.report), or None when absent.

SQLiteRunStore.get_report_record

methodprotolink.storage.SQLiteRunStore.get_report_record

source

get_report_record(
  run_id: str,
) -> RunReportRecord | None

Load the indexed and serialized record without reconstructing a RunReport.

Parameters

run_idstrrequired: Primary-key identifier.

Returns

recordRunReportRecord | None: Parsed record, or None when absent.

SQLiteRunStore.list_report_records

methodprotolink.storage.SQLiteRunStore.list_report_records

source

list_report_records(
  *,
  limit: int = 100,
  session_id: str | None = None,
  agent_name: str | None = None,
) -> list[RunReportRecord]

Query recent run-report records with optional session and agent filters.

Parameters

limitintdefault: 100: SQLite result limit; it is not range-validated.
session_idstr | Nonedefault: None: Match one session exactly.
agent_namestr | Nonedefault: None: Match one agent name exactly.

Returns

recordslist[RunReportRecord]: Matching records ordered by newest created_at first.

SQLiteRunStore.delete_task

methodprotolink.storage.SQLiteRunStore.delete_task

source

delete_task(
  task_id: str,
) -> None

Delete one task snapshot by primary key.

Parameters

task_idstrrequired: Task identifier to remove.

Returns

NoneNone: The transaction commits even when no matching row existed.

SQLiteRunStore.delete_report

methodprotolink.storage.SQLiteRunStore.delete_report

source

delete_report(
  run_id: str,
) -> None

Delete one run report by primary key.

Parameters

run_idstrrequired: Run identifier to remove.

Returns

NoneNone: The transaction commits even when no matching row existed.

Checkpoint recovery records

StorageCheckpointStore

classprotolink.StorageCheckpointStore

source

StorageCheckpointStore(storage: Storage)

Adapt a dedicated Storage namespace to the CheckpointStore protocol for durable resource recovery. One live writer owns the namespace. SQLiteStorage survives process exit; InMemoryStorage does not.

Parameters

storageStoragerequired: Namespace used exclusively for resource changes. The adapter exposes it as storage. Protect its original bytes and keep it separate from redacted run snapshots and mutable Agent state.

CheckpointStore contract

save(change: ResourceChange)None: Persist a complete record under its change_id. Repeated saves replace that record while keeping its insertion position. A durable save must complete before resource mutation begins.
get(change_id: str)ResourceChange | None: Return a detached recovery record, or None when absent. Bytes and nested metadata remain intact.
list_changes(...)list[ResourceChange]: Query detached records using the filters and pagination described below.

CheckpointStore is a structural protocol in protolink.core.resources. Custom backends implement these methods; inventory never executes a continuation or changes the recovery state. Storage and decoding errors propagate.

StorageCheckpointStore.list_changes

methodprotolink.StorageCheckpointStore.list_changes

source

list_changes(
  *,
  limit: int = 100,
  offset: int = 0,
  state: str | None = None,
  resource_id: str | None = None,
  run_id: str | None = None,
  task_id: str | None = None,
) -> list[ResourceChange]

Load the namespace once and list records in reverse insertion order. All supplied filters match exactly and combine with AND. Pagination applies after filtering. Updating a record does not move it and separate page requests are independent reads of the namespace.

Parameters

limitintdefault: 100: Maximum returned records. Zero returns an empty list.
offsetintdefault: 0: Matching records to skip before collecting results.
statestr | Nonedefault: None: Exact stored state, such as prepared, applied, restoring, restored, failed, or uncertain. Inspection preserves the stored state.
resource_idstr | Nonedefault: None: Exact resource identity in before.revision.resource_id; the query does not resolve or inspect paths.
run_idstr | Nonedefault: None: Run that performed the original write.
task_idstr | Nonedefault: None: Task that performed the original write.

Returns

changeslist[ResourceChange]: Detached records containing before/after snapshots, state, change/action/run/task IDs, optional errors, and restoration correlation fields. Snapshots include original bytes and POSIX modes. Restoration IDs are available as restore_run_id and restore_task_id; run/task filters target the original write.

Raises

ValueError: A limit or offset is negative.
storage or decoding error: Backend read failures and invalid stored recovery records propagate.

Recovery data

Results require the same access protection as get(). Apply redaction only to presentation copies; raw records remain necessary for restoration. Inventory neither resolves uncertainty nor resumes interrupted work.

Usage Examples

Standalone Usage

You can use the storage module independently of the agent system.

from protolink.storage import SQLiteStorage

# Initialize storage
storage = SQLiteStorage(db_path="data.db", namespace="user_settings")

# Save some data
settings = {"theme": "dark", "notifications": True}
storage.save(settings)

# Load data later
loaded_settings = storage.load()
print(loaded_settings["theme"])  # Output: dark

# Update data
loaded_settings["notifications"] = False
storage.update(loaded_settings)

# Delete data
storage.delete()

In-Memory TTL Usage

InMemoryStorage is useful when state should disappear with the process or after a period of inactivity:

from protolink.storage import InMemoryStorage

cache = InMemoryStorage(namespace="session-42", ttl=300)
cache.save({"messages": 4})

# A successful load refreshes the five-minute idle timeout.
value = cache.load()

# Use the same backing store, heap and TTL policy when pruning shared entries.
removed = cache.cleanup_expired()

Durable Run Records

Use SQLiteRunStore for execution history rather than mutable Agent state:

from protolink import Message, RunContext, Task
from protolink.storage import SQLiteRunStore

run_store = SQLiteRunStore("runs.db")
task = Task.create(Message(role="user").add_text("prepare the release notes"))
context = RunContext(run_id="release-2026-07", session_id="release")

record = run_store.save_task(
    task,
    context=context,
    agent_name="release-writer",
    metadata={"environment": "staging"},
)

recent = run_store.list_task_records(session_id="release", limit=20)

Redaction at persistence

from protolink import RedactionPolicy, SQLiteRunStore

policy = RedactionPolicy(sensitive_values=frozenset({"known-credential-value"}))
run_store = SQLiteRunStore("runs.db", redaction_policy=policy)
# Pass run_store=run_store to every Agent that should save protected snapshots.

The configured policy masks copies of all task/report payloads and caller metadata before SQLite writes, including intermediate Agent snapshots, nested events, tool output, errors and final task data. Returned index records contain the same masked payloads. Live Tasks, Reports and model/tool inputs are unchanged. With redaction_policy=None, the existing raw persistence contract remains available. Policies are supplied when opening the store and are not saved in the database.

Sensitive field names are masked recursively. sensitive_values additionally masks literal, case-sensitive occurrences of known secrets in free text, with longer matches first and before optional truncation. The default policy masks data_base64 recovery fields in presentation copies. It does not discover unknown secrets in arbitrary text.

Run/task/session/trace identifiers and other relational index columns remain stable, so keep secrets out of those fields. Existing database rows are not retroactively sanitized. ApprovalBroker and StorageCheckpointStore retain their separate protected storage: do not redact recovery bytes or the exact approval specification used for execution.

Agent Memory Integration

Agents can use the storage field to persist their state, conversation context, or learned information.

from protolink.agents import Agent
from protolink.storage import SQLiteStorage


class PersistentAgent(Agent):
    async def handle_task(self, task):
        # Load previous state
        state = self.storage.load() or {"count": 0}

        # Increment a counter
        state["count"] += 1

        # Save updated state
        self.storage.save(state)

        return await super().handle_task(task)

State System Integration

ProtoLink includes a unified State system. When you provide a storage instance and enable conversation, the conversation module automatically performs whole-payload load() and save() operations. The tools, task and flow modules currently expose storage-backed extension points; applications define their own persistence conventions on top.

Module	Storage Usage
conversation	Stores a serialized map of `session_id` to `ConversationHistory` lists.
tools	Retains the shared Storage reference but currently exposes no public persistence methods.
task	Retains the shared Storage reference but currently exposes no public persistence methods.
flow	`to_dict()` reads the shared storage payload; applications own any write/checkpoint convention.

This high-level system is the recommended way to manage LLM conversation persistence. Enabled modules currently receive the same Storage object rather than hidden per-module namespaces, so applications combining module-specific data should partition it explicitly.

Error Handling

Storage errors are intentionally visible to the caller:

Connection and SQL errors: context managers close short-lived SQLite connections, but sqlite3.Error subclasses still propagate.
Serialization errors: non-JSON-serializable data raises from json.dumps() before the write commits.
Deserialization errors: malformed stored JSON raises from json.loads().
File permissions and missing parent directories: both SQLite adapters rely on the configured path being writable; neither creates parent directories.
Concurrent writes: SQLite locking and busy errors are not retried by the storage classes.

Storage Types​

Configuration​

Storage API Reference​

Generic storage contract​

Storage​

Abstract methods

Storage.save​

Parameters

Returns

Storage.load​

Returns

Storage.update​

Parameters

Returns

Storage.delete​

Returns

SQLite key/value storage​

SQLiteStorage​

Parameters

Attributes

Raises

SQLiteStorage.save​

Parameters

Returns

Raises

SQLiteStorage.load​

Returns

Raises

SQLiteStorage.update​

Parameters

Returns

SQLiteStorage.delete​

Returns

Raises

In-memory storage​

InMemoryStorage​

Parameters

Attributes

InMemoryStorage.save​

Parameters

Returns

InMemoryStorage.load​

Returns

InMemoryStorage.update​

Parameters

Returns

InMemoryStorage.delete​

Returns

InMemoryStorage.cleanup_expired​

Returns

Complexity

Durable execution records​

TaskRecord​

Parameters

TaskRecord.to_dict​

Returns

RunReportRecord​

Parameters

RunReportRecord.to_dict​

Returns

RunStore​

Task methods

Report methods

SQLiteRunStore​

Parameters

Attributes

Raises

SQLiteRunStore.save_task​

Parameters

Returns

Raises

SQLiteRunStore.get_task​

Parameters

Returns

Raises

SQLiteRunStore.get_task_record​

Parameters

Returns

SQLiteRunStore.list_task_records​

Parameters

Storage Types

Configuration

Storage API Reference

Generic storage contract

Storage

Storage.save

Storage.load

Storage.update

Storage.delete

SQLite key/value storage

SQLiteStorage

SQLiteStorage.save

SQLiteStorage.load

SQLiteStorage.update

SQLiteStorage.delete

In-memory storage

InMemoryStorage

InMemoryStorage.save

InMemoryStorage.load

InMemoryStorage.update

InMemoryStorage.delete

InMemoryStorage.cleanup_expired

Durable execution records

TaskRecord

TaskRecord.to_dict

RunReportRecord

RunReportRecord.to_dict

RunStore

SQLiteRunStore

SQLiteRunStore.save_task

SQLiteRunStore.get_task

SQLiteRunStore.get_task_record

SQLiteRunStore.list_task_records

SQLiteRunStore.save_report

SQLiteRunStore.get_report

SQLiteRunStore.get_report_record

SQLiteRunStore.list_report_records

SQLiteRunStore.delete_task

SQLiteRunStore.delete_report

Checkpoint recovery records

StorageCheckpointStore

StorageCheckpointStore.list_changes

Usage Examples

Standalone Usage

In-Memory TTL Usage

Durable Run Records

Redaction at persistence

Agent Memory Integration

State System Integration

Error Handling