Struct InMemoryTaskStore 

pub struct InMemoryTaskStore { /* private fields */ }

In-memory TaskStore backed by a pre-allocated HashMap with secondary indexes under a single RwLock.

Suitable for testing and single-process deployments. Data is lost when the process exits.

The internal HashMap is pre-allocated to the configured max_capacity (default 10,000) to prevent latency spikes from table resizing. Without pre-allocation, HashMap doubles its capacity when load factor exceeds ~87.5%, triggering a full rehash of every stored entry. Pre-allocation eliminates these unpredictable latency cliffs entirely.

Indexing strategy

Index	Structure	Purpose
Primary	HashMap<TaskId, TaskEntry>	O(1) get/save
Order	BTreeMap<u64, TaskId>	O(log n + page_size) update-order pagination
Context	HashMap<String, BTreeMap<u64, TaskId>>	O(log m + page_size) filtered list

The order index is keyed by a monotonic per-write sequence and iterated in reverse to return most-recently-updated tasks first (spec §3.1.4) without the O(n log n) per-call sort that previously caused 20-70× regressions at 10K+ tasks. The context index avoids full-scan filtering by pre-partitioning task IDs by context, preserving the same update-order within each context.

Eviction behavior

Eviction runs every N writes (configurable via TaskStoreConfig::eviction_interval) and whenever the store exceeds max_capacity. If the system goes idle (no save() calls), completed tasks persist in memory past their TTL.

Operators should call run_eviction() periodically (e.g. every 60 seconds via tokio::time::interval) to ensure timely cleanup of terminal tasks during idle periods.

The write that triggers a sweep pays for it

This paragraph used to say the sweep “runs as a background task” and that “writers are not blocked during the O(n) cleanup”. Neither is true and both were corrected on 2026-08-19. There is no spawn: the sweep is awaited inside save, so the caller that triggers it waits for it, and it holds the write lock for its whole duration, so every other writer waits too. What is genuinely decoupled is only the lock acquisition — the insert’s write lock is released before the sweep takes its own.

The size of that, MEASURED (debug profile, tokio multi-thread, 50,000 terminal tasks, eviction_interval 1000):

	latency
quietest of 1,000 consecutive saves	3.99 µs
slowest of the same 1,000 (the one that swept)	4.54 ms

One write in eviction_interval costs about 1,100× a quiet one at that size, and it stalls every concurrent writer, not just itself. The capacity pass is much cheaper — measured 3.5× and 3.7× on two runs at 10,000 entries — because it removes only the overflow rather than scanning for it.

This is a shape to plan for, not a defect to route around: the TTL pass is O(n) unavoidably (finding expired entries means looking at all of them), which is exactly why it is amortized behind eviction_interval rather than run per write. What it means in practice is that eviction_interval is a tail-latency knob as much as a cleanliness one, and that a deployment sensitive to p99.9 write latency should raise it and call run_eviction() from its own scheduler instead — where the stall lands somewhere it chose.

Concurrency

For high-concurrency production deployments, consider SqliteTaskStore which uses a connection pool and row-level locking. The in-memory store uses a single RwLock and is optimized for testing and moderate load.

Implementations

impl InMemoryTaskStore

pub async fn run_eviction(&self)

Runs background eviction of expired and over-capacity entries.

Call this periodically (e.g. every 60 seconds) to clean up terminal tasks that would otherwise persist until the next save() call.

impl InMemoryTaskStore

pub fn new() -> InMemoryTaskStore

Creates a new empty in-memory task store with default configuration.

Default: max 10,000 tasks, 1-hour TTL for terminal tasks. The internal HashMap is pre-allocated to the configured max_capacity to prevent resize-induced latency spikes during operation.

pub fn with_config(config: TaskStoreConfig) -> InMemoryTaskStore

Creates a new in-memory task store with custom configuration.

The internal HashMap is pre-allocated to config.max_capacity (or a sensible default if None) to prevent resize-induced latency spikes.

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

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

Defaults to NoopMetrics. Two appends leave the log short of an event the agent emitted and still return Ok(()): one for a task this store no longer holds, and one landing on a position another writer already took. Neither can be an error — see event_append_error — so this is how they become visible.

pub async fn idempotency_key_count(&self) -> usize

How many idempotency keys this store currently holds.

Not covered by count, which counts tasks. The two diverge on purpose: a key deliberately outlives the task it names, because dropping it would let a delayed retry of the same send execute a second time — see StoreData::idempotency_index. Anything deciding that a store is finished with therefore has to ask about both.

pub async fn is_prunable(&self) -> bool

Whether this store can be discarded without losing anything.

true only when it holds no tasks and no idempotency keys. A caller that reclaims empty stores — a tenant partition map, say — must ask this rather than count() == 0: an emptied store can still be holding the keys that stop a retry executing twice, and discarding it reopens exactly the double execution the keys exist to prevent.

Trait Implementations

impl Debug for InMemoryTaskStore

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

Formats the value using the given formatter.

impl Default for InMemoryTaskStore

fn default() -> InMemoryTaskStore

Returns the “default value” for a type.

impl TaskStore for InMemoryTaskStore

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

Edits the stored status in place rather than replacing the record.

save here is a deep clone of the whole task, history included, so a status transition on a long conversation costs what that conversation has accumulated. This costs a TaskStatus and two index re-keys.

Falls back to save when the record is absent, which is the one case the in-place update cannot apply — the same discipline TaskStore::save_artifact_delta follows, and for the same reason: a dropped transition would be worse than a slow one.

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

Applies the delta to the stored task in place, copying only what grew.

save clones the whole task, so using it per artifact event makes a stream cost quadratic in its own length — see TaskStore::save_artifact_delta for the measurement. Here the work is proportional to the appended parts instead of the accumulated ones.

Falls back to save whenever the stored record is not the one this delta describes: absent, no artifacts, index out of range, a different artifact at that index, or fewer parts present than the delta claims were appended. Those are all “cannot apply safely”, and a whole-record replace is always right — a wrong in-place edit would not be.

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>>

Appends the new messages to the stored history in place.

save here is a deep clone of the whole task, so a send on an aged channel costs what that channel has accumulated. This extends a Vec by the messages the turn actually added.

Falls back to save when the record is absent, which is the one case an in-place append cannot serve — and task carries the full initial history in that case, because a first turn has nothing to append to.

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 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 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 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 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 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 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 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.