Trait TaskStore 

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

    // Provided methods
    fn count<'a>(
        &'a self,
    ) -> Pin<Box<dyn Future<Output = A2aResult<u64>> + 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 = A2aResult<IdempotencyClaim>> + Send + 'a>> { ... }
    fn release_idempotency_key<'a>(
        &'a self,
        key: &'a str,
    ) -> Pin<Box<dyn Future<Output = A2aResult<()>> + Send + 'a>> { ... }
    fn save_artifact_delta<'a>(
        &'a self,
        task: &'a Task,
        delta: ArtifactDelta,
    ) -> Pin<Box<dyn Future<Output = A2aResult<()>> + 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 = A2aResult<()>> + Send + 'a>> { ... }
    fn last_event_seq<'a>(
        &'a self,
        task_id: &'a TaskId,
    ) -> Pin<Box<dyn Future<Output = A2aResult<u64>> + Send + 'a>> { ... }
    fn read_events<'a>(
        &'a self,
        task_id: &'a TaskId,
        after_seq: u64,
        limit: usize,
    ) -> Pin<Box<dyn Future<Output = A2aResult<Vec<RecordedEvent>>> + Send + 'a>> { ... }
    fn earliest_event_seq<'a>(
        &'a self,
        task_id: &'a TaskId,
    ) -> Pin<Box<dyn Future<Output = A2aResult<Option<u64>>> + Send + 'a>> { ... }
    fn event_log_covers<'a>(
        &'a self,
        task_id: &'a TaskId,
        after_seq: u64,
    ) -> Pin<Box<dyn Future<Output = A2aResult<bool>> + 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 = A2aResult<()>> + Send + 'a>>

Saves (creates or updates) a task.

Errors

Returns an A2aError if the store operation fails.

fn get<'a>( &'a self, id: &'a TaskId, ) -> Pin<Box<dyn Future<Output = A2aResult<Option<Task>>> + 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 = A2aResult<TaskListResponse>> + 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 = A2aResult<bool>> + 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 = A2aResult<()>> + 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 = A2aResult<u64>> + 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 = A2aResult<IdempotencyClaim>> + 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 = A2aResult<()>> + 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 = A2aResult<()>> + 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.

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 = A2aResult<()>> + 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 = A2aResult<u64>> + 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 = A2aResult<Vec<RecordedEvent>>> + 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 = A2aResult<Option<u64>>> + 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 = A2aResult<bool>> + 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