Struct SqliteTaskStore 

pub struct SqliteTaskStore { /* private fields */ }

SQLite-backed TaskStore.

Stores tasks as JSON blobs in a tasks table. Suitable for single-node production deployments that need persistence across restarts.

Schema

The store auto-creates the following table on first use:

CREATE TABLE IF NOT EXISTS tasks (
    id         TEXT PRIMARY KEY,
    context_id TEXT NOT NULL,
    state      TEXT NOT NULL,
    data       TEXT NOT NULL,
    updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%d %H:%M:%f','now'))
);

list() returns tasks most-recently-updated first (spec §3.1.4), ordered by (updated_at DESC, id DESC) with a composite row-value cursor. updated_at is written at millisecond precision in a fixed-width format so TEXT comparison matches chronological order.

Implementations

impl SqliteTaskStore

pub const fn with_max_page_size(self, max: u32) -> SqliteTaskStore

Caps the page size list returns, however large a page is asked for.

Defaults to DEFAULT_MAX_PAGE_SIZE, which explains why this store needs its own knob rather than reading TaskStoreConfig.

pub fn with_metrics(self, metrics: MetricsHandle) -> SqliteTaskStore

Sets where this store reports an event it could not record.

Defaults to NoopMetrics. An append is ON CONFLICT (task_id, seq) DO NOTHING, so one that lands on a position another writer already holds returns Ok(()) having written nothing — see event_append_error. This is how that becomes visible.

pub async fn new(url: &str) -> Result<SqliteTaskStore, Error>

Opens (or creates) a SQLite database and initializes the schema.

Errors

Returns an error if the database cannot be opened or the schema migration fails.

pub async fn with_migrations(url: &str) -> Result<SqliteTaskStore, Error>

Opens a SQLite database with automatic schema migration.

Runs all pending migrations before returning the store. This is the recommended constructor for production deployments because it ensures the schema is always up to date without duplicating DDL statements.

Errors

Returns an error if the database cannot be opened or any migration fails.

pub async fn from_pool(pool: Pool<Sqlite>) -> Result<SqliteTaskStore, Error>

Creates a store from an existing connection pool.

Errors

Returns an error if the schema migration fails.

pub async fn purge_expired( &self, policy: &RetentionPolicy, ) -> Result<PurgeReport, A2aError>

Deletes terminal tasks that have outlived policy.

Nothing calls this for you. A persistent store keeps every task until an operator says otherwise — see retention for why that is the default and why the in-memory store does the opposite — so this is the hook for whatever already schedules work: a cron entry, a Kubernetes CronJob, a tokio interval in your own binary.

Only Completed, Failed, Canceled and Rejected tasks are eligible. A task still Working, or parked in InputRequired waiting on a human, is never deleted however old it is.

Safe to run from several replicas at once: each batch is a single DELETE whose subquery picks the rows, so two sweeps racing delete disjoint sets rather than colliding.

Errors

Returns an error if a delete fails. A sweep that fails partway has still committed its earlier batches; the counts in the returned report are lost in that case, but the deletions are not undone and the next sweep simply continues.

Trait Implementations

impl Clone for SqliteTaskStore

fn clone(&self) -> SqliteTaskStore

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for SqliteTaskStore

fn fmt(&self, f: &mut Formatter<'_>) -> Result<(), Error>

Formats the value using the given formatter.

impl TaskStore for SqliteTaskStore

fn save_artifact_delta<'a>( &'a self, task: &'a Task, delta: ArtifactDelta, ) -> Pin<Box<dyn Future<Output = Result<(), A2aError>> + Send + 'a>>

Appends into the stored JSON document instead of rewriting it.

save serializes the whole task in Rust and ships it as a bind parameter, so a streaming agent pays for every artifact it has already persisted on every subsequent event. This sends only what changed and lets SQLite splice it into the document with json_set.

What this does and does not remove

Removed: the Rust-side serde_json::to_string of the whole task, and the transfer of the whole document as a parameter. Both scale with the stream so far.

Not removed: SQLite still parses and rewrites the row internally, so the statement remains linear in document size. A blob-per-task schema cannot avoid that; only a normalized artifacts table could, and the measurement in benches/benches/backpressure.rs says that is not where this store’s time goes — the per-event round trip dominates by roughly 3:1 at 502 events. Doing the larger surgery for the smaller term would be the wrong trade, and it is recorded here rather than left implied.

updated_at is deliberately untouched: it carries the status timestamp that orders list (§3.1.4), and appending an artifact does not change a task’s status. This matches InMemoryTaskStore, which keeps the task’s list position across an append.

Falls back to save whenever the delta cannot be applied exactly. The refused cases, and why each one is refused, are documented on the private statement builder this calls.

fn save_status_delta<'a>( &'a self, task: &'a Task, ) -> Pin<Box<dyn Future<Output = Result<(), A2aError>> + Send + 'a>>

Rewrites the stored document’s status alone.

save serializes the whole task — history included, up to 1,024 messages — and ships it as a bind parameter, so a status transition on an aged channel costs what that channel has accumulated. This sends one TaskStatus.

Three columns are what save would have left behind and so are what this has to leave behind: the document’s $.status, the state column that list filters on, and updated_at, which carries the status timestamp that §3.1.4 orders by. Missing the last one would leave the row correct and mis-ordered.

It deliberately does not touch the journal, and that is the one place it diverges from save on purpose. save deletes the journal rows because it has just rewritten data with every part in it, so they are superseded. This rewrites only the status, so the document still lacks those parts and the rows are still the record of them. Deleting them here would silently drop appended artifact parts.

Falls back to save when no row matched, which means the task is not stored yet and there is nothing to update in place.

fn supports_idempotency(&self) -> bool

Whether this store can back idempotency keys.

fn claim_idempotency_key<'a>( &'a self, key: &'a str, message_id: &'a MessageId, task_id: &'a TaskId, ) -> Pin<Box<dyn Future<Output = Result<IdempotencyClaim, A2aError>> + Send + 'a>>

Atomically claims key for task_id on behalf of message_id.

fn release_idempotency_key<'a>( &'a self, key: &'a str, ) -> Pin<Box<dyn Future<Output = Result<(), A2aError>> + Send + 'a>>

Releases a key claimed by claim_idempotency_key, so it can be claimed again.

fn save<'a>( &'a self, task: &'a Task, ) -> Pin<Box<dyn Future<Output = Result<(), A2aError>> + Send + 'a>>

Saves (creates or updates) a task.

fn get<'a>( &'a self, id: &'a TaskId, ) -> Pin<Box<dyn Future<Output = Result<Option<Task>, A2aError>> + Send + 'a>>

Retrieves a task by its ID, returning None if not found.

fn list<'a>( &'a self, params: &'a ListTasksParams, ) -> Pin<Box<dyn Future<Output = Result<TaskListResponse, A2aError>> + Send + 'a>>

Lists tasks matching the given filter parameters.

fn insert_if_absent<'a>( &'a self, task: &'a Task, ) -> Pin<Box<dyn Future<Output = Result<bool, A2aError>> + Send + 'a>>

Atomically inserts a task only if no task with the same ID exists.

fn supports_event_log(&self) -> bool

Whether this store keeps a per-task event log.

fn append_event<'a>( &'a self, task_id: &'a TaskId, seq: u64, event: &'a StreamResponse, ) -> Pin<Box<dyn Future<Output = Result<(), A2aError>> + Send + 'a>>

Appends one event at seq in the task’s log.

fn last_event_seq<'a>( &'a self, task_id: &'a TaskId, ) -> Pin<Box<dyn Future<Output = Result<u64, A2aError>> + Send + 'a>>

The highest seq this task’s log holds, or 0 when it is empty.

fn earliest_event_seq<'a>( &'a self, task_id: &'a TaskId, ) -> Pin<Box<dyn Future<Output = Result<Option<u64>, A2aError>> + Send + 'a>>

The lowest seq this task’s log still holds, or None when it holds nothing — or when the store cannot say, which is the default.

fn read_events<'a>( &'a self, task_id: &'a TaskId, after_seq: u64, limit: usize, ) -> Pin<Box<dyn Future<Output = Result<Vec<RecordedEvent>, A2aError>> + Send + 'a>>

Reads a task’s events in order, starting after after_seq.

fn delete<'a>( &'a self, id: &'a TaskId, ) -> Pin<Box<dyn Future<Output = Result<(), A2aError>> + Send + 'a>>

Deletes a task by its ID.

fn count<'a>( &'a self, ) -> Pin<Box<dyn Future<Output = Result<u64, A2aError>> + Send + 'a>>

Returns the total number of tasks in the store.

fn save_appending_history<'a>( &'a self, task: &'a Task, messages: &'a [Message], max_history: usize, ) -> Pin<Box<dyn Future<Output = Result<(), A2aError>> + Send + 'a>>

Persists a task whose history grew by messages, without the caller shipping the history it has already stored.

fn event_log_covers<'a>( &'a self, task_id: &'a TaskId, after_seq: u64, ) -> Pin<Box<dyn Future<Output = Result<bool, A2aError>> + Send + 'a>>

Whether a replay starting after after_seq can be served in full.