Documentation
¶
Overview ¶
Package taskstore provides a durable, tenant-aware taskstore.Store implementation for the A2A task lifecycle. The a2a-go server stack keeps tasks in an in-memory store by default (lost on restart, no per-tenant filter); this package backs them on the SAME PostgreSQL database as the conversation history (CONVERSATION_STORE_URL) so tasks survive restarts and carry the owning Milo project on every row.
The exported store implements the a2a-go github.com/a2aproject/a2a-go/v2/a2asrv/taskstore.Store interface (imported here as a2astore) — Create/Update/Get/List with the StoredTask / UpdateRequest / TaskVersion optimistic-concurrency contract. It ADDS a project-scoped listing primitive (PostgresStore.ListForProjects) that the interface's project-blind List cannot express, so a future tenant-scoped list endpoint has a tenant-safe query to build on.
Index ¶
- func WithProjectScope(ctx context.Context, projects []string, all bool) context.Context
- type PostgresStore
- func (s *PostgresStore) Close()
- func (s *PostgresStore) Create(ctx context.Context, task *a2a.Task) (a2astore.TaskVersion, error)
- func (s *PostgresStore) Get(ctx context.Context, taskID a2a.TaskID) (*a2astore.StoredTask, error)
- func (s *PostgresStore) List(ctx context.Context, req *a2a.ListTasksRequest) (*a2a.ListTasksResponse, error)
- func (s *PostgresStore) ListForProjects(ctx context.Context, projects []string, req *a2a.ListTasksRequest) (*a2a.ListTasksResponse, error)
- func (s *PostgresStore) Ping(ctx context.Context) error
- func (s *PostgresStore) Update(ctx context.Context, req *a2astore.UpdateRequest) (a2astore.TaskVersion, error)
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func WithProjectScope ¶
WithProjectScope returns a context that scopes an interface-level List to the given projects. It exists so a future scoped ListTasks endpoint can pass the caller's granted projects through a2a-go's project-blind List signature. Passing all=true lists every project's tasks and is intended for operational tooling only, never a tenant-facing path.
Types ¶
type PostgresStore ¶
type PostgresStore struct {
// contains filtered or unexported fields
}
PostgresStore is a durable, tenant-aware a2astore.Store on PostgreSQL. Safe for concurrent use; updates to the same task serialize on its row via SELECT ... FOR UPDATE. Construct with NewPostgresStore, release with Close.
func NewPostgresStore ¶
func NewPostgresStore(ctx context.Context, databaseURL string, logger *slog.Logger) (*PostgresStore, error)
NewPostgresStore connects to databaseURL (a postgres:// URL — the same CONVERSATION_STORE_URL the history store uses), verifies the connection, and applies the schema. It fails fast on an unreachable database: a service configured for durable tasks must not silently fall back to amnesia.
func (*PostgresStore) Create ¶
func (s *PostgresStore) Create(ctx context.Context, task *a2a.Task) (a2astore.TaskVersion, error)
Create implements a2astore.Store: it inserts a new task at version 1, denormalizing the owning project and context onto their columns for scoped listing. Returns a2astore.ErrTaskAlreadyExists when the ID already exists.
func (*PostgresStore) Get ¶
func (s *PostgresStore) Get(ctx context.Context, taskID a2a.TaskID) (*a2astore.StoredTask, error)
Get implements a2astore.Store: it returns the stored task and its version, or a2a.ErrTaskNotFound.
func (*PostgresStore) List ¶
func (s *PostgresStore) List(ctx context.Context, req *a2a.ListTasksRequest) (*a2a.ListTasksResponse, error)
List implements a2astore.Store. The A2A ListTasks RPC carries no caller identity in its request, so this method CANNOT scope the result to the caller on its own — the server middleware therefore denies the ListTasks RPC outright (internal/server/middleware.go). To keep the store tenant-safe even if it is ever reached, List returns only tasks whose project is in the scope carried on the context (WithProjectScope); with no scope on the context it returns an empty page rather than leaking cross-tenant tasks. The tenant-safe primitive a future scoped endpoint should call is PostgresStore.ListForProjects.
func (*PostgresStore) ListForProjects ¶
func (s *PostgresStore) ListForProjects(ctx context.Context, projects []string, req *a2a.ListTasksRequest) (*a2a.ListTasksResponse, error)
ListForProjects lists tasks owned by any of the granted projects, newest activity first, with the same paging/filter semantics as [List]. It is the tenant-safe listing primitive: pass the caller's granted project set. An empty projects slice returns an empty page (no grants ⇒ no tasks).
func (*PostgresStore) Ping ¶
func (s *PostgresStore) Ping(ctx context.Context) error
Ping verifies the database is reachable. It backs the server's readiness probe (GET /readyz) so traffic is withheld until durable storage is available.
func (*PostgresStore) Update ¶
func (s *PostgresStore) Update(ctx context.Context, req *a2astore.UpdateRequest) (a2astore.TaskVersion, error)
Update implements a2astore.Store: it replaces the stored task under optimistic concurrency. The current row is locked (SELECT ... FOR UPDATE) so concurrent updates serialize; a non-missing PrevVersion that no longer matches the stored version is rejected with a2astore.ErrConcurrentModification, and an absent task with a2a.ErrTaskNotFound.
Source Files