Trait TaskStore 

pub trait TaskStore:
    Send
    + Sync
    + 'static {
    // Required methods
    fn save<'a>(
        &'a self,
        task: &'a Task,
    ) -> Pin<Box<dyn Future<Output = Result<(), A2aError>> + Send + 'a>>;
    fn get<'a>(
        &'a self,
        id: &'a TaskId,
    ) -> Pin<Box<dyn Future<Output = Result<Option<Task>, A2aError>> + Send + 'a>>;
    fn list<'a>(
        &'a self,
        params: &'a ListTasksParams,
    ) -> Pin<Box<dyn Future<Output = Result<TaskListResponse, A2aError>> + Send + 'a>>;
    fn insert_if_absent<'a>(
        &'a self,
        task: &'a Task,
    ) -> Pin<Box<dyn Future<Output = Result<bool, A2aError>> + Send + 'a>>;
    fn delete<'a>(
        &'a self,
        id: &'a TaskId,
    ) -> Pin<Box<dyn Future<Output = Result<(), A2aError>> + Send + 'a>>;

    // Provided methods
    fn count<'a>(
        &'a self,
    ) -> Pin<Box<dyn Future<Output = Result<u64, A2aError>> + Send + 'a>> { ... }
    fn supports_idempotency(&self) -> bool { ... }
    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>> { ... }
    fn release_idempotency_key<'a>(
        &'a self,
        key: &'a str,
    ) -> Pin<Box<dyn Future<Output = Result<(), A2aError>> + Send + 'a>> { ... }
    fn save_artifact_delta<'a>(
        &'a self,
        task: &'a Task,
        delta: ArtifactDelta,
    ) -> Pin<Box<dyn Future<Output = Result<(), A2aError>> + Send + 'a>> { ... }
    fn save_status_delta<'a>(
        &'a self,
        task: &'a Task,
    ) -> Pin<Box<dyn Future<Output = Result<(), A2aError>> + Send + 'a>> { ... }
    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>> { ... }
    fn supports_event_log(&self) -> bool { ... }
    fn append_event<'a>(
        &'a self,
        task_id: &'a TaskId,
        seq: u64,
        event: &'a StreamResponse,
    ) -> Pin<Box<dyn Future<Output = Result<(), A2aError>> + Send + 'a>> { ... }
    fn last_event_seq<'a>(
        &'a self,
        task_id: &'a TaskId,
    ) -> Pin<Box<dyn Future<Output = Result<u64, A2aError>> + Send + 'a>> { ... }
    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>> { ... }
    fn earliest_event_seq<'a>(
        &'a self,
        task_id: &'a TaskId,
    ) -> Pin<Box<dyn Future<Output = Result<Option<u64>, A2aError>> + Send + 'a>> { ... }
    fn event_log_covers<'a>(
        &'a self,
        task_id: &'a TaskId,
        after_seq: u64,
    ) -> Pin<Box<dyn Future<Output = Result<bool, A2aError>> + Send + 'a>> { ... }
}

Trait for persisting and retrieving Task objects.

All methods return Pin<Box<dyn Future>> for object safety — this trait is used as Box<dyn TaskStore>.

Object safety

Do not add async fn methods; use the explicit Pin<Box<...>> form.

Example

use std::future::Future;
use std::pin::Pin;
use a2a_protocol_types::error::A2aResult;
use a2a_protocol_types::params::ListTasksParams;
use a2a_protocol_types::responses::TaskListResponse;
use a2a_protocol_types::task::{Task, TaskId};
use a2a_protocol_server::store::TaskStore;

/// A no-op store that rejects all operations (for illustration).
struct NullStore;

impl TaskStore for NullStore {
    fn save<'a>(&'a self, _task: &'a Task)
        -> Pin<Box<dyn Future<Output = A2aResult<()>> + Send + 'a>>
    {
        Box::pin(async { Ok(()) })
    }

    fn get<'a>(&'a self, _id: &'a TaskId)
        -> Pin<Box<dyn Future<Output = A2aResult<Option<Task>>> + Send + 'a>>
    {
        Box::pin(async { Ok(None) })
    }

    fn list<'a>(&'a self, _params: &'a ListTasksParams)
        -> Pin<Box<dyn Future<Output = A2aResult<TaskListResponse>> + Send + 'a>>
    {
        Box::pin(async { Ok(TaskListResponse::new(vec![])) })
    }

    fn insert_if_absent<'a>(&'a self, _task: &'a Task)
        -> Pin<Box<dyn Future<Output = A2aResult<bool>> + Send + 'a>>
    {
        Box::pin(async { Ok(true) })
    }

    fn delete<'a>(&'a self, _id: &'a TaskId)
        -> Pin<Box<dyn Future<Output = A2aResult<()>> + Send + 'a>>
    {
        Box::pin(async { Ok(()) })
    }
}

Required Methods

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

Saves (creates or updates) a task.

Terminal states are final

Every store shipped in this crate refuses a write — through this method or any of the delta methods below — that would move a stored terminal task to a different state, and does so atomically with the write. The refusal is an UnsupportedOperation error that TerminalStateConflict::from_error recognises. See crate::store::terminal for why, and for what a custom store has to do to give the same guarantee.

Errors

Returns an A2aError if the store operation fails, including the terminal-state refusal described above.

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.

Errors

Returns an A2aError if the store operation fails.

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.

Errors

Returns an A2aError if the store operation fails.

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.

Returns Ok(true) if the task was inserted, Ok(false) if a task with the same ID already exists (no modification made).

Errors

Returns an A2aError if the store operation fails.

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

Deletes a task by its ID.

Errors

Returns an A2aError if the store operation fails.

Provided Methods

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

Returns the total number of tasks in the store.

Useful for monitoring, metrics, and capacity management. Has a default implementation that returns 0 so existing implementations are not broken when this method is added.

Errors

Returns an A2aError if the store operation fails.

fn supports_idempotency(&self) -> bool

Whether this store can back idempotency keys.

Defaults to false, and that default is load-bearing: a server advertises the idempotency extension only where this is true, and refuses a send carrying a key where it is not. A store that has not implemented claim_idempotency_key therefore produces a missing advertisement and a loud refusal, never a send that quietly runs twice. Silent at-least-once is what the feature exists to remove; it must not also be what forgetting to implement it produces.

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.

Concurrent claims of one key must resolve to exactly one Claimed; every other caller observes Replay or Conflict. An implementation that is not atomic here reintroduces the double execution this prevents.

What counts as the same request

The message_id, not the message body. A genuine retry resends the identical message — the caller kept it in order to resend it — so its id is unchanged, while a distinct send carries a fresh one. This compares message identity and deliberately does not hash content: a caller reusing one MessageId for two bodies has already broken an invariant the protocol cannot check on its behalf.

Errors

The default implementation reports an unsupported operation; it is never reached unless supports_idempotency is true. Implementations return an error if the store operation fails.

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.

Why this exists

A claim is taken before the send’s remaining side effects — the queue lease, the task row, the inline push config — precisely so that two racing duplicates cannot both get past it. Any of those can still fail, and a claim left behind by a send that never created a task is worse than no claim at all: the caller’s legitimate retry would replay to a task id that never existed. Every failure path after the claim releases it, which is the same discipline the queue lease and cancellation token already follow.

Releasing a key that is not held is not an error; a failure path may run after another caller has already taken over.

Errors

The default implementation reports an unsupported operation. It is unreachable in practice: a release only follows a successful claim, which only a store with supports_idempotency can grant.

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

Persists an artifact change that has already been applied to task.

Why this exists

A streaming agent emits one artifact event per chunk, and the obvious implementation persists each one with save — which hands the store the whole task. The task grows with every chunk, so the cost of one event is proportional to the number of events before it, and the cost of a stream is quadratic in its length. Measured on the backpressure/append_volume benchmark, a 502-event stream spent 43.4 ms against the in-memory store versus 3.2 ms against a store that discards everything: 13.5× of that stream was re-persisting artifacts already persisted.

delta says exactly what changed, so a store that can update a record in place does work proportional to the change rather than to the record.

Implementing this

The default replaces the whole record via save, which is always correct — every existing implementation keeps working unchanged, and a store with no incremental update path should keep it. Overriding is worthwhile for any store where applying a delta is cheaper than rewriting the record.

All three stores shipped here override it, and what each one wins differs with its storage model:

Store	Approach	Measured on a 500-chunk stream
InMemoryTaskStore	Mutates the stored task in place	43.4 ms to 2.5 ms
SqliteTaskStore	json_set splices the tail into the document	144.5 ms to 127.6 ms
PostgresTaskStore	jsonb_set with || array concat	798 ms to 500 ms

The in-memory win is the largest because a full save there is a deep clone and a delta is a Vec extend. The SQL stores keep one JSON document per row, so they still rewrite the row internally; what the delta removes is the Rust-side serialization of the whole task and its transfer as a bind parameter. That is enough to flatten Postgres’s per-event cost — 874, 1183, 1597 µs at 50, 250 and 500 chunks with save, against 853, 840, 1000 µs with the delta — but not to make either SQL store as cheap as memory. Only normalising artifacts into their own table would do that, and the same measurements put the per-event round trip well above the document-size term, so it would buy the smaller half.

An override must leave the store holding exactly what save(task) would have left it holding. delta describes a change already present in task; if an implementation cannot apply it — the record is missing, or its shape does not match — it must fall back to save(task) rather than persist a divergent record.

“What save would have left it holding” includes the bookkeeping

A delta is a cheaper way to perform a write, not a way to perform fewer writes. Whatever per-write maintenance an implementation’s save does — advancing an eviction or compaction counter, refreshing a last-touched timestamp, updating a metric — an override has to do too, or that maintenance silently stops happening in proportion to how often the fast path is taken.

Both deltas shipped here got this wrong, and the failure is worth stating because it is invisible to the obvious test. InMemoryTaskStore paces its TTL sweep off a counter that every write advances, and neither override advanced it. The reasoning in the artifact one was explicit and wrong in an instructive way: an in-place delta adds no entry, so it cannot push the store past its capacity bound. True — and irrelevant to the other bound, because expiry is driven by elapsed time rather than by growth, so a stream that only appends to tasks already stored still ages every other task in the store. The better the fast path worked, the more rarely anything was collected. Every functional test still passed: the store held exactly the right bytes, just for ever.

Errors

Returns an A2aError if the store operation fails.

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

Persists a task whose status alone changed, without rewriting the rest of the record.

The same argument as TaskStore::save_artifact_delta, applied to the other per-event write on the hot path. A Task carries its history, which the send path grows by one message per turn up to an internal cap of 1,024 messages (handler::messaging::MAX_TASK_HISTORY_MESSAGES, which is crate-private and so not linkable from here), so a turn emitting n status events through save on a channel holding h messages does n * h work. Measured back to back on one turn of 512 events, concurrency one, in-memory store, by tests/swarm_scale::cost::a_turn_that_emits_many_events_on_an_aged_channel: with save, 1,706µs / 18,609µs / 54,301µs at h of 1 / 200 / 600; with this method, 1,044µs / 1,280µs / 2,248µs. The turn stops growing with the channel’s age, which matters more than the 24x at h = 600.

It does not measurably change a turn that emits one event — four other O(history) copies dominate that. docs/swarm-scale-findings.md has both runs and the attribution.

What an override must preserve

It must leave the store holding exactly what save(task) would have, which for an ordered store includes the position: §3.1.4 orders by status timestamp, so a status change moves the record and an in-place edit still has to re-key its indexes. An implementation that cannot apply the change — no such record — must fall back to save(task) rather than drop the transition.

Implementing this

The default replaces the whole record via save, which is always correct. All three stores shipped here override it, and what each one wins differs with its storage model. Measured on this box, 100 status events on a channel holding 200 messages, medians of five runs:

Store	Approach	save	this method
InMemoryTaskStore	Edits the stored status, re-keys the indexes	8,499µs	639µs
SqliteTaskStore	json_set on $.status, plus state/updated_at	122,071µs	37,216µs
PostgresTaskStore	jsonb_set on ARRAY['status'], plus state/updated_at	287,287µs	120,104µs

The in-memory win is the largest because a full save there is a deep clone and the delta is a field assignment. The SQL stores keep one JSON document per row and still rewrite the row internally; what the delta removes is the Rust-side serialization of the whole task — history included — and its transfer as a bind parameter.

Two columns are as much part of “what save would have left” as the document is, and an override that misses either is wrong in a way no round-trip test of get would show: the state column that list filters on, and updated_at, which carries the §3.1.4 ordering key.

SqliteTaskStore additionally must not delete its artifact journal here, though its save does. save deletes those rows because it has just rewritten the document with every part in it; a status delta has not, so the rows are still the only record of the appended parts and deleting them would lose data.

Callers must use this only when the status is genuinely the only field that moved; the background processor appends to history on an agent Message event and calls save for exactly that reason.

Errors

Returns an A2aError if the store operation fails.

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.

task.history is ignored. The appended messages come from messages and nothing else. That is deliberate and is the whole point: the caller does not have to hold, clone or transfer the conversation in order to add one turn to it, so a send stops costing what the channel has accumulated. Reusing task.history for the tail — the shape TaskStore::save_artifact_delta uses — would have put the caller back in possession of the whole thing.

Every other field of task is the new snapshot and replaces what is stored, exactly as save would.

max_history is the cap the merged history is trimmed to, oldest first. It is a parameter rather than a constant here because retention is the handler’s policy, not the store’s; a store that invented its own cap would silently disagree with the one the rest of the server enforces.

Implementing this

The default reads the record, appends, trims and writes the whole thing back through save, which is always correct and is what the send path did before this method existed — so a store that does not override it is no worse off than it was. Overriding is worthwhile for any store that can append without rewriting: the in-memory one extends a Vec in place, and the SQL stores splice the tail into the stored document.

An override must leave the store holding exactly what the default would have left it holding, including the trim, and must fall back to inserting task when no such record exists rather than dropping the turn.

Errors

Returns an A2aError if the store operation fails.

fn supports_event_log(&self) -> bool

Whether this store keeps a per-task event log.

false by default. A server reads this before offering resumption-from-offset, so a store without a log degrades to the snapshot behaviour rather than promising replay it cannot deliver.

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.

seq is a position, not a counter: the same (task_id, seq) written twice must leave one row, so replaying an append is idempotent rather than duplicating an event. That is the property sqlite_store::journal already relies on, and for the same reason — it makes a retried or overlapping write safe without a read first.

Errors

A2aError if the store fails, or unsupported_operation when the store keeps no 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.

Needed because a task outlives any one executor invocation: a task parked at input-required and then continued gets a second processor, and a seq restarting at 1 would collide with positions the first one already wrote. Since appends are idempotent by position, those collisions would be silently dropped — the continuation’s events would simply not be recorded. Resuming the numbering from here is what stops that.

Errors

A2aError if the store fails, or unsupported_operation when the store keeps no log.

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.

after_seq is exclusive, so a subscriber that has seen event n asks for n and receives n+1 onward — which is exactly the Last-Event-ID contract, and avoids the off-by-one that an inclusive offset invites at every call site.

Errors

A2aError if the store fails, or unsupported_operation when the store keeps no log.

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.

Why a store has to be able to say this

read_events filters seq > after_seq and returns whatever survives. If the head of the log is gone — truncated by TaskStoreConfig::max_events_per_task, or swept as an orphan on a SQL backend — the first event replayed is not after_seq + 1, and nothing in that answer says so. The subscriber cannot tell it apart from the numbering gaps this design calls normal (an append that failed skips a position), so it silently receives a stream with a hole in it and believes it resumed.

The default, and why it is Ok(None) rather than an error

The other event-log methods default to unsupported_operation, because a store that forgot to implement them should report no log rather than an empty one. This one cannot: it was added to a trait that out-of-tree stores already implement, and defaulting it to an error would break a working store that keeps a perfectly good log (STABILITY.md §4). It therefore defaults to “cannot say”, which is what every caller already assumed before this existed.

None is consequently not evidence that nothing is missing. Read it through event_log_covers, which spells that out in one place.

Errors

A2aError if the store fails.

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.

false means the log no longer holds after_seq + 1: the caller asked to resume from a position that has been dropped, and serving read_events would hand it a gapped stream it cannot detect. Answer such a subscriber with an error, or with a fresh snapshot — anything but a silent partial replay.

true means “no gap can be proven”, which is the honest reading: a store whose earliest_event_seq is the default answers true always, exactly as callers behaved before either method existed.

Provided rather than implemented per store, so the comparison — and the off-by-one in it, after_seq being exclusive — lives once.

Errors

A2aError if the store fails.

Dyn Compatibility

This trait is dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety".

Implementors

impl TaskStore for InMemoryTaskStore

impl TaskStore for PostgresTaskStore

impl TaskStore for SqliteTaskStore

impl TaskStore for TenantAwareInMemoryTaskStore

impl TaskStore for TenantAwarePostgresTaskStore

impl TaskStore for TenantAwareSqliteTaskStore