Package org.a2aproject.sdk.server.tasks

Class InMemoryTaskStore

java.lang.Object
org.a2aproject.sdk.server.tasks.InMemoryTaskStore
All Implemented Interfaces:
TaskStateProvider, TaskStore
@ApplicationScoped public class InMemoryTaskStore extends Object implements TaskStore, TaskStateProvider
In-memory implementation of TaskStore and TaskStateProvider.

This implementation uses a ConcurrentHashMap to store tasks in memory. Tasks are lost on application restart. For persistent storage, use a database-backed implementation such as the JPA TaskStore in the extras module.

This is the default TaskStore used when no other implementation is provided.

Exception Behavior

InMemoryTaskStore has minimal exception scenarios compared to database-backed implementations:
  • No TaskSerializationException: Task objects are stored directly in memory without serialization. No JSON parsing or schema compatibility issues can occur.
  • No TaskPersistenceException: ConcurrentHashMap operations do not involve I/O, network, or transactional concerns. Standard put/get/remove operations are guaranteed to succeed under normal JVM operation.
  • OutOfMemoryError (potential): The only failure scenario is JVM heap exhaustion if too many tasks are stored. This is an Error (not Exception) and indicates a fatal system condition requiring JVM restart and capacity planning.

Design Rationale

This implementation intentionally does NOT throw TaskStoreException or its subclasses because:
  • No serialization step exists - tasks stored as Java objects
  • No I/O or network operations that can fail
  • ConcurrentHashMap guarantees thread-safe operations without checked exceptions
  • Memory exhaustion (OutOfMemoryError) is an unrecoverable system failure

Comparison to Database Implementations

Database-backed implementations (e.g., JpaDatabaseTaskStore) throw exceptions for:
  • Serialization errors (JSON parsing, schema mismatches)
  • Connection failures (network, timeouts)
  • Transaction failures (deadlocks, constraint violations)
  • Capacity issues (disk full, quota exceeded)
InMemoryTaskStore avoids all of these by operating entirely in-process.

Memory Management Considerations

Callers should monitor memory usage and implement task cleanup policies: 

 // Example: Delete finalized tasks older than 48 hours
 ListTasksParams params = new ListTasksParams.Builder()
     .statusTimestampBefore(Instant.now().minus(Duration.ofHours(48)))
     .build();

 List<Task> oldTasks = taskStore.list(params, context).tasks();
 oldTasks.stream()
     .filter(task -> task.status().state().isFinal())
     .forEach(task -> taskStore.delete(task.id()));

Thread Safety

All operations are thread-safe via ConcurrentHashMap. Multiple threads can concurrently save, get, list, and delete tasks without synchronization. Last-write-wins semantics apply for concurrent save() calls to the same task ID.
See Also:

  • Constructor Details

    • InMemoryTaskStore

      public InMemoryTaskStore()

    • InMemoryTaskStore

      @Inject public InMemoryTaskStore(@Any jakarta.enterprise.inject.Instance<TaskAuthorizationProvider> authorizationProviderInstance)

    • InMemoryTaskStore

      public InMemoryTaskStore(@Nullable TaskAuthorizationProvider authorizationProvider)
      Creates an in-memory task store with optional task-level authorization.

      This constructor supports programmatic wiring in non-CDI runtimes such as Spring Framework.

      Parameters:
      authorizationProvider - provider used to filter tasks during list operations; null permits all tasks

  • Method Details

    • save

      public void save(org.a2aproject.sdk.spec.Task task, boolean isReplicated)
      Description copied from interface: TaskStore
      Saves or updates a task.
      Specified by:
      save in interface TaskStore
      Parameters:
      task - the task to save
      isReplicated - true if this task update came from a replicated event, false if it originated locally. Used to prevent feedback loops in replicated scenarios (e.g., don't fire TaskFinalizedEvent for replicated updates)

    • get

      public @Nullable org.a2aproject.sdk.spec.Task get(String taskId)
      Description copied from interface: TaskStore
      Retrieves a task by its ID.
      Specified by:
      get in interface TaskStore
      Parameters:
      taskId - the task identifier
      Returns:
      the task if found, null otherwise

    • delete

      public void delete(String taskId)
      Description copied from interface: TaskStore
      Deletes a task by its ID.
      Specified by:
      delete in interface TaskStore
      Parameters:
      taskId - the task identifier

    • list

      public org.a2aproject.sdk.jsonrpc.common.wrappers.ListTasksResult list(org.a2aproject.sdk.spec.ListTasksParams params, @Nullable ServerCallContext context)
      Description copied from interface: TaskStore
      List tasks with optional filtering and pagination.

      Authorization filtering: When a TaskAuthorizationProvider bean is present, implementations must call checkRead for each candidate task and exclude tasks for which the check returns false. The filtering should be applied before pagination so that page sizes are correct from the caller's perspective. If no provider is present, all tasks are returned.

      ⚠ Custom implementation warning: Returning unfiltered results bypasses the authorization model and can leak tasks belonging to other users. Custom implementations must apply per-task checkRead filtering before pagination. The TaskAuthorizationProvider should be declared as a CDI dependency and injected via the constructor or @Inject.

      Specified by:
      list in interface TaskStore
      Parameters:
      params - the filtering and pagination parameters
      context - the server call context (used for authorization filtering)
      Returns:
      the list of tasks matching the criteria with pagination info

    • isTaskActive

      public boolean isTaskActive(String taskId)
      Description copied from interface: TaskStateProvider
      Determines whether a task is considered active for queue management purposes.

      This method includes the grace period in its check. A task is considered active if:

      • Its state is not final, OR
      • Its state is final but finalized within the grace period (now < finalizedAt + gracePeriod)

      This method is used to decide whether to process late-arriving events.

      Specified by:
      isTaskActive in interface TaskStateProvider
      Parameters:
      taskId - the ID of the task to check
      Returns:
      true if the task is active (or recently finalized within grace period), false otherwise

    • isTaskFinalized

      public boolean isTaskFinalized(String taskId)
      Description copied from interface: TaskStateProvider
      Determines whether a task is in a final state, ignoring the grace period.

      This method performs an immediate check: returns true only if the task is in a final state (COMPLETED, CANCELED, FAILED, etc.), regardless of when it was finalized.

      This method is used by cleanup callbacks and MainQueue closing logic to decide whether a queue can be closed and removed. By ignoring the grace period, it ensures responsive cleanup while late in-flight events are still handled by the grace period mechanism for isTaskActive.

      Specified by:
      isTaskFinalized in interface TaskStateProvider
      Parameters:
      taskId - the ID of the task to check
      Returns:
      true if the task is in a final state (ignoring grace period), false otherwise